Repository navigation
CAMEL-25837: camel-hazelcast - the queue consumer in poll mode stops without losing items - #28258
Conversation
…without losing items In poll mode the hazelcast-queue consumer stopped by interrupting its polling thread (shutdownNow), and the item of a poll is removed from the queue when the poll returns it: - an interrupted poll keeps waiting on the member that owns the queue, until its pollingTimeout (10 seconds by default). The next item offered in that time is taken by that poll and nobody routes it. This happens at every stop of an idle route (route stop, restart, master failover, application shutdown); - an exchange that was being routed was interrupted, and the processor chain was stopped before the exchange completed, so a graceful shutdown did not let it complete: its item was lost. The consumer now lets its thread end on its own: it still waits up to pollingTimeout for an item, but in IQueue.poll calls of at most one second, so it sees a stop within a second; doStop waits for the current poll and exchange to complete (a graceful executor shutdown) and stops the processor afterwards. 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 analysis holds for Hazelcast 5.4. On a client, ClientInvocationFuture.onInterruptDetected ends the caller's wait while the PollOperation stays parked on the member. On an embedded member it is even worse: AbstractInvocationFuture.get() ignores the interrupt and keeps parking until the poll completes. So not interrupting, waiting for the thread, and stopping the processor afterwards (moving super.doStop() to the end) is the right fix. The listen mode is unchanged. This does not overlap with #28261 (SEDA) or #28264 (topic). LGTM once the build jobs are green.
- Optional:
pollItem()can be simpler. When it returns null, thewhile (isRunAllowed())loop inrun()just calls it again, so the deadline loop does the same as a singlequeue.poll(Math.min(config.getPollingTimeout(), MAX_POLL_WAIT), TimeUnit.MILLISECONDS). Either way,pollingTimeoutnow has no visible effect on polling (it only sets the delay after a poll error). That's fine, but it means the deadline code adds nothing. - Not blocking: a stop now waits for the exchange in flight (up to
shutdownAwaitTermination) instead of interrupting it. That is the point of the fix and an improvement. #28261 takes the same approach for the SEDA consumer but still polls for the fullpollTimeout. I suggested the same 1 s cap there, so both consumers behave the same. - Optional: see the inline comment on the test helper.
Claude Code on behalf of davsclaus. This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.
…meout, and count only this context's consumer threads in the test Per review: - pollItem() is a single IQueue.poll(Math.min(pollingTimeout, MAX_POLL_WAIT)): a poll that returns nothing goes round the loop in run() again, so the deadline loop added nothing. - HazelcastQueueConsumerPollStopTest counts only the QueueConsumer threads of this CamelContext, so a thread left by another test class in the same fork does not make the wait time out. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
|
Thanks Claus. Pushed 44fe4de:
camel-hazelcast: 245 tests, 0 failures, 1 error ( Claude Code on behalf of allthingssecurity |
|
🧪 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, 5m 21s total)Total reactor time: 5m 21s
Top 20 slowest modules:
|
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
CAMEL-25837: Hazelcast queue consumer shutdown fix ✅
This is a well-designed fix for a real data-loss bug in the Hazelcast queue consumer's poll mode. Verified locally — tests fail before the fix and pass after.
What the fix does
-
stopLatch.countDown()before executor shutdown — signals the consumer task to exit cleanly before the executor is torn down, preventing the task from being forcibly interrupted mid-exchange. -
shutdownNow()→shutdownGraceful(executor)/executor.shutdown()— replaces abrupt termination with graceful shutdown. The defaultshutdownAwaitTerminationof 10 s comfortably covers the worst case: ≤1 s poll (capped byMAX_POLL_WAIT) + in-flight exchange processing time. -
MAX_POLL_WAIT = 1000L+Math.min(config.getPollingTimeout(), MAX_POLL_WAIT)— caps single poll waits at 1 s. This ensures the consumer thread sees the stop signal within 1 s even if the configuredpollingTimeoutis large, without losing any queued items. -
super.doStop()moved to the end — correct ordering: signal → drain → delegate to parent.
Static analysis
broad-exception-catch in the catch (Exception e) block is pre-existing, unchanged code — not introduced by this PR.
Companion PR
#28261 applies the same fix to the SEDA consumer — not a duplicate, a necessary companion.
No issues found.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
Thanks Shashank. 44fe4de addresses both points:
pollItem()is now a singlepoll(Math.min(pollingTimeout, MAX_POLL_WAIT)). It behaves the same as the deadline loop: a null goes roundrun()again, and a stop is seen within 1 s.pollingTimeoutof 0 or less behaves as before the PR.- The test counts only threads named
(<context>)andQueueConsumer. The closing paren keepscamel-1from matchingcamel-10, and the threads of an earlier test in the class are gone because its stop waits for them. The count of 3 is awaited, so the short gaps between 1 s polls do not make it flaky.
LGTM, to merge once CI is 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-25837
With
queueConsumerMode=Pollthe hazelcast-queue consumer takes items withIQueue.poll(pollingTimeout)in its own thread, and a poll removes the item from the queue.doStopstopped that thread withshutdownNow, which interrupts it:pollingTimeout, 10 s by default) and takes the first item offered in that time, which nobody routes. This happens at every stop of the route (stop, restart, failover, shutdown) while producers keep offering.super.doStop()) before the thread was done and the interrupt ends any blocking call of the route, so the exchange failed silently and its item was lost, also with a graceful shutdown.Changes (
HazelcastQueueConsumer):IQueue.pollnow waits at most one second (Math.min(pollingTimeout, 1000)); a poll that returns nothing goes round the loop again, so the consumer still waits for items as long as it runs, and sees a stop within a second. An idle consumer makes one poll call per second instead of one perpollingTimeout, andpollingTimeoutnow only bounds a single poll (and sets the delay after a poll error).doStopno longer interrupts the thread: it waits for the current poll and exchange (graceful executor shutdown, bounded byshutdownAwaitTermination) and stops the processor afterwards.Without the one-second calls, the graceful wait would last up to
pollingTimeout, which is 10 s by default likeshutdownAwaitTermination, so the stop would time out and interrupt the poll again.The listen mode (the default) is unchanged. No option changed, so no upgrade guide entry.
Tests:
HazelcastQueueConsumerPollStopTest(3 tests, an embedded Hazelcast member). Without the change all three fail: the item offered after the stop is taken from the queue (expected: <1> but was: <0>, initemOfferedAfterTheRouteStoppedStaysInTheQueueandlongPollingTimeoutWaitsForAnItemAndStillStopsPromptly), and the exchange in flight never reaches the mock (Expected: <1> but was: <0>). With the change all three pass.longPollingTimeoutWaitsForAnItemAndStillStopsPromptlyusespollingTimeout=60000: an item offered after 3 s of waiting is routed, the route stops in less than 5 s, and an item offered afterwards stays in the queue. With a singlepoll(pollingTimeout)and the graceful stop, it fails (Stopping the route took 10004 ms).HazelcastUserConfigSerializationFilterWarningTest.warnsForUserClientConfigWithoutFilter: a client that cannot reach a cluster here; it errors the same way without the change).How it was found: variant analysis of CAMEL-25790 (a consumer with its own loop and executor, and what its stop does to that loop), 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