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/08/04 08:41:24 UTC
(camel) branch main updated: CAMEL-23129: Fix CountDownLatch count mismatch in ThreadPerTaskSedaConsumer
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 96e78f96c127 CAMEL-23129: Fix CountDownLatch count mismatch in ThreadPerTaskSedaConsumer
96e78f96c127 is described below
commit 96e78f96c1271dabd918feb0232b744d5168682d
Author: Guillaume Nodet <gn...@gmail.com>
AuthorDate: Tue Aug 4 10:41:16 2026 +0200
CAMEL-23129: Fix CountDownLatch count mismatch in ThreadPerTaskSedaConsumer
SedaConsumer.doStart() creates latch with concurrentConsumers count, but
ThreadPerTaskSedaConsumer uses a single coordinator thread. When
concurrentConsumers > 1 (used as concurrency limit), the latch count
exceeds the coordinator thread count, causing prepareShutdown() to wait
the full shutdown timeout. Fix by overriding doStart() to set latch
count to 1, matching the single coordinator thread.
Co-Authored-By: Claude Opus 4.6 <no...@anthropic.com>
---
.../apache/camel/component/seda/SedaConsumer.java | 2 +-
.../component/seda/ThreadPerTaskSedaConsumer.java | 13 ++++++++
.../seda/ThreadPerTaskSedaConsumerTest.java | 38 +++++++++++++++++++---
3 files changed, 48 insertions(+), 5 deletions(-)
diff --git a/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaConsumer.java b/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaConsumer.java
index 9432ed7860bc..e4503f9f7ca8 100644
--- a/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaConsumer.java
+++ b/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaConsumer.java
@@ -49,7 +49,7 @@ public class SedaConsumer extends DefaultConsumer implements Runnable, ShutdownA
private static final Logger LOG = LoggerFactory.getLogger(SedaConsumer.class);
private final AtomicInteger taskCount = new AtomicInteger();
- private volatile CountDownLatch latch;
+ protected volatile CountDownLatch latch;
private volatile boolean shutdownPending;
private volatile boolean forceShutdown;
private ExecutorService executor;
diff --git a/components/camel-seda/src/main/java/org/apache/camel/component/seda/ThreadPerTaskSedaConsumer.java b/components/camel-seda/src/main/java/org/apache/camel/component/seda/ThreadPerTaskSedaConsumer.java
index d41c09e17f7d..a764c9e2fe39 100644
--- a/components/camel-seda/src/main/java/org/apache/camel/component/seda/ThreadPerTaskSedaConsumer.java
+++ b/components/camel-seda/src/main/java/org/apache/camel/component/seda/ThreadPerTaskSedaConsumer.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.component.seda;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
@@ -68,6 +69,18 @@ public class ThreadPerTaskSedaConsumer extends SedaConsumer {
.newSingleThreadExecutor(this, getEndpoint().getEndpointUri() + "-coordinator");
}
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+ // SedaConsumer.doStart() creates the latch with concurrentConsumers count,
+ // but ThreadPerTaskSedaConsumer uses a single coordinator thread that polls
+ // the queue and dispatches each exchange to the task executor.
+ // The concurrentConsumers value is used here as a concurrency limit for the
+ // task executor (via a Semaphore), not as the number of polling threads.
+ // Override the latch to match the actual coordinator thread count (1).
+ latch = new CountDownLatch(1);
+ }
+
@Override
protected void setupTasks() {
// Create task executor - uses virtual threads when enabled
diff --git a/core/camel-core/src/test/java/org/apache/camel/component/seda/ThreadPerTaskSedaConsumerTest.java b/core/camel-core/src/test/java/org/apache/camel/component/seda/ThreadPerTaskSedaConsumerTest.java
index b3cf9e5af948..8cc6e7b6bd3c 100644
--- a/core/camel-core/src/test/java/org/apache/camel/component/seda/ThreadPerTaskSedaConsumerTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/component/seda/ThreadPerTaskSedaConsumerTest.java
@@ -16,18 +16,22 @@
*/
package org.apache.camel.component.seda;
+import java.util.concurrent.TimeUnit;
+
import org.apache.camel.ContextTestSupport;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
import org.junit.jupiter.api.Test;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
/**
* Test for the virtualThreadPerTask mode of SEDA consumer
*/
-public class ThreadPerTaskSedaConsumerTest extends ContextTestSupport {
+class ThreadPerTaskSedaConsumerTest extends ContextTestSupport {
@Test
- public void testVirtualThreadPerTask() throws Exception {
+ void testVirtualThreadPerTask() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:result");
mock.expectedMessageCount(10);
@@ -39,7 +43,7 @@ public class ThreadPerTaskSedaConsumerTest extends ContextTestSupport {
}
@Test
- public void testVirtualThreadPerTaskWithConcurrencyLimit() throws Exception {
+ void testVirtualThreadPerTaskWithConcurrencyLimit() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:limited");
mock.expectedMessageCount(5);
@@ -51,7 +55,7 @@ public class ThreadPerTaskSedaConsumerTest extends ContextTestSupport {
}
@Test
- public void testVirtualThreadPerTaskHighThroughput() throws Exception {
+ void testVirtualThreadPerTaskHighThroughput() throws Exception {
int messageCount = 100;
MockEndpoint mock = getMockEndpoint("mock:throughput");
mock.expectedMessageCount(messageCount);
@@ -63,6 +67,32 @@ public class ThreadPerTaskSedaConsumerTest extends ContextTestSupport {
mock.assertIsSatisfied();
}
+ @Test
+ void testShutdownWithConcurrencyLimitCompletesQuickly() throws Exception {
+ // Send messages so the route is actively used
+ for (int i = 0; i < 5; i++) {
+ template.sendBody("seda:limited?virtualThreadPerTask=true&concurrentConsumers=2", "Message " + i);
+ }
+
+ MockEndpoint mock = getMockEndpoint("mock:limited");
+ mock.expectedMessageCount(5);
+ mock.assertIsSatisfied();
+
+ // Stop the context and verify it completes quickly.
+ // Before the fix, the CountDownLatch was initialized with concurrentConsumers
+ // count (2) but only 1 coordinator thread counts down, so prepareShutdown()
+ // would wait the full shutdown timeout before proceeding.
+ long start = System.nanoTime();
+ context.stop();
+ long elapsed = TimeUnit.NANOSECONDS.toSeconds(System.nanoTime() - start);
+
+ // Shutdown should complete well within the default timeout (300s).
+ // Use a generous 30s bound to avoid flakiness, but this is still much less
+ // than the full shutdown strategy timeout that would be hit without the fix.
+ assertTrue(elapsed < 30, "Context stop took " + elapsed + "s, expected < 30s. "
+ + "The CountDownLatch count likely does not match the coordinator thread count.");
+ }
+
@Override
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {