This is an automated email from the ASF dual-hosted git repository. gnodet pushed a commit to branch seda-threadpertasksedaconsumer-fix-countdownlatch in repository https://gitbox.apache.org/repos/asf/camel.git
commit 05c387f49dc3661148164c0c59cf1fc23039c888 Author: Guillaume Nodet <[email protected]> AuthorDate: Fri Jul 31 16:26:46 2026 +0200 CAMEL-23129: Fix CountDownLatch count mismatch in ThreadPerTaskSedaConsumer SedaConsumer.doStart() creates latch = new CountDownLatch(concurrentConsumers), but ThreadPerTaskSedaConsumer uses a single coordinator thread that polls the queue and dispatches each exchange to the task executor. When concurrentConsumers is used as a concurrency limit (e.g. 2), the latch count exceeds the actual coordinator thread count (1), causing prepareShutdown() to always wait the full shutdown timeout before proceeding. Fix by overriding doStart() in ThreadPerTaskSedaConsumer to set latch count to 1, matching the single coordinator thread. The concurrentConsumers value continues to serve as the task executor concurrency limit via the Semaphore. Co-Authored-By: Claude Opus 4.6 <[email protected]> --- .../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() {
