You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2026/10/01 20:47:25 UTC
(camel) branch main updated: CAMEL-24039: Fix race condition in SedaAsyncProducerTest
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 b3e834972715 CAMEL-24039: Fix race condition in SedaAsyncProducerTest
b3e834972715 is described below
commit b3e834972715d1389b3e5df4583f790ec72495c0
Author: Guillaume Nodet <gn...@gmail.com>
AuthorDate: Thu Oct 1 22:47:17 2026 +0200
CAMEL-24039: Fix race condition in SedaAsyncProducerTest
The test mutated a plain String field from both the test thread ("send")
and the async route thread ("process") with a non-atomic read-modify-write,
relying on a 100ms delay to order them. On loaded CI machines the test
thread could be preempted longer than that, so "process" was recorded first
and the test failed intermittently.
The field is now an AtomicReference updated with updateAndGet, and a
CountDownLatch makes the route processor wait until the test thread has
recorded "send", giving a deterministic happens-before ordering instead of
a timing assumption. Follow-up to PR #24639, which only raised timeouts.
Closes #27217
Co-authored-by: Claude Sonnet 4.6 <no...@anthropic.com>
---
.../component/seda/SedaAsyncProducerTest.java | 44 +++++++++++++++++-----
1 file changed, 35 insertions(+), 9 deletions(-)
diff --git a/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaAsyncProducerTest.java b/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaAsyncProducerTest.java
index 187f28ae9b5c..b1f6333c4006 100644
--- a/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaAsyncProducerTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaAsyncProducerTest.java
@@ -16,8 +16,10 @@
*/
package org.apache.camel.component.seda;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.Exchange;
@@ -38,7 +40,20 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
*/
public class SedaAsyncProducerTest extends ContextTestSupport {
- private String route = "";
+ /**
+ * Thread-safe accumulator for the route execution order. Each step appends its token atomically using
+ * AtomicReference.updateAndGet() to avoid the non-atomic read-modify-write of the plain String field that caused
+ * the flaky ordering assertions.
+ */
+ private final AtomicReference<String> route = new AtomicReference<>("");
+
+ /**
+ * Latch used in testAsyncProducer to guarantee that the test thread records "send" before the route processor is
+ * allowed to record "process". Without this, a heavily loaded CI machine can preempt the test thread for >100ms
+ * after asyncRequestBody() returns, letting the async thread win the race even though a 100ms delay was inserted in
+ * the route.
+ */
+ private final CountDownLatch sendLatch = new CountDownLatch(1);
@Test
public void testAsyncProducer() throws Exception {
@@ -48,12 +63,14 @@ public class SedaAsyncProducerTest extends ContextTestSupport {
// using the new async API we can fire a real async message
Future<String> future = template.asyncRequestBody("direct:start", "Hello World", String.class);
- // I should happen before mock
- route = route + "send";
+ // Record "send" atomically, then signal the route processor it may proceed.
+ // The processor waits on sendLatch so ordering is deterministic regardless of CI load.
+ route.updateAndGet(s -> s + "send");
+ sendLatch.countDown();
MockEndpoint.assertIsSatisfied(context, 30, TimeUnit.SECONDS);
- assertEquals("sendprocess", route, "Send should occur before processor");
+ assertEquals("sendprocess", route.get(), "Send should occur before processor");
// and get the response with the future handle
String response = future.get();
@@ -70,14 +87,18 @@ public class SedaAsyncProducerTest extends ContextTestSupport {
exchange.getIn().setBody("Hello World");
exchange.setPattern(ExchangePattern.InOut);
exchange.setProperty(Exchange.ASYNC_WAIT, WaitForTaskToComplete.IfReplyExpected);
+ // WaitForTaskToComplete.IfReplyExpected on InOut blocks until processing is complete,
+ // so the processor always records "process" before template.send() returns.
+ // The latch is not used in this test; count it down immediately so the processor is not blocked.
+ sendLatch.countDown();
template.send("direct:start", exchange);
- // I should not happen before mock
- route = route + "send";
+ // I should not happen before mock – processor already finished above
+ route.updateAndGet(s -> s + "send");
MockEndpoint.assertIsSatisfied(context, 30, TimeUnit.SECONDS);
- assertEquals("processsend", route, "Send should occur before processor");
+ assertEquals("processsend", route.get(), "Send should occur after processor");
String response = exchange.getMessage().getBody(String.class);
assertEquals("Bye World", response);
@@ -91,8 +112,13 @@ public class SedaAsyncProducerTest extends ContextTestSupport {
errorHandler(noErrorHandler());
from("direct:start").delay(100).process(new Processor() {
- public void process(Exchange exchange) {
- route = route + "process";
+ public void process(Exchange exchange) throws Exception {
+ // Wait until the test thread has recorded "send" (only relevant for
+ // testAsyncProducer; testAsyncProducerWait counts down the latch before send).
+ if (!sendLatch.await(30, TimeUnit.SECONDS)) {
+ throw new IllegalStateException("Timed out waiting for sendLatch");
+ }
+ route.updateAndGet(s -> s + "process");
// set the response
exchange.getMessage().setBody("Bye World");
}