introduce method of flushing by topic (can be extended to flush by topic(s) instead).

victokoh · victokoh · commit 9c9d62fb1fcb · 2025-05-20T09:53:41.000+01:00
diff --git a/transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/DefaultDialect.java b/transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/DefaultDialect.java
@@ -28,6 +28,7 @@ static Builder builder(String name) {
   @Getter private final String lock;
   @Getter private final String checkSql;
   @Getter private final String fetchNextInAllTopics;
+  @Getter private final String fetchNextInSelectedTopics;
   @Getter private final String fetchCurrentVersion;
   @Getter private final String fetchNextSequence;
   private final Collection<Migration> migrations;
@@ -78,6 +79,12 @@ static final class Builder {
             + " AND seq = ("
             + "SELECT MIN(seq) FROM {{table}} b WHERE b.topic=a.topic AND b.processed = false"
             + ") LIMIT {{batchSize}}";
+    private String fetchNextInSelectedTopics =
+        "SELECT {{allFields}} FROM {{table}} a"
+            + " WHERE processed = false AND topic IN ({{topicNames}}) AND nextAttemptTime < ?"
+            + " AND seq = ("
+            + "SELECT MIN(seq) FROM {{table}} b WHERE b.topic=a.topic AND b.processed = false"
+            + ") LIMIT {{batchSize}}";
     private String fetchCurrentVersion = "SELECT version FROM TXNO_VERSION FOR UPDATE";
     private String fetchNextSequence = "SELECT seq FROM TXNO_SEQUENCE WHERE topic = ? FOR UPDATE";
 
@@ -183,6 +190,7 @@ Dialect build() {
           lock,
           checkSql,
           fetchNextInAllTopics,
+          fetchNextInSelectedTopics,
           fetchCurrentVersion,
           fetchNextSequence,
           migrations.values()) {
diff --git a/transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/DefaultPersistor.java b/transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/DefaultPersistor.java
@@ -339,6 +339,30 @@ public Collection<TransactionOutboxEntry> selectNextInTopics(
     }
   }
 
+  @Override
+  public Collection<TransactionOutboxEntry> selectNextInSelectedTopics(
+      Transaction tx, List<String> topicNames, int batchSize, Instant now) throws Exception {
+    var sql =
+        dialect
+            .getFetchNextInSelectedTopics()
+            .replace("{{table}}", tableName)
+            .replace(
+                "{{topicNames}}",
+                topicNames.stream()
+                    .map(it -> "'%s'".replace("%s", it))
+                    .reduce((a, b) -> a.concat(",").concat(b))
+                    .orElse("''"))
+            .replace("{{batchSize}}", Integer.toString(batchSize))
+            .replace("{{allFields}}", ALL_FIELDS);
+    //noinspection resource
+    try (PreparedStatement stmt = tx.connection().prepareStatement(sql)) {
+      stmt.setTimestamp(1, Timestamp.from(now));
+      var results = new ArrayList<TransactionOutboxEntry>();
+      gatherResults(stmt, results);
+      return results;
+    }
+  }
+
   @Override
   public int deleteProcessedAndExpired(Transaction tx, int batchSize, Instant now)
       throws Exception {
diff --git a/transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/Dialect.java b/transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/Dialect.java
@@ -22,6 +22,8 @@ public interface Dialect {
 
   String getFetchNextInAllTopics();
 
+  String getFetchNextInSelectedTopics();
+
   String getFetchCurrentVersion();
 
   String getFetchNextSequence();
@@ -44,6 +46,10 @@ public interface Dialect {
               "WITH raw AS(SELECT {{allFields}}, (ROW_NUMBER() OVER(PARTITION BY topic ORDER BY seq)) as rn"
                   + " FROM {{table}} WHERE processed = false AND topic <> '*')"
                   + " SELECT * FROM raw WHERE rn = 1 AND nextAttemptTime < ? LIMIT {{batchSize}}")
+          .fetchNextInSelectedTopics(
+              "WITH raw AS(SELECT {{allFields}}, (ROW_NUMBER() OVER(PARTITION BY topic ORDER BY seq)) as rn"
+                  + " FROM {{table}} WHERE processed = false AND topic IN ({{topicNames}}))"
+                  + " SELECT * FROM raw WHERE rn = 1 AND nextAttemptTime < ? LIMIT {{batchSize}}")
           .deleteExpired(
               "DELETE FROM {{table}} WHERE nextAttemptTime < ? AND processed = true AND blocked = false"
                   + " LIMIT {{batchSize}}")
@@ -64,6 +70,10 @@ public interface Dialect {
               "WITH raw AS(SELECT {{allFields}}, (ROW_NUMBER() OVER(PARTITION BY topic ORDER BY seq)) as rn"
                   + " FROM {{table}} WHERE processed = false AND topic <> '*')"
                   + " SELECT * FROM raw WHERE rn = 1 AND nextAttemptTime < ? LIMIT {{batchSize}}")
+          .fetchNextInSelectedTopics(
+              "WITH raw AS(SELECT {{allFields}}, (ROW_NUMBER() OVER(PARTITION BY topic ORDER BY seq)) as rn"
+                  + " FROM {{table}} WHERE processed = false AND topic IN ({{topicNames}}))"
+                  + " SELECT * FROM raw WHERE rn = 1 AND nextAttemptTime < ? LIMIT {{batchSize}}")
           .deleteExpired(
               "DELETE FROM {{table}} WHERE id IN "
                   + "(SELECT id FROM {{table}} WHERE nextAttemptTime < ? AND processed = true AND blocked = false LIMIT {{batchSize}})")
@@ -93,6 +103,10 @@ public interface Dialect {
               "WITH cte1 AS (SELECT {{allFields}}, (ROW_NUMBER() OVER(PARTITION BY topic ORDER BY seq)) as rn"
                   + " FROM {{table}} WHERE processed = 0 AND topic <> '*')"
                   + " SELECT * FROM cte1 WHERE rn = 1 AND nextAttemptTime < ? AND ROWNUM <= {{batchSize}}")
+          .fetchNextInSelectedTopics(
+              "WITH cte1 AS (SELECT {{allFields}}, (ROW_NUMBER() OVER(PARTITION BY topic ORDER BY seq)) as rn"
+                  + " FROM {{table}} WHERE processed = 0 AND topic IN ({{topicNames}}))"
+                  + " SELECT * FROM cte1 WHERE rn = 1 AND nextAttemptTime < ? AND ROWNUM <= {{batchSize}}")
           .deleteExpired(
               "DELETE FROM {{table}} WHERE nextAttemptTime < ? AND processed = 1 AND blocked = 0 "
                   + "AND ROWNUM <= {{batchSize}}")
@@ -161,6 +175,12 @@ public interface Dialect {
                   + " AND seq = ("
                   + "SELECT MIN(seq) FROM {{table}} b WHERE b.topic=a.topic AND b.processed = 0"
                   + ")")
+          .fetchNextInSelectedTopics(
+              "SELECT TOP {{batchSize}} {{allFields}} FROM {{table}} a"
+                  + " WHERE processed = 0 AND topic IN ({{topicNames}}) AND nextAttemptTime < ?"
+                  + " AND seq = ("
+                  + "SELECT MIN(seq) FROM {{table}} b WHERE b.topic=a.topic AND b.processed = 0"
+                  + ")")
           .fetchNextSequence(
               "SELECT seq FROM TXNO_SEQUENCE WITH (UPDLOCK, ROWLOCK, READPAST) WHERE topic = ?")
           .booleanValueFrom(v -> v ? "1" : "0")
diff --git a/transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/Persistor.java b/transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/Persistor.java
@@ -118,6 +118,19 @@ List<TransactionOutboxEntry> selectBatch(Transaction tx, int batchSize, Instant
   Collection<TransactionOutboxEntry> selectNextInTopics(Transaction tx, int batchSize, Instant now)
       throws Exception;
 
+  /**
+   * Selects the next items in all selected topics as a batch for processing. Does not lock.
+   *
+   * @param tx The current {@link Transaction}.
+   * @param topicNames The topics to select records from.
+   * @param batchSize The maximum number of records to select.
+   * @param now The time to use when selecting records.
+   * @return The records.
+   * @throws Exception Any exception.
+   */
+  Collection<TransactionOutboxEntry> selectNextInSelectedTopics(
+      Transaction tx, List<String> topicNames, int batchSize, Instant now) throws Exception;
+
   /**
    * Deletes records which have processed and passed their expiry time, in specified batch sizes.
    *
diff --git a/transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/StubPersistor.java b/transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/StubPersistor.java
@@ -52,6 +52,12 @@ public Collection<TransactionOutboxEntry> selectNextInTopics(
     return List.of();
   }
 
+  @Override
+  public Collection<TransactionOutboxEntry> selectNextInSelectedTopics(
+      Transaction tx, List<String> topicNames, int batchSize, Instant now) throws Exception {
+    return List.of();
+  }
+
   @Override
   public int deleteProcessedAndExpired(Transaction tx, int batchSize, Instant now) {
     return 0;
diff --git a/transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/TransactionOutbox.java b/transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/TransactionOutbox.java
@@ -102,6 +102,15 @@ default boolean flush() {
    */
   boolean flush(Executor executor);
 
+  /**
+   * Flushes a specific topic
+   *
+   * @param executor
+   * @param topicName
+   * @return
+   */
+  boolean flushTopic(Executor executor, String topicName);
+
   /**
    * Unblocks a blocked entry and resets the attempt count so that it will be retried again.
    * Requires an active transaction and a transaction manager that supports thread local context.
diff --git a/transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/TransactionOutboxImpl.java b/transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/TransactionOutboxImpl.java
@@ -149,6 +149,36 @@ public boolean flush(Executor executor) {
         .orElse(false);
   }
 
+  @Override
+  public boolean flushTopic(Executor executor, String topicName) {
+    if (!initialized.get()) {
+      throw new IllegalStateException("Not initialized");
+    }
+    Instant now = clockProvider.get().instant();
+    List<CompletableFuture<Boolean>> futures = new ArrayList<>();
+    futures.add(
+        CompletableFuture.runAsync(() -> expireIdempotencyProtection(now), executor)
+            .thenApply(it -> false));
+
+    futures.add(
+        CompletableFuture.supplyAsync(
+            () -> {
+              log.debug("Flushing selected topic {}", topicName);
+              return doFlush(
+                  tx ->
+                      uncheckedly(
+                          () ->
+                              persistor.selectNextInSelectedTopics(
+                                  tx, List.of(topicName), flushBatchSize, now)));
+            },
+            executor));
+
+    return futures.stream()
+        .reduce((f1, f2) -> f1.thenCombine(f2, (d1, d2) -> d1 || d2))
+        .map(CompletableFuture::join)
+        .orElse(false);
+  }
+
   private void expireIdempotencyProtection(Instant now) {
     long totalRecordsDeleted = 0;
     int recordsDeleted;
diff --git a/transactionoutbox-testing/src/main/java/com/gruelbox/transactionoutbox/testing/AbstractAcceptanceTest.java b/transactionoutbox-testing/src/main/java/com/gruelbox/transactionoutbox/testing/AbstractAcceptanceTest.java
@@ -501,6 +501,63 @@ final void retryBehaviour() throws Exception {
         singleThreadPool);
   }
 
+  @Test
+  final void flushOnlyASpecifiedTopic() throws Exception {
+    TransactionManager transactionManager = txManager();
+    CountDownLatch successLatch = new CountDownLatch(1);
+    AtomicInteger attempts = new AtomicInteger(-1);
+    var processedEntryListener = new ProcessedEntryListener(successLatch);
+    TransactionOutbox outbox =
+        TransactionOutbox.builder()
+            .transactionManager(transactionManager)
+            .persistor(Persistor.forDialect(connectionDetails().dialect()))
+            .instantiator(new FailingInstantiator(attempts))
+            .submitter(Submitter.withExecutor(singleThreadPool))
+            .attemptFrequency(Duration.ofMillis(500))
+            .listener(processedEntryListener)
+            .blockAfterAttempts(5)
+            .build();
+
+    clearOutbox();
+
+    var selectedTopic = "SELECTED_TOPIC";
+    withRunningFlusher(
+        outbox,
+        () -> {
+          transactionManager.inTransaction(
+              () -> {
+                outbox.with().schedule(InterfaceProcessor.class).process(2, "Whee");
+                outbox
+                    .with()
+                    .ordered(selectedTopic)
+                    .schedule(InterfaceProcessor.class)
+                    .process(3, "Whoo");
+                outbox
+                    .with()
+                    .ordered("IGNORED_TOPIC")
+                    .schedule(InterfaceProcessor.class)
+                    .process(2, "Wheeeee");
+              });
+
+          assertTrue(successLatch.await(20, SECONDS), "Timeout waiting for success");
+          var successes = processedEntryListener.getSuccessfulEntries();
+          var failures = processedEntryListener.getFailingEntries();
+
+          // then we only expect the selected topic we're flushing to have had eventually succeeded
+          // as the other work would not have been picked up for a retry
+          assertEquals(1, successes.stream().map(TransactionOutboxEntry::getId).distinct().count());
+
+          // a single failure is expected; from the unordered submission (as these run automatically
+          assertEquals(1, failures.stream().filter(it -> it.getTopic() == null).count());
+
+          // all other failures accounted for belong to the selected topic
+          assertEquals(
+              2, failures.stream().filter(it -> selectedTopic.equals(it.getTopic())).count());
+        },
+        singleThreadPool,
+        selectedTopic);
+  }
+
   @Test
   final void onSchedulingFailure_BubbleExceptionsUp() throws Exception {
     Assumptions.assumeTrue(
diff --git a/transactionoutbox-testing/src/main/java/com/gruelbox/transactionoutbox/testing/BaseTest.java b/transactionoutbox-testing/src/main/java/com/gruelbox/transactionoutbox/testing/BaseTest.java
@@ -79,14 +79,22 @@ protected void withRunningFlusher(TransactionOutbox outbox, ThrowingRunnable run
 
   protected void withRunningFlusher(
       TransactionOutbox outbox, ThrowingRunnable runnable, Executor executor) throws Exception {
+    withRunningFlusher(outbox, runnable, executor, null);
+  }
+
+  protected void withRunningFlusher(
+      TransactionOutbox outbox, ThrowingRunnable runnable, Executor executor, String topicName)
+      throws Exception {
     Thread backgroundThread =
         new Thread(
             () -> {
               while (!Thread.interrupted()) {
                 try {
                   // Keep flushing work until there's nothing left to flush
                   log.info("Starting flush...");
-                  while (outbox.flush(executor)) {
+                  while (topicName == null
+                      ? outbox.flush(executor)
+                      : outbox.flushTopic(executor, topicName)) {
                     log.info("More work to do...");
                   }
                   log.info("Done!");
diff --git a/transactionoutbox-testing/src/main/java/com/gruelbox/transactionoutbox/testing/ProcessedEntryListener.java b/transactionoutbox-testing/src/main/java/com/gruelbox/transactionoutbox/testing/ProcessedEntryListener.java
@@ -0,0 +1,62 @@
+package com.gruelbox.transactionoutbox.testing;
+
+import com.gruelbox.transactionoutbox.TransactionOutboxEntry;
+import com.gruelbox.transactionoutbox.TransactionOutboxListener;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Collects an ordered list of tx outbox entries that have been 'processed' i.e. succeeded or failed
+ * in processing.
+ */
+@Slf4j
+public final class ProcessedEntryListener implements TransactionOutboxListener {
+  private final CountDownLatch successLatch;
+
+  private final CopyOnWriteArrayList<TransactionOutboxEntry> successfulEntries =
+      new CopyOnWriteArrayList<>();
+  private final CopyOnWriteArrayList<TransactionOutboxEntry> failingEntries =
+      new CopyOnWriteArrayList<>();
+
+  public ProcessedEntryListener(CountDownLatch successLatch) {
+    this.successLatch = successLatch;
+  }
+
+  @Override
+  public void success(TransactionOutboxEntry entry) {
+    var copy = from(entry);
+    successfulEntries.add(copy);
+    log.info(
+        "Received success #{}. Counting down at {}",
+        successfulEntries.size(),
+        successLatch.getCount());
+    successLatch.countDown();
+  }
+
+  @Override
+  public void failure(TransactionOutboxEntry entry, Throwable cause) {
+    failingEntries.add(from(entry));
+  }
+
+  /**
+   * Retrieve an unmodifiable copy of {@link #successfulEntries}. Beware, expectation is that this
+   * does not/ should not get accessed until the correct number of {@link
+   * #success(TransactionOutboxEntry)} and {@link #blocked(TransactionOutboxEntry, Throwable)}}
+   * counts have occurred.
+   *
+   * @return unmodifiable list of ordered outbox entry events.
+   */
+  public List<TransactionOutboxEntry> getSuccessfulEntries() {
+    return List.copyOf(successfulEntries);
+  }
+
+  public List<TransactionOutboxEntry> getFailingEntries() {
+    return List.copyOf(failingEntries);
+  }
+
+  private TransactionOutboxEntry from(TransactionOutboxEntry entry) {
+    return entry.toBuilder().build();
+  }
+}
