Skip to content

CAMEL-24024: Harden InOutQueueProducerAsyncLoadTest against CI flakes #25333

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
davsclaus merged 2 commits into from
Aug 5, 2026
Merged

CAMEL-24024: Harden InOutQueueProducerAsyncLoadTest against CI flakes #25333

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b9b28e1
CAMEL-24024: Harden InOutQueueProducerAsyncLoadTest against CI flakes
cursoragent Aug 5, 2026
b8a1631
CAMEL-24024: Catch Throwable in async workers and fix finally asserts
cursoragent Aug 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
84 changes: 45 additions & 39 deletions ...c/test/java/org/apache/camel/component/sjms/producer/InOutQueueProducerAsyncLoadTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,30 +36,34 @@
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());
mc2.setMessageListener(new MyMessageListener());
}

@AfterEach
public void cleanupConsumers() throws JMSException {
void cleanupConsumers() throws JMSException {
MyMessageListener l1 = (MyMessageListener) mc1.getMessageListener();
l1.close();
mc1.close();
Expand All @@ -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 (Throwable e) {

@gnodet gnodet Aug 5, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[Low — style] catch (Throwable e) is slightly broader than necessary. Since the intent is to catch both Exception (from requestBody) and AssertionError (from assertNotNull/assertEquals), using a multi-catch would be more precise:

Suggested change
} catch (Throwable e) {
} catch (Exception | AssertionError e) {

Functionally equivalent here since the error is tracked via failures, so not blocking — just a minor precision improvement.

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
Expand Down Expand Up @@ -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();
}
}
}
}