This is an automated email from the ASF dual-hosted git repository.
davsclaus 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 b94a1ce63df2 CAMEL-24024: Harden InOutQueueProducerAsyncLoadTest
against CI flakes
b94a1ce63df2 is described below
commit b94a1ce63df2c625660b0b0d64414095efe9b08c
Author: Omar Atie <[email protected]>
AuthorDate: Wed Aug 5 02:17:12 2026 -0700
CAMEL-24024: Harden InOutQueueProducerAsyncLoadTest against CI flakes
Poll with Awaitility until inflight repository is drained instead of
asserting immediately after executor shutdown. Track JMS listener
failures with AtomicInteger instead of fail() from callback thread.
Use try/finally for executor shutdown and drop unnecessary public modifiers.
Closes #25333
Co-authored-by: Cursor Agent <[email protected]>
Co-authored-by: Claude Opus 4.6 <[email protected]>
---
.../producer/InOutQueueProducerAsyncLoadTest.java | 84 ++++++++++++----------
1 file changed, 45 insertions(+), 39 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 d1b228a88b52..50e1bbc2627c 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
@@ -36,22 +36,26 @@ 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 {
+class InOutQueueProducerAsyncLoadTest extends JmsTestSupport {
private static final String TEST_DESTINATION_NAME =
"in.out.queue.producer.test.InOutQueueProducerAsyncLoadTest";
+ private static final int MESSAGE_COUNT = 500;
+ private static final long LATCH_TIMEOUT_SECONDS = 60;
+ private static final long EXECUTOR_SHUTDOWN_SECONDS = 10;
+ private static final long INFLIGHT_TIMEOUT_SECONDS = 30;
+
private MessageConsumer mc1;
private MessageConsumer mc2;
-
- public InOutQueueProducerAsyncLoadTest() {
- }
+ private final AtomicInteger listenerErrors = new AtomicInteger();
@BeforeEach
- public void setupConsumers() throws Exception {
+ void setupConsumers() throws Exception {
+ listenerErrors.set(0);
mc1 = createQueueConsumer(TEST_DESTINATION_NAME + ".request");
mc2 = createQueueConsumer(TEST_DESTINATION_NAME + ".request");
mc1.setMessageListener(new MyMessageListener());
@@ -59,7 +63,7 @@ public class InOutQueueProducerAsyncLoadTest extends
JmsTestSupport {
}
@AfterEach
- public void cleanupConsumers() throws JMSException {
+ void cleanupConsumers() throws JMSException {
MyMessageListener l1 = (MyMessageListener) mc1.getMessageListener();
l1.close();
mc1.close();
@@ -70,43 +74,42 @@ public class InOutQueueProducerAsyncLoadTest extends
JmsTestSupport {
/**
* Test to verify that when using the consumer listener for the InOut
producer we get the correct message back.
- *
- * @throws Exception
*/
@Test
- public void testInOutQueueProducer() throws Exception {
- final int messageCount = 500;
- final CountDownLatch latch = new CountDownLatch(messageCount);
+ void testInOutQueueProducer() throws Exception {
+ final CountDownLatch latch = new CountDownLatch(MESSAGE_COUNT);
final AtomicInteger failures = new AtomicInteger();
ExecutorService executor = Executors.newFixedThreadPool(2);
-
- for (int i = 1; i <= messageCount; i++) {
- final int tempI = i;
- 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();
- }
- });
+ try {
+ for (int i = 1; i <= MESSAGE_COUNT; i++) {
+ final int tempI = i;
+ 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 (Throwable e) {
+ failures.incrementAndGet();
+ log.error("Failed to process message {}", tempI, e);
+ } finally {
+ latch.countDown();
+ }
+ });
+ }
+ assertTrue(latch.await(LATCH_TIMEOUT_SECONDS, SECONDS), "Not all
messages were processed within the timeout");
+ assertEquals(0, failures.get(), "Some messages failed during
processing");
+ assertEquals(0, listenerErrors.get(), "Some JMS listener callbacks
failed");
+ } finally {
+ executor.shutdown();
}
- // 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");
+ assertTrue(executor.awaitTermination(EXECUTOR_SHUTDOWN_SECONDS,
SECONDS), "Executor did not terminate in time");
- executor.shutdown();
- assertTrue(executor.awaitTermination(10, SECONDS), "Executor did not
terminate in time");
-
- // verify all inflight messages have completed
- assertEquals(0, context.getInflightRepository().size());
+ // async route completion can lag behind request/reply futures — poll
until inflight is drained
+ await().atMost(INFLIGHT_TIMEOUT_SECONDS, SECONDS)
+ .untilAsserted(() -> assertEquals(0,
context.getInflightRepository().size()));
}
@Override
@@ -141,12 +144,15 @@ public class InOutQueueProducerAsyncLoadTest extends
JmsTestSupport {
}
mp.send(response);
} catch (JMSException e) {
- fail(e.getLocalizedMessage());
+ listenerErrors.incrementAndGet();
+ log.error("Failed to process JMS message in test listener", e);
}
}
public void close() throws JMSException {
- mp.close();
+ if (mp != null) {
+ mp.close();
+ }
}
}
}
