CAMEL-23129: Fix CountDownLatch count mismatch in ThreadPerTaskSedaConsumer - #25281
Open
gnodet wants to merge 1 commit into
Open
CAMEL-23129: Fix CountDownLatch count mismatch in ThreadPerTaskSedaConsumer#25281gnodet wants to merge 1 commit into
gnodet wants to merge 1 commit into
Conversation
…nsumer 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 <noreply@anthropic.com>
Contributor
|
🌟 Thank you for your contribution to the Apache Camel project! 🌟 🐫 Apache Camel Committers, please review the following items:
|
Contributor
|
🧪 CI tested the following changed modules:
🔬 Scalpel shadow comparison — Scalpel: 476 tested, 29 compile-only — current: 476 all testedMaveniverse Scalpel detected 505 affected modules (current approach: 476).
|
gnodet
marked this pull request as ready for review
August 3, 2026 20:36
gnodet
commented
Aug 3, 2026
gnodet
left a comment
gnodet
left a comment
Contributor
Author
There was a problem hiding this comment.
Correct root cause fix — the coordinator thread count is 1 but the latch was initialized with concurrentConsumers (e.g. 2), causing shutdown to always wait the full timeout. The fix works and the test is well-designed. Two minor suggestions below.
This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.
Claude Code on behalf of @gnodet
|
|
||
| private final AtomicInteger taskCount = new AtomicInteger(); | ||
| private volatile CountDownLatch latch; | ||
| protected volatile CountDownLatch latch; |
Contributor
Author
There was a problem hiding this comment.
Design nit: Changing private → protected on the latch field exposes mutable internal state to all subclasses. A template method would achieve the same goal without field exposure:
// In SedaConsumer:
protected int getLatchCount() {
return getEndpoint().getConcurrentConsumers();
}
// In doStart():
int count = getLatchCount();
if (count > 0) {
latch = new CountDownLatch(count);
}Then ThreadPerTaskSedaConsumer just overrides getLatchCount() to return 1 — no need to touch latch directly, and it eliminates the "create then immediately replace" pattern where super.doStart() allocates a CountDownLatch(N) that is discarded one line later.
Non-blocking — the current approach is functionally correct.
| // 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. " |
Contributor
Author
There was a problem hiding this comment.
Convention nit: Per project conventions, new test code should use AssertJ assertions. Consider:
assertThat(elapsed)
.as("Context stop took %ds; CountDownLatch count likely mismatches coordinator thread count", elapsed)
.isLessThan(30L);This provides better failure messages and aligns with the project's preference for AssertJ over JUnit assertions.
oscerd
approved these changes
Aug 4, 2026
davsclaus
approved these changes
Aug 4, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Claude Code on behalf of gnodet
Fix the root cause of the SEDA virtual thread shutdown hang (CAMEL-23129). The previous fix added a timeout to
latch.await()as a defensive workaround, but the underlying mismatch remained.Root cause:
SedaConsumer.doStart()createslatch = new CountDownLatch(concurrentConsumers), butThreadPerTaskSedaConsumeruses a single coordinator thread. WhenconcurrentConsumers > 1(used as a concurrency limit, e.g., 2), the latch is initialized with count=2 but only 1 coordinator thread ever callscountDown()— soprepareShutdown()always waits the full shutdown timeout before proceeding.Fix: Override
doStart()inThreadPerTaskSedaConsumerto setlatch = new CountDownLatch(1), matching the single coordinator thread. TheconcurrentConsumersvalue continues to serve as the task executor concurrency limit via theSemaphore. The task executor shutdown is handled separately inThreadPerTaskSedaConsumer.prepareShutdown().Changes
SedaConsumer: Changelatchfield visibility fromprivatetoprotectedso subclasses can override the countThreadPerTaskSedaConsumer: OverridedoStart()to set latch count to 1ThreadPerTaskSedaConsumerTest: Add test verifying shutdown with concurrency limit completes quickly (no full timeout wait); apply JUnit 5 conventions (droppublic)Test plan
ThreadPerTaskSedaConsumerTesttests pass (virtualThreadPerTask basic, concurrency limit, high throughput)testShutdownWithConcurrencyLimitCompletesQuicklyverifies context stop completes in < 30s withconcurrentConsumers=2