improve Vert.x tests

The improvements mainly consist of running each individual call on a separate duplicated context, which better simulates what happens in practice (at least in Quarkus). Further, a new test for timeouts with Vert.x was added.
smallrye · Jan 3, 2025 · 9118c62 · 9118c62
1 parent 0e2d97d
commit 9118c62
Show file tree

Hide file tree

Showing 15 changed files with 469 additions and 104 deletions.
diff --git a/testsuite/integration/src/test/java/io/smallrye/faulttolerance/vertx/ContextDescription.java b/testsuite/integration/src/test/java/io/smallrye/faulttolerance/vertx/ContextDescription.java
@@ -0,0 +1,47 @@
+package io.smallrye.faulttolerance.vertx;
+
+import java.util.Locale;
+import java.util.Objects;
+
+public final class ContextDescription {
+    public final ExecutionStyle executionStyle;
+    public final String contextClass;
+    public final String uuid;
+    public final String contextHash;
+
+    ContextDescription(ExecutionStyle executionStyle, String contextClass, String uuid, String contextHash) {
+        this.executionStyle = executionStyle;
+        this.contextClass = contextClass;
+        this.contextHash = contextHash;
+        this.uuid = uuid;
+    }
+
+    public boolean isDuplicatedContext() {
+        return "DuplicatedContext".equals(contextClass);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (!(o instanceof ContextDescription)) {
+            return false;
+        }
+        ContextDescription that = (ContextDescription) o;
+        return Objects.equals(executionStyle, that.executionStyle)
+                && Objects.equals(contextClass, that.contextClass)
+                && Objects.equals(uuid, that.uuid)
+                && Objects.equals(contextHash, that.contextHash);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(executionStyle, contextClass, uuid, contextHash);
+    }
+
+    @Override
+    public String toString() {
+        return executionStyle.toString().toLowerCase(Locale.ROOT)
+                + "|" + contextClass
+                + "|" + uuid
+                + "|" + contextHash;
+    }
+}
diff --git a/testsuite/integration/src/test/java/io/smallrye/faulttolerance/vertx/ExecutionStyle.java b/testsuite/integration/src/test/java/io/smallrye/faulttolerance/vertx/ExecutionStyle.java
@@ -0,0 +1,7 @@
+package io.smallrye.faulttolerance.vertx;
+
+public enum ExecutionStyle {
+    EVENT_LOOP,
+    WORKER,
+    UNKNOWN,
+}
diff --git a/testsuite/integration/src/test/java/io/smallrye/faulttolerance/vertx/VertxContext.java b/testsuite/integration/src/test/java/io/smallrye/faulttolerance/vertx/VertxContext.java
@@ -0,0 +1,71 @@
+package io.smallrye.faulttolerance.vertx;
+
+import java.util.UUID;
+
+import io.vertx.core.Context;
+import io.vertx.core.impl.ContextInternal;
+
+// assumes that verticles are not used and no context is created explicitly,
+// which means that all contexts are event loop contexts
+public class VertxContext {
+    private final ContextInternal context;
+
+    public static VertxContext current() {
+        return new VertxContext(ContextInternal.current());
+    }
+
+    private VertxContext(ContextInternal context) {
+        this.context = context;
+    }
+
+    public VertxContext duplicate() {
+        return new VertxContext(context.duplicate());
+    }
+
+    public void execute(ExecutionStyle style, Runnable runnable) {
+        switch (style) {
+            case EVENT_LOOP:
+                context.runOnContext(ignored -> {
+                    runnable.run();
+                });
+                break;
+            case WORKER:
+                context.executeBlocking(() -> {
+                    runnable.run();
+                    return null;
+                });
+                break;
+            default:
+                throw new UnsupportedOperationException("" + style);
+        }
+    }
+
+    public void setTimer(long delayInMillis, Runnable runnable) {
+        boolean moveToWorker = Context.isOnWorkerThread();
+        context.setTimer(delayInMillis, ignored -> {
+            if (moveToWorker) {
+                context.executeBlocking(() -> {
+                    runnable.run();
+                    return null;
+                });
+            } else {
+                runnable.run();
+            }
+        });
+    }
+
+    public ContextDescription describe() {
+        String uuid = context.getLocal("my-uuid");
+        if (uuid == null) {
+            uuid = UUID.randomUUID().toString();
+            context.putLocal("my-uuid", uuid);
+        }
+
+        ExecutionStyle executionStyle = Context.isOnEventLoopThread()
+                ? ExecutionStyle.EVENT_LOOP
+                : (Context.isOnWorkerThread() ? ExecutionStyle.WORKER : ExecutionStyle.UNKNOWN);
+
+        return new ContextDescription(executionStyle, context.getClass().getSimpleName(), uuid,
+                "" + System.identityHashCode(context));
+    }
+}
diff --git a/...c/test/java/io/smallrye/faulttolerance/vertx/bulkhead/AsyncBulkheadOnVertxThreadTest.java b/...c/test/java/io/smallrye/faulttolerance/vertx/bulkhead/AsyncBulkheadOnVertxThreadTest.java
@@ -3,6 +3,8 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
 
+import java.util.HashSet;
+import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.TimeUnit;
 
@@ -12,39 +14,55 @@
 import org.junit.jupiter.api.Test;
 
 import io.smallrye.faulttolerance.vertx.AbstractVertxTest;
+import io.smallrye.faulttolerance.vertx.ContextDescription;
+import io.smallrye.faulttolerance.vertx.ExecutionStyle;
+import io.smallrye.faulttolerance.vertx.VertxContext;
 
 public class AsyncBulkheadOnVertxThreadTest extends AbstractVertxTest {
     @BeforeEach
     public void setUp() {
-        MyService.invocationThreads.clear();
+        MyService.currentContexts.clear();
     }
 
     @Test
-    public void nonblockingBulkhead(MyService myService) {
-        CopyOnWriteArrayList<Object> results = new CopyOnWriteArrayList<>();
+    public void eventLoop(MyService myService) {
+        test(myService, ExecutionStyle.EVENT_LOOP);
+    }
+
+    @Test
+    public void worker(MyService myService) {
+        test(myService, ExecutionStyle.WORKER);
+    }
+
+    private void test(MyService myService, ExecutionStyle executionStyle) {
+        List<Object> results = new CopyOnWriteArrayList<>();
 
         runOnVertx(() -> {
+            VertxContext ctx = VertxContext.current();
             for (int i = 0; i < 10; i++) {
-                myService.hello().whenComplete((value, error) -> {
-                    results.add(error == null ? value : error);
+                ctx.duplicate().execute(executionStyle, () -> {
+                    MyService.currentContexts.add(VertxContext.current().describe());
+                    myService.hello().whenComplete((value, error) -> {
+                        MyService.currentContexts.add(VertxContext.current().describe());
+                        results.add(error == null ? value : error);
+                    });
                 });
             }
         });
 
-        // 3 immediate invocations + 3 queued invocations + 4 rejected from bulkhead
+        // 3 immediate calls + 3 queued calls + 4 rejected from bulkhead
         await().atMost(5, TimeUnit.SECONDS).until(() -> results.size() == 10);
 
         assertThat(results).haveExactly(6,
                 new Condition<>("Hello!"::equals, "successful result"));
         assertThat(results).haveExactly(4,
                 new Condition<>(it -> it instanceof BulkheadException, "failed result"));
 
-        // 3 immediate invocations + 3 queued invocations
-        // 2 identical items for each invocation
-        assertThat(MyService.invocationThreads).hasSize(12);
-        assertThat(MyService.invocationThreads).allSatisfy(thread -> {
-            assertThat(thread).contains("vert.x-eventloop");
-        });
-        assertThat(MyService.invocationThreads).containsOnly(MyService.invocationThreads.peek());
+        // 3 immediate calls + 3 queued calls: 4 identical items for each
+        // 4 rejected calls: 2 identical items for each
+        assertThat(MyService.currentContexts).hasSize(32);
+        assertThat(MyService.currentContexts).allMatch(it -> executionStyle == it.executionStyle);
+        assertThat(MyService.currentContexts).allMatch(ContextDescription::isDuplicatedContext);
+        assertThat(new HashSet<>(MyService.currentContexts)).hasSize(10);
     }
 }
diff --git a/testsuite/integration/src/test/java/io/smallrye/faulttolerance/vertx/bulkhead/MyService.java b/testsuite/integration/src/test/java/io/smallrye/faulttolerance/vertx/bulkhead/MyService.java
@@ -9,21 +9,22 @@
 
 import org.eclipse.microprofile.faulttolerance.Bulkhead;
 
-import io.smallrye.common.annotation.NonBlocking;
-import io.vertx.core.Vertx;
+import io.smallrye.faulttolerance.api.AsynchronousNonBlocking;
+import io.smallrye.faulttolerance.vertx.ContextDescription;
+import io.smallrye.faulttolerance.vertx.VertxContext;
 
 @ApplicationScoped
 public class MyService {
-    static final Queue<String> invocationThreads = new ConcurrentLinkedQueue<>();
+    static final Queue<ContextDescription> currentContexts = new ConcurrentLinkedQueue<>();
 
-    @NonBlocking
+    @AsynchronousNonBlocking
     @Bulkhead(value = 3, waitingTaskQueue = 3)
     public CompletionStage<String> hello() {
-        invocationThreads.add(Thread.currentThread().getName());
+        currentContexts.add(VertxContext.current().describe());
 
         CompletableFuture<String> result = new CompletableFuture<>();
-        Vertx.currentContext().owner().setTimer(1000, ignored -> {
-            invocationThreads.add(Thread.currentThread().getName());
+        VertxContext.current().setTimer(1000, () -> {
+            currentContexts.add(VertxContext.current().describe());
 
             result.complete("Hello!");
         });

diff --git a/.../io/smallrye/faulttolerance/vertx/bulkhead/retry/AsyncBulkheadRetryOnVertxThreadTest.java b/.../io/smallrye/faulttolerance/vertx/bulkhead/retry/AsyncBulkheadRetryOnVertxThreadTest.java
@@ -3,6 +3,8 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
 
+import java.util.HashSet;
+import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.TimeUnit;
 
@@ -12,40 +14,55 @@
 import org.junit.jupiter.api.Test;
 
 import io.smallrye.faulttolerance.vertx.AbstractVertxTest;
+import io.smallrye.faulttolerance.vertx.ContextDescription;
+import io.smallrye.faulttolerance.vertx.ExecutionStyle;
+import io.smallrye.faulttolerance.vertx.VertxContext;
 
 public class AsyncBulkheadRetryOnVertxThreadTest extends AbstractVertxTest {
     @BeforeEach
     public void setUp() {
-        MyService.invocationThreads.clear();
+        MyService.currentContexts.clear();
     }
 
     @Test
-    public void nonblockingBulkhead(MyService myService) {
-        CopyOnWriteArrayList<Object> results = new CopyOnWriteArrayList<>();
+    public void eventLoop(MyService myService) {
+        test(myService, ExecutionStyle.EVENT_LOOP);
+    }
+
+    @Test
+    public void worker(MyService myService) {
+        test(myService, ExecutionStyle.WORKER);
+    }
+
+    private void test(MyService myService, ExecutionStyle executionStyle) {
+        List<Object> results = new CopyOnWriteArrayList<>();
 
         runOnVertx(() -> {
+            VertxContext ctx = VertxContext.current();
             for (int i = 0; i < 20; i++) {
-                myService.hello().whenComplete((value, error) -> {
-                    results.add(error == null ? value : error);
+                ctx.duplicate().execute(executionStyle, () -> {
+                    MyService.currentContexts.add(VertxContext.current().describe());
+                    myService.hello().whenComplete((value, error) -> {
+                        MyService.currentContexts.add(VertxContext.current().describe());
+                        results.add(error == null ? value : error);
+                    });
                 });
             }
         });
 
-        // 3 immediate invocations + 3 immediately queued invocations + 6 successfully retried rejections from bulkhead
-        // + 8 unsuccessfully retried rejections from bulkhead
+        // 3 immediate calls + 3 immediately queued calls + 6 successfully retried rejections + 8 unsuccessfully retried rejections
         await().atMost(5, TimeUnit.SECONDS).until(() -> results.size() == 20);
 
         assertThat(results).haveExactly(12,
                 new Condition<>("Hello!"::equals, "successful result"));
         assertThat(results).haveExactly(8,
                 new Condition<>(it -> it instanceof BulkheadException, "failed result"));
 
-        // 3 immediate invocations + 3 queued invocations + 6 successfully retried rejections from bulkhead
-        // 2 identical items for each invocation
-        assertThat(MyService.invocationThreads).hasSize(24);
-        assertThat(MyService.invocationThreads).allSatisfy(thread -> {
-            assertThat(thread).contains("vert.x-eventloop");
-        });
-        assertThat(MyService.invocationThreads).containsOnly(MyService.invocationThreads.peek());
+        // 3 immediate calls + 3 queued calls + 6 successfully retried rejections: 4 identical items for each
+        // 8 unsuccessfully retried rejections: 2 identical items for each
+        assertThat(MyService.currentContexts).hasSize(64);
+        assertThat(MyService.currentContexts).allMatch(it -> executionStyle == it.executionStyle);
+        assertThat(MyService.currentContexts).allMatch(ContextDescription::isDuplicatedContext);
+        assertThat(new HashSet<>(MyService.currentContexts)).hasSize(20);
     }
 }
diff --git a/.../integration/src/test/java/io/smallrye/faulttolerance/vertx/bulkhead/retry/MyService.java b/.../integration/src/test/java/io/smallrye/faulttolerance/vertx/bulkhead/retry/MyService.java
@@ -11,26 +11,27 @@
 import org.eclipse.microprofile.faulttolerance.Retry;
 
 import io.smallrye.faulttolerance.api.AsynchronousNonBlocking;
-import io.vertx.core.Vertx;
+import io.smallrye.faulttolerance.vertx.ContextDescription;
+import io.smallrye.faulttolerance.vertx.VertxContext;
 
 @ApplicationScoped
 public class MyService {
-    static final Queue<String> invocationThreads = new ConcurrentLinkedQueue<>();
+    static final Queue<ContextDescription> currentContexts = new ConcurrentLinkedQueue<>();
 
     @AsynchronousNonBlocking
     @Bulkhead(value = 3, waitingTaskQueue = 3)
     @Retry(maxRetries = 1, delay = 1000, jitter = 0)
     public CompletionStage<String> hello() {
-        invocationThreads.add(Thread.currentThread().getName());
+        currentContexts.add(VertxContext.current().describe());
 
         // Note that the Vert.x timer is rather inaccurate. If the retry delay (as defined above)
         // is close to the completion delay (as defined below), this test may fail spuriously.
         // The underlying reason in such case is that the completion is executed _later_ than
         // the retry, which violates the basic assumption of this test.
 
         CompletableFuture<String> result = new CompletableFuture<>();
-        Vertx.currentContext().owner().setTimer(200, ignored -> {
-            invocationThreads.add(Thread.currentThread().getName());
+        VertxContext.current().setTimer(200, () -> {
+            currentContexts.add(VertxContext.current().describe());
 
             result.complete("Hello!");
         });