2.x fix unstable messaging tests (backport) (#8453)

* Intermittent SubscriberPublMsgToPaylRetComplVoidBean test fix Signed-off-by: Daniel Kec <[email protected]> * Intermittent SubscriberPublMsgToMsgRetComplBean test fix Signed-off-by: Daniel Kec <[email protected]> * Fix copyright --------- Signed-off-by: Daniel Kec <[email protected]> Co-authored-by: Daniel Kec <[email protected]>
helidon-io · Mar 5, 2024 · 4471a92 · 4471a92
1 parent 9990916
commit 4471a92
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 4 deletions.
diff --git a/...o/helidon/microprofile/messaging/inner/subscriber/SubscriberPublMsgToMsgRetComplBean.java b/...o/helidon/microprofile/messaging/inner/subscriber/SubscriberPublMsgToMsgRetComplBean.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2020, 2023 Oracle and/or its affiliates.
+ * Copyright (c) 2020, 2024 Oracle and/or its affiliates.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,6 +19,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 
 import javax.enterprise.context.ApplicationScoped;
@@ -42,6 +43,7 @@
 public class SubscriberPublMsgToMsgRetComplBean implements AssertableTestBean, AsyncTestBean {
 
     CopyOnWriteArraySet<String> resultData = new CopyOnWriteArraySet<>();
+    private final CountDownLatch countDownLatch = new CountDownLatch(TEST_DATA.size());
 
     private final ExecutorService executor = createExecutor();
 
@@ -54,11 +56,15 @@ public Publisher<Message<String>> sourceForCsVoidMessage() {
 
     @Incoming("cs-void-message")
     public CompletionStage<Void> consumeMessageAndReturnCompletionStageOfVoid(Message<String> message) {
-        return CompletableFuture.runAsync(() -> resultData.add(message.getPayload()), executor);
+        return CompletableFuture.runAsync(() -> {
+            resultData.add(message.getPayload());
+            countDownLatch.countDown();
+        }, executor);
     }
 
     @Override
     public void assertValid() {
+        await("Messages not delivered in time!", countDownLatch);
         assertWithOrigin("Result doesn't match", resultData, is(TEST_DATA));
     }
 

diff --git a/...idon/microprofile/messaging/inner/subscriber/SubscriberPublMsgToPaylRetComplVoidBean.java b/...idon/microprofile/messaging/inner/subscriber/SubscriberPublMsgToPaylRetComplVoidBean.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2020, 2023 Oracle and/or its affiliates.
+ * Copyright (c) 2020, 2024 Oracle and/or its affiliates.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,6 +19,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 
 import javax.enterprise.context.ApplicationScoped;
@@ -42,6 +43,7 @@
 public class SubscriberPublMsgToPaylRetComplVoidBean implements AssertableTestBean, AsyncTestBean {
 
     CopyOnWriteArraySet<String> resultData = new CopyOnWriteArraySet<>();
+    private final CountDownLatch countDownLatch = new CountDownLatch(TEST_DATA.size());
     private final ExecutorService executor = createExecutor();
 
     @Outgoing("cs-void-payload")
@@ -53,11 +55,15 @@ public Publisher<Message<String>> sourceForCsVoidPayload() {
 
     @Incoming("cs-void-payload")
     public CompletionStage<Void> consumePayloadAndReturnCompletionStageOfVoid(String payload) {
-        return CompletableFuture.runAsync(() -> resultData.add(payload), executor);
+        return CompletableFuture.runAsync(() -> {
+            resultData.add(payload);
+            countDownLatch.countDown();
+        }, executor);
     }
 
     @Override
     public void assertValid() {
+        await("Messages not delivered in time!", countDownLatch);
         assertWithOrigin("Result doesn't match", resultData, is(TEST_DATA));
     }