Repository navigation
CAMEL-25843: camel-hazelcast - the reliable topic consumer goes on after it lost messages - #28264
Conversation
|
🌟 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,
I checked this against Hazelcast 5.4.0 (the version we use). ReliableTopicProxy and ClientReliableTopicProxy use the listener as is when it implements ReliableMessageListener, and wrap a plain MessageListener in ReliableMessageListenerAdapter, whose isLossTolerant() is false. MessageRunner.accept then cancels it on the first gap. For a loss tolerant listener Hazelcast logs the gap only at FINEST (MessageRunner.isLossTolerable), so the WARN from storeSequence is the only thing a user sees. ReadManyOperation clamps a stale sequence to the head, so the runner reads on and no StaleSequenceException is thrown, on a member or a client. retrieveInitialSequence() = -1 and isTerminal() = false match what the adapter did. The code is good; one thing is needed before merge (point 1).
- Blocking: this changes what users see. Before, a consumer that fell behind stopped receiving for good. Now it skips the lost messages and goes on. Please add a short note to
camel-4x-upgrade-guide-4_24.adoc(it already has several camel-hazelcast entries), and a sentence on the hazelcast-topic page next to thereliableoption. Today the page does not say what happens when a reliable listener falls behind. - Optional: see the inline comment on
onCancel(). Hazelcast can still cancel the listener for other reasons, and then the route also stays started without receiving anything. - Optional: the WARN only says "overwritten". Messages can also be lost because they expired (ringbuffer time-to-live), as the class javadoc says.
Claude Code on behalf of davsclaus. This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.
…ter it lost messages A reliable topic keeps its messages in a ringbuffer. When a listener falls behind by more than the ringbuffer holds (its capacity, 10000 by default, or its time to live), the messages it did not read yet are overwritten, and Hazelcast removes a listener that is not loss tolerant for good. The consumer registered a plain MessageListener, which Hazelcast treats as not loss tolerant: after one burst that a slow route could not keep up with, the route stayed started and received nothing again, and nothing was logged by Camel. The reliable topic consumer now registers a ReliableMessageListener that is loss tolerant: it logs a WARN with the number of messages lost and goes on with the oldest message still in the ringbuffer. It starts with the messages published from then on, as before. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
… topic consumer, and report a listener that Hazelcast cancels Per review: - Upgrade guide 4.24: a reliable topic consumer that falls behind now skips the lost messages and goes on (next to the other camel-hazelcast entries); a paragraph on the hazelcast-topic page about reliable=true (catalog copy updated). - CamelReliableMessageListener.onCancel(): when Hazelcast cancels the listener for another reason (an error reading the ringbuffer, a destroyed topic, an instance no longer active) while the consumer is started, report it through the consumer's exception handler; a normal stop reports nothing. - The WARN says the lost messages were overwritten or expired. - Tests: a destroyed topic is reported once; a route stop is not reported. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
7a9738d to
a6de186
Compare
|
Thanks Claus. I rebased on main (the 4.24 guide had moved on; the existing commit is unchanged) and pushed a6de186 on top:
camel-hazelcast: 249 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.
Solid fix. The ReliableMessageListener implementation is correct, the key behaviors are well-covered by tests, and all three of @davsclaus's review points are addressed.
Implementation review:
isLossTolerant() = truedirectly fixes the root cause: Hazelcast'sReliableMessageListenerAdapterwraps a plainMessageListenerwithisLossTolerant()false, which causesMessageRunner.accept()to cancel the listener on the first gap. The new listener avoids this.storeSequence()gap arithmetic is correct per the Hazelcast contract (sequences are monotonically increasing;storeSequenceis called in order by a single thread per listener —lastSequencedoes not need to bevolatile).onCancel()guard viaisStarted()is correct:BaseService.stop()transitions status toSTOPPINGbefore callingdoStop(), which callsremoveMessageListener(), which triggersonCancel(). At that pointisStarted()returnsfalse(STOPPING ≠ STARTED). No false alarm on normal route stop.isTerminal() = falseis correct:CamelListener.sendExchange()wrapsprocessor.process()and routes exceptions to the exception handler — route failures never reach Hazelcast asisTerminalinputs.
Tests: consumerGoesOnAfterFallingBehindTheRingbuffer simulates the exact production failure (capacity 10 ringbuffer, 30-message burst, deterministic blocking via CountDownLatch). The other two tests cover the onCancel behavior cleanly. Pre-fix behavior: without CamelReliableMessageListener, the mock would never receive "after" (Hazelcast removes the plain listener on the gap) and listenerCancelledByHazelcastIsReported would timeout (no onCancel override). Tests correctly validate the fix.
LGTM.
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, all three points are addressed:
- The upgrade guide entry in 4_24 and the paragraph on the hazelcast-topic page (with the catalog copy) read well.
onCancel(): I checked it against Hazelcast 5.4.0.removeMessageListenercancels the runner in the calling thread, so theisStarted()check skips the normal stop. A destroyed topic is reported once, because the runner has already leftrunnersMapwhen the route stops later. The two tests cover both cases.- The WARN and the javadoc now say "overwritten or expired".
One optional nit inline about shutdown noise. Approving, 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.
…r that Hazelcast cancels while Hazelcast or Camel shuts down Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
Thanks Shashank, the shutdown nit is addressed. onCancel() now logs at DEBUG when the instance's lifecycle service is not running or the CamelContext is stopping, and a destroyed topic is still reported. I checked Hazelcast 5.4.0: HazelcastInstanceProxy and HazelcastClientProxy return a TerminatedLifecycleService from getLifecycleService() once the instance is gone, so the check cannot throw. The new test and the reworded upgrade guide entry look good.
Approving, 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.
|
🧪 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, 25 compile-only — current: 8 all testedMaveniverse Scalpel detected 8 affected modules (current approach: 8). Skip-tests mode would test 8 modules (3 direct + 7 downstream), skip tests for 25 (generated code, meta-modules) Modules Scalpel would test (8)
Modules with tests skipped (25)
All tested modules (35 modules, 4m 23s total)Total reactor time: 4m 23s
Top 20 slowest modules:
|
Description
CAMEL-25843
With
reliable=truethe hazelcast-topic consumer listened with a plainMessageListener. A reliable topic keeps its messages in a ringbuffer of limited capacity; when a listener falls behind by more than that, Hazelcast removes a listener that is not loss tolerant for good (MessageRunner.acceptcancels it;ReliableMessageListenerAdapter.isLossTolerant()is false). After one burst that a slow route could not keep up with, the route stayed started and received nothing again, and Camel logged nothing.Changes:
CamelReliableMessageListener(aReliableMessageListener, loss tolerant): logs a WARN with the number of messages lost (overwritten or expired in the ringbuffer) and goes on with the oldest message still in the ringbuffer; starts with the messages published from then on, as before.MessageRunner.cancel()callsonCancel(): an error reading the ringbuffer, such as a destroyed topic), the listener reports it through the consumer's exception handler (WARN by default) while the consumer is started. A normal stop removes the listener while the consumer is stopping, which reports nothing. A cancel while the Hazelcast instance is shutting down (its lifecycle service is no longer running; the runner fails withHazelcastInstanceNotActiveException) or while the CamelContext is stopping is logged at DEBUG only: Hazelcast's JVM shutdown hook runs at the same time as Camel's, so on an ordinary shutdown the member can stop while the route is still started (review follow-up 9224021).HazelcastTopicConsumeruses it for a reliable topic. The plain topic is unchanged.No option changed. Users can notice the change (a consumer that fell behind used to stop receiving; now it skips the lost messages and goes on), so there is an entry in the 4.24 upgrade guide, next to the other camel-hazelcast entries, and a paragraph on the hazelcast-topic page about
reliable=true(catalog copy updated).Tests:
HazelcastReliableTopicConsumerLossTest(embedded member, ringbuffer capacity 10): the route is busy with a first message while 30 more are published, then one more is published. Without the change it is never received (Expected: <1> but was: <0>); with the change it is, and a WARN reports the messages that were overwritten (about 20; the exact number depends on how far the ringbuffer has moved when the listener reads again).listenerCancelledByHazelcastIsReporteddestroys the topic while the route runs: Hazelcast cancels the listener and the consumer's exception handler gets one report.listenerRemovedWhenTheConsumerStopsIsNotReportedstops the route: no report. WithoutonCancel()nothing reports the cancelled listener, so the first one cannot pass; the second is a control.listenerCancelledWhenTheMemberShutsDownIsNotReportedshuts the member down while the route is started and checks for 1 s that nothing is reported. Without the shutdown check it fails (the exception handler getsHazelcast cancelled the listener of the reliable topic news).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-25815 (the server side ends a consumer's subscription and the consumer never subscribes again), 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