(camel) branch main updated: CAMEL-24027: Fix flaky InOutQueueProducerAsyncLoadTest (#24626)

[gnodet]

This is an automated email from the ASF dual-hosted git repository.

gnodet pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git


The following commit(s) were added to refs/heads/main by this push:
     new 9493e1d0c6d3 CAMEL-24027: Fix flaky InOutQueueProducerAsyncLoadTest 
(#24626)
9493e1d0c6d3 is described below

commit 9493e1d0c6d3bada7c4e6d483bb5f3c3216aab33
Author: Guillaume Nodet <gno...@gmail.com>
AuthorDate: Sun Jul 12 22:05:51 2026 +0200

    CAMEL-24027: Fix flaky InOutQueueProducerAsyncLoadTest (#24626)
    
    CAMEL-24027: Fix flaky InOutQueueProducerAsyncLoadTest
    
    Co-authored-by: Claude Opus 4.6 <nore...@anthropic.com>
---
 .../producer/InOutQueueProducerAsyncLoadTest.java  | 46 ++++++++++++----------
 1 file changed, 26 insertions(+), 20 deletions(-)

diff --git 
a/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/producer/InOutQueueProducerAsyncLoadTest.java
 
b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/producer/InOutQueueProducerAsyncLoadTest.java
index fcdd949090cf..d1b228a88b52 100644
--- 
a/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/producer/InOutQueueProducerAsyncLoadTest.java
+++ 
b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/producer/InOutQueueProducerAsyncLoadTest.java
@@ -16,8 +16,10 @@
  */
 package org.apache.camel.component.sjms.producer;
 
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import jakarta.jms.JMSException;
 import jakarta.jms.Message;
@@ -34,9 +36,9 @@ import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 import static java.util.concurrent.TimeUnit.SECONDS;
-import static org.awaitility.Awaitility.await;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 
 public class InOutQueueProducerAsyncLoadTest extends JmsTestSupport {
@@ -73,32 +75,36 @@ public class InOutQueueProducerAsyncLoadTest extends 
JmsTestSupport {
      */
     @Test
     public void testInOutQueueProducer() throws Exception {
+        final int messageCount = 500;
+        final CountDownLatch latch = new CountDownLatch(messageCount);
+        final AtomicInteger failures = new AtomicInteger();
 
         ExecutorService executor = Executors.newFixedThreadPool(2);
 
-        for (int i = 1; i <= 5000; i++) {
+        for (int i = 1; i <= messageCount; i++) {
             final int tempI = i;
-            Runnable worker = new Runnable() {
-
-                @Override
-                public void run() {
-                    try {
-                        final String requestText = "Message " + tempI;
-                        final String responseText = "Response Message " + 
tempI;
-                        String response = template.requestBody("direct:start", 
requestText, String.class);
-                        assertNotNull(response);
-                        assertEquals(responseText, response);
-                    } catch (Exception e) {
-                        log.error("Failed to process message {}", tempI, e);
-                    }
+            executor.execute(() -> {
+                try {
+                    final String requestText = "Message " + tempI;
+                    final String responseText = "Response Message " + tempI;
+                    String response = template.requestBody("direct:start", 
requestText, String.class);
+                    assertNotNull(response);
+                    assertEquals(responseText, response);
+                } catch (Exception e) {
+                    failures.incrementAndGet();
+                    log.error("Failed to process message {}", tempI, e);
+                } finally {
+                    latch.countDown();
                 }
-            };
-            executor.execute(worker);
+            });
         }
-        // shut down the executor and wait for all submitted tasks to complete
+        // wait for all submitted tasks to complete
+        assertTrue(latch.await(60, SECONDS), "Not all messages were processed 
within the timeout");
+        assertEquals(0, failures.get(), "Some messages failed during 
processing");
+
         executor.shutdown();
-        await().atMost(30, SECONDS)
-                .until(() -> executor.isTerminated());
+        assertTrue(executor.awaitTermination(10, SECONDS), "Executor did not 
terminate in time");
+
         // verify all inflight messages have completed
         assertEquals(0, context.getInflightRepository().size());
     }