Repository navigation
CAMEL-25840: camel-hazelcast - the SEDA consumer waits for its threads before it stops the processor - #28261
Conversation
…s before it stops the processor HazelcastSedaConsumer.doStop shut its executor down without waiting and then stopped the processor, while its threads could still be waiting in poll(pollTimeout) (1 second by default). A poll takes the item from the queue: an item offered in that time was taken, its exchange failed with RouteStoppingException and, without transacted, the item was lost. doStop now waits for the consumer threads to end their current poll and exchange (a graceful executor shutdown) before it stops the processor, and shuts the error-delay scheduler down after them, since the threads use it: a thread in the delay after a failed poll waits for a task of that scheduler, which a shutdown cancels, so it was left waiting for ever when the scheduler was shut down first. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
|
🌟 Thank you for your contribution to the Apache Camel project! 🌟 🐫 Apache Camel Committers, please review the following items:
|
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
Thanks Shashank,
The new order is right. BackgroundTask.run uses scheduleWithFixedDelay and then waits on latch.await() with no limit (withUnlimitedDuration()). ScheduledThreadPoolExecutor.shutdown() cancels periodic tasks, so on main a thread in the error delay was left waiting for ever. With the scheduler shut down after the consumer threads, the delay ends and the thread sees the stop. The change only touches doStop, so it merges cleanly with CAMEL-25816 (d4688eb). It does not overlap with #28258 or #28264. LGTM once the build jobs are green.
- Not blocking: the graceful wait only helps while
pollTimeoutis belowshutdownAwaitTermination(10 s by default). With a largerpollTimeout, every stop now waits 10 s, logsForcing shutdown of ExecutorServiceat WARN and then interrupts the poll. In Hazelcast 5.4 the interrupt does not fix it. On a client,ClientInvocationFuture.onInterruptDetectedends the wait but the poll stays parked on the member and takes the next item. On an embedded member,AbstractInvocationFuture.get()ignores the interrupt and parks until the poll completes, so the item is polled and then routed into a stopped processor. Either way the item is lost again, after up to 2 x 10 s. #28258 avoids this for the queue consumer by polling at most 1 s at a time. Here it is even simpler: a null poll just goes back round the loop (and commits the empty transaction), soqueue.poll(Math.min(pollTimeout, 1000), TimeUnit.MILLISECONDS)keeps the behaviour and bounds the stop to about 1 s. During a CamelContext shutdown,DefaultShutdownStrategystops the consumers one route after the other, so that bound adds up over many routes. It would also make the two consumers behave the same on stop. - Not blocking: a thread in the error delay still waits the full
onErrorDelaybefore it sees the stop (1 s by default, so fine). The queue consumer ends that delay at once with itsstopLatch. Worth doing the same here if you touch this again, but not needed for this PR.
Claude Code on behalf of davsclaus. This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.
… at one second Per review: the graceful wait in doStop only helped while pollTimeout was below shutdownAwaitTermination; with a larger pollTimeout every stop waited 10 s and then interrupted the poll, which in Hazelcast 5.4 does not end the poll on the member. A single poll now waits Math.min(pollTimeout, 1000) ms (as the queue consumer does since CAMEL-25837); a poll that returns nothing goes round the loop and commits the empty transaction as before. New test longPollTimeoutStillStopsPromptlyAndLeavesTheItemInTheQueue (pollTimeout=60000). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
|
Thanks Claus. Pushed ee5eb3f:
camel-hazelcast: 245 tests, 0 failures, 1 error ( Claude Code on behalf of allthingssecurity |
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Fix is correct and well-tested. The ordering in doStop() is right: shutdownGraceful(executor) waits for consumer threads to finish their current poll (capped at MAX_POLL_WAIT = 1 s, so threads return quickly), then taskExecutor is shut down (safe, since threads no longer need it), then super.doStop() stops the processor. Without the fix, the processor was stopped before the threads — a thread returning from a poll with an item in hand would hit a stopped processor and lose the item.
The shutdownGraceful semantics are sound: Camel's BaseExecutorServiceManager calls executor.shutdown() (orderly — no interruption), then awaitTermination for up to 10 s. Since the poll is now capped at 1 s, all threads return within 1 s, see isRunAllowed() == false (the service status is already STOPPING when doStop() runs), and exit the loop. The 10 s graceful timeout is more than enough.
Pre-fix test validation (static):
itemOfferedAfterTheRouteStoppedStaysInTheQueue: waits for the consumer to be inpoll(), stops the route, offers an item, asserts it stays in the queue. Without the fix, the consumer thread returns from poll, gets the item, and calls the stopped processor →RouteStoppingException→ item silently discarded. ✅ Correctly exercises the fixed path.longPollTimeoutStillStopsPromptlyAndLeavesTheItemInTheQueue: verifies theMath.min(pollTimeout, MAX_POLL_WAIT)cap — without it, a 60 spollTimeoutwould cause the stop to wait for the graceful timeout (10 s) before forcing shutdown. ✅ Correct.threadInTheDelayAfterAFailedPollEndsWhenTheRouteStops: verifies thetaskExecutoris shut down AFTER the consumer threads (not before) — without the fix,taskExecutor.shutdown()cancelled the latch task the consumer thread was waiting on, leaving it blocked forever. ✅ Correct.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
|
🧪 CI tested the following changed modules:
🔁 1 test passed only after a retry on JDK 25 (1 retried attempt) Recovered flaky tests on JDK 25 (1)
🔬 Scalpel shadow comparison — Scalpel: 8 of 707 tested, 27 compile-only — current: 8 all testedMaveniverse Scalpel detected 8 affected modules (current approach: 8). Skip-tests mode would test 8 modules (1 direct + 7 downstream), skip tests for 27 (generated code, meta-modules) Modules Scalpel would test (8)
Modules with tests skipped (27)
All tested modules (35 modules, 4m 53s total)Total reactor time: 4m 53s
Top 20 slowest modules:
|
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
Thanks Shashank,
The new commit looks good. A single poll now waits at most MAX_POLL_WAIT (1 s), so shutdownGraceful no longer runs into its 10 s timeout and the forced interrupt when pollTimeout is large. The default pollTimeout is already 1000, so the default behaviour does not change. A null poll goes back round the loop as before. The new test longPollTimeoutStillStopsPromptlyAndLeavesTheItemInTheQueue covers the capped case, and the queue and SEDA consumers now stop the same way. LGTM once the build jobs are green.
Claude Code on behalf of @davsclaus. This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.
Description
CAMEL-25840
HazelcastSedaConsumer.doStopshut its executor down without waiting and then stopped the processor, while its threads usually still waited inqueue.poll(pollTimeout)(1 s by default). A poll takes the item from the queue: an item offered in that window was taken by the stopping consumer, its exchange failed withRouteStoppingException, and withouttransacted(the default) the item was lost, with one WARN. On a busy queue this happens at nearly every stop of the route (stop, restart, failover, shutdown), up to one item per consumer thread.Changes (
HazelcastSedaConsumer.doStop):shutdownAwaitTermination), then stop the processor.Math.min(pollTimeout, 1000), as CAMEL-25837: camel-hazelcast - the queue consumer in poll mode stops without losing items #28258 does for the queue consumer); a poll that returns nothing goes round the loop again (and commits the empty transaction) as before. Without it the graceful wait would only help whilepollTimeoutis belowshutdownAwaitTermination(10 s): with a largerpollTimeoutevery stop would wait 10 s, logForcing shutdown of ExecutorServiceand interrupt the poll, which in Hazelcast 5.4 does not end the poll on the member (review follow-up).BackgroundTask.run, a latch) for a task of that scheduler, and shutting the scheduler down cancels that task. On main the scheduler was shut down first, so such a thread was left waiting for ever; with the graceful wait it would also hold the stop until the 10 s timeout.A stop takes at most about one second longer. No option changed, so no upgrade guide entry. Besides
doStop, only the timeout of the poll call changes; the branch merges cleanly with the change of CAMEL-25816 to the loop of the same class (checked withgit merge-treeagainst main).Tests:
HazelcastSedaConsumerStopTest(3 tests; an embedded Hazelcast member, and a mocked instance for the failing queue):itemOfferedAfterTheRouteStoppedStaysInTheQueuestops the idle route while it polls, then offer an item. Without the change the item is taken from the queue (expected: <1> but was: <0>); with the change it stays and the next start routes it.longPollTimeoutStillStopsPromptlyAndLeavesTheItemInTheQueue(same class,pollTimeout=60000): an item is routed, the route stops in less than 5 s, and an item offered afterwards stays in the queue. Without the 1 s cap it fails:Stopping the route took 10002 ms(the graceful timeout).threadInTheDelayAfterAFailedPollEndsWhenTheRouteStops(same class): a route on a queue whose every poll fails (onErrorDelay=3000) is stopped while its thread waits in the error delay; the stop takes less than 8 s and the thread ends. Without the change the thread is left (A consumer thread is left: [... hazelcast-seda://failing ...]).HazelcastUserConfigSerializationFilterWarningTest.warnsForUserClientConfigWithoutFilter: a client that cannot reach a cluster here; it errors the same way without the change), 1 flake (HazelcastAggregationRepositoryRoutesTest.checkAggregationFromTwoRoutesfailed once and passed on the rerun; it does not use the SEDA consumer).How it was found: variant analysis of CAMEL-25790 (a consumer with its own loops, and what its stop does to them), next to CAMEL-25837, then the JUnit test.
What was not run: a Hazelcast client (only an embedded member); the full root build.
Target
mainbranch)Tracking
Apache Camel coding standards and style
mvn clean install -DskipTestslocally from root folder and I have committed all auto-generated changes.(I built and tested camel-hazelcast. No option and no component page changed. I did not run the full root build.)
AI-assisted contributions
Co-authored-bytrailers) and the PR description identifies the AI tool used.This PR was prepared with Claude Code (Claude Opus 5.5). The commit carries a
Co-Authored-Bytrailer.Claude Code on behalf of allthingssecurity
🤖 Generated with Claude Code