+1 modifications to flushBy(topicName) functionality given PR feedback.

Author: victokoh

transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/DefaultPersistor.java

@@ -15,6 +15,7 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
+import java.util.stream.Collectors;
 import lombok.AccessLevel;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
@@ -342,21 +343,23 @@ public Collection<TransactionOutboxEntry> selectNextInTopics(
   @Override
   public Collection<TransactionOutboxEntry> selectNextInSelectedTopics(
       Transaction tx, List<String> topicNames, int batchSize, Instant now) throws Exception {
+
+    var topicsInParameterList = topicNames.stream().map(it -> "?").collect(Collectors.joining(","));
     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("{{topicNames}}", topicsInParameterList)
             .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 counter = 1;
+      for (var topicName : topicNames) {
+        stmt.setString(counter, topicName);
+        counter++;
+      }
+      stmt.setTimestamp(counter, Timestamp.from(now));
       var results = new ArrayList<TransactionOutboxEntry>();
       gatherResults(stmt, results);
       return results;

transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/TransactionOutbox.java

@@ -2,6 +2,8 @@
 
 import java.time.Clock;
 import java.time.Duration;
+import java.util.Arrays;
+import java.util.List;
 import java.util.concurrent.Executor;
 import java.util.function.Supplier;
 import lombok.ToString;
@@ -103,13 +105,26 @@ default boolean flush() {
   boolean flush(Executor executor);
 
   /**
-   * Flushes a specific topic
+   * Flushes a specific topic (or set of topics)
    *
-   * @param executor
-   * @param topicName
-   * @return
+   * @param executor to be used for parallelising work (note that the method overall is blocking and
+   *     this is solely ued for fork-join semantics).
+   * @param topicNames the list of specific topics to flush
+   * @return true if any work was flushed
+   */
+  default boolean flushTopics(Executor executor, String... topicNames) {
+    return flushTopics(executor, Arrays.asList(topicNames));
+  }
+
+  /**
+   * Flushes a specific topic (or set of topics)
+   *
+   * @param executor to be used for parallelising work (note that the method overall is blocking and
+   *     this is solely ued for fork-join semantics).
+   * @param topicNames the list of specific topics to flush
+   * @return true if any work was flushed
    */
-  boolean flushTopic(Executor executor, String topicName);
+  boolean flushTopics(Executor executor, List<String> topicNames);
 
   /**
    * Unblocks a blocked entry and resets the attempt count so that it will be retried again.

transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/TransactionOutboxImpl.java

@@ -150,33 +150,17 @@ public boolean flush(Executor executor) {
   }
 
   @Override
-  public boolean flushTopic(Executor executor, String topicName) {
+  public boolean flushTopics(Executor executor, List<String> topicNames) {
     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);
+    log.debug("Flushing selected topics {}", topicNames);
+    return doFlush(
+        tx ->
+            uncheckedly(
+                () -> persistor.selectNextInSelectedTopics(tx, topicNames, flushBatchSize, now)));
   }
 
   private void expireIdempotencyProtection(Instant now) {

transactionoutbox-testing/src/main/java/com/gruelbox/transactionoutbox/testing/AbstractAcceptanceTest.java

@@ -505,57 +505,59 @@ final void retryBehaviour() throws Exception {
   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))
+            .instantiator(
+                Instantiator.using(
+                    clazz ->
+                        (InterfaceProcessor)
+                            (foo, bar) ->
+                                LOGGER.info(
+                                    "Entered the method to process successfully. Processing ({}, {})",
+                                    foo,
+                                    bar)))
             .submitter(Submitter.withExecutor(singleThreadPool))
             .attemptFrequency(Duration.ofMillis(500))
             .listener(processedEntryListener)
-            .blockAfterAttempts(5)
             .build();
 
     clearOutbox();
 
     var selectedTopic = "SELECTED_TOPIC";
-    withRunningFlusher(
-        outbox,
+    transactionManager.inTransaction(
         () -> {
-          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");
-              });
+          outbox
+              .with()
+              .ordered(selectedTopic)
+              .schedule(InterfaceProcessor.class)
+              .process(1, "Whoo");
+          outbox
+              .with()
+              .ordered("IGNORED_TOPIC")
+              .schedule(InterfaceProcessor.class)
+              .process(2, "Wheeeee");
+        });
+    assertFalse(
+        successLatch.await(5, SECONDS),
+        "At this point, nothing should have been picked up for processing");
 
-          assertTrue(successLatch.await(20, SECONDS), "Timeout waiting for success");
-          var successes = processedEntryListener.getSuccessfulEntries();
-          var failures = processedEntryListener.getFailingEntries();
+    outbox.flushTopics(singleThreadPool, selectedTopic);
 
-          // 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());
+    assertTrue(successLatch.await(5, SECONDS), "Should have successfully processed something");
 
-          // a single failure is expected; from the unordered submission (as these run automatically
-          assertEquals(1, failures.stream().filter(it -> it.getTopic() == null).count());
+    var successes = processedEntryListener.getSuccessfulEntries();
+    var failures = processedEntryListener.getFailingEntries();
 
-          // all other failures accounted for belong to the selected topic
-          assertEquals(
-              2, failures.stream().filter(it -> selectedTopic.equals(it.getTopic())).count());
-        },
-        singleThreadPool,
-        selectedTopic);
+    // 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::getTopic).distinct().count());
+    assertEquals(selectedTopic, successes.get(0).getTopic());
+
+    // no failures expected
+    assertEquals(0, failures.size());
   }
 
   @Test

transactionoutbox-testing/src/main/java/com/gruelbox/transactionoutbox/testing/BaseTest.java

@@ -94,7 +94,7 @@ protected void withRunningFlusher(
                   log.info("Starting flush...");
                   while (topicName == null
                       ? outbox.flush(executor)
-                      : outbox.flushTopic(executor, topicName)) {
+                      : outbox.flushTopics(executor, topicName)) {
                     log.info("More work to do...");
                   }
                   log.info("Done!");