added concurrent test for spring

gruelbox · Oct 23, 2024 · e5e8905 · e5e8905
1 parent 86378f8
commit e5e8905
Show file tree

Hide file tree

Showing 3 changed files with 75 additions and 4 deletions.
diff --git a/...ionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/DefaultSequenceGenerator.java b/...ionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/DefaultSequenceGenerator.java
@@ -2,6 +2,8 @@
 
 import com.gruelbox.transactionoutbox.spi.Utils;
 import java.sql.ResultSet;
+import java.util.function.Predicate;
+
 import lombok.AccessLevel;
 import lombok.Builder;
 import lombok.RequiredArgsConstructor;

diff --git a/transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/spi/Utils.java b/transactionoutbox-core/src/main/java/com/gruelbox/transactionoutbox/spi/Utils.java
@@ -3,10 +3,17 @@
 import com.gruelbox.transactionoutbox.ThrowingRunnable;
 import com.gruelbox.transactionoutbox.UncheckedException;
 import java.sql.SQLIntegrityConstraintViolationException;
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
 import java.util.concurrent.Callable;
+import java.util.function.Predicate;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
 import lombok.extern.slf4j.Slf4j;
 import org.slf4j.Logger;
 import org.slf4j.event.Level;
@@ -25,12 +32,33 @@ public static boolean safelyRun(String gerund, ThrowingRunnable runnable) {
     }
   }
 
-  public static boolean indexViolation(Exception e) {
-    return (e instanceof SQLIntegrityConstraintViolationException)
-        || (e.getClass().getName().equals("org.postgresql.util.PSQLException")
+  public static boolean indexViolation(Throwable e) {
+    boolean indexViolation = (e instanceof SQLIntegrityConstraintViolationException)
+            || (e.getClass().getName().equals("org.postgresql.util.PSQLException")
             && e.getMessage().contains("constraint"))
-        || (e.getClass().getName().equals("com.microsoft.sqlserver.jdbc.SQLServerException")
+            || (e.getClass().getName().equals("com.microsoft.sqlserver.jdbc.SQLServerException")
             && e.getMessage().contains("duplicate key"));
+    if (indexViolation) {
+      return true;
+    }
+
+    // we check cause, because the error can be thrown into the proxy
+    // and wrapped in java.lang.reflect.UndeclaredThrowableException
+    Set<Throwable> throwables = new HashSet<>();
+    Throwable cause = e.getCause();
+    while (cause != null && throwables.add(cause)) {
+      if ( (cause instanceof SQLIntegrityConstraintViolationException)
+              || (cause.getClass().getName().equals("org.postgresql.util.PSQLException")
+              && cause.getMessage().contains("constraint"))
+              || (cause.getClass().getName().equals("com.microsoft.sqlserver.jdbc.SQLServerException")
+              && cause.getMessage().contains("duplicate key"))) {
+        return true;
+      }
+
+      cause = cause.getCause();
+    }
+
+    return false;
   }
 
   @SuppressWarnings("unused")

diff --git a/...ava/com/gruelbox/transactionoutbox/spring/example/EventuallyConsistentControllerTest.java b/...ava/com/gruelbox/transactionoutbox/spring/example/EventuallyConsistentControllerTest.java
@@ -4,8 +4,16 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
 
 import java.net.URL;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -85,4 +93,37 @@ void testCheckOrdered() {
                 assertThat(externalQueueService.getSent())
                     .containsExactly(joe, dave, neil, tupac, jeff));
   }
+
+  @Test
+  void testCheckOrderedConcurrent() {
+      ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
+
+      try {
+          var joe = new Customer(1L, "Joe", "Strummer");
+          var dave = new Customer(2L, "Dave", "Grohl");
+          var neil = new Customer(3L, "Neil", "Diamond");
+          var tupac = new Customer(4L, "Tupac", "Shakur");
+          var jeff = new Customer(5L, "Jeff", "Mills");
+
+          List<CompletableFuture<Void>> tasks = Stream.of(joe, dave, neil, tupac, jeff)
+                  .map(customer -> CompletableFuture.runAsync(() -> {
+                      var url = base.toString() + "/customer?ordered=true";
+                      assertTrue(template.postForEntity(url, customer, Void.class).getStatusCode().is2xxSuccessful());
+                  }, executorService).exceptionally((throwable) -> fail("Can't create ordered task for customer " + customer, throwable)))
+                  .collect(Collectors.toUnmodifiableList());
+
+          CompletableFuture.allOf(tasks.toArray(CompletableFuture[]::new)).join();
+
+          await()
+                  .atMost(10, SECONDS)
+                  .pollDelay(1, SECONDS)
+                  .untilAsserted(
+                          () ->
+                                  assertThat(externalQueueService.getSent())
+                                          .containsExactlyInAnyOrder(joe, dave, neil, tupac, jeff));
+
+      } finally {
+          executorService.shutdownNow();
+      }
+  }
 }