From b9b28e11535328d1b781674c816b75d32e277ec9 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 5 Aug 2026 03:06:21 +0000 Subject: [PATCH 1/2] CAMEL-24024: Harden InOutQueueProducerAsyncLoadTest against CI flakes Poll until inflight repository is drained instead of asserting immediately after executor shutdown. Track JMS listener failures with AtomicInteger instead of fail() from the callback thread. Use try/finally for executor shutdown and drop unnecessary public modifiers. Co-authored-by: Claude Opus 4.6 --- .../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 d1b228a88b52f..c6a13ee9889c7 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.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 void setupConsumers() throws Exception { } @AfterEach - public void cleanupConsumers() throws JMSException { + void cleanupConsumers() throws JMSException { MyMessageListener l1 = (MyMessageListener) mc1.getMessageListener(); l1.close(); mc1.close(); @@ -70,43 +74,42 @@ public void cleanupConsumers() throws JMSException { /** * 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 (Exception 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(); + assertTrue(executor.awaitTermination(EXECUTOR_SHUTDOWN_SECONDS, SECONDS), "Executor did not terminate in time"); } - // 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(); - 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 void onMessage(Message message) { } 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(); + } } } } From b8a16313d11ceddec983287c95b66aa2add2d9ff Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 5 Aug 2026 03:10:33 +0000 Subject: [PATCH 2/2] CAMEL-24024: Catch Throwable in async workers and fix finally asserts Count assertion failures from worker threads and avoid masking the primary test failure when executor shutdown times out in finally. Co-authored-by: Claude Opus 4.6 --- .../sjms/producer/InOutQueueProducerAsyncLoadTest.java | 4 ++-- 1 file changed, 2 insertions(+), 2 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 c6a13ee9889c7..50e1bbc2627c8 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 @@ -91,7 +91,7 @@ void testInOutQueueProducer() throws Exception { String response = template.requestBody("direct:start", requestText, String.class); assertNotNull(response); assertEquals(responseText, response); - } catch (Exception e) { + } catch (Throwable e) { failures.incrementAndGet(); log.error("Failed to process message {}", tempI, e); } finally { @@ -104,8 +104,8 @@ void testInOutQueueProducer() throws Exception { assertEquals(0, listenerErrors.get(), "Some JMS listener callbacks failed"); } finally { executor.shutdown(); - assertTrue(executor.awaitTermination(EXECUTOR_SHUTDOWN_SECONDS, SECONDS), "Executor did not terminate in time"); } + assertTrue(executor.awaitTermination(EXECUTOR_SHUTDOWN_SECONDS, SECONDS), "Executor did not terminate in time"); // async route completion can lag behind request/reply futures — poll until inflight is drained await().atMost(INFLIGHT_TIMEOUT_SECONDS, SECONDS)