Repository navigation
CAMEL-25847: camel-nats - a stopped JetStream push consumer unsubscribes through its dispatcher - #28320
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, good catch. On main, doStop cannot end a push subscription: dispatcher.unsubscribe(topic) misses it because it is on the deliver inbox, and jetStreamSubscription.unsubscribe() is refused by jnats (NatsSubscription.unsubscribe throws "Subscriptions that belong to a dispatcher cannot respond to unsubscribe directly"). With a shared connection, the stopped route therefore keeps getting messages. Unsubscribing through the dispatcher and closing it fixes that. Deferring the close until the in-flight message is done is a nice touch, because jnats NatsDispatcher.stop cancels its thread with Future.cancel(true), and that would interrupt the exchange. Both tests fail without the fix.
TL;DR
LGTM once CI is green (still running). One optional suggestion.
Blockers: None
Other findings: 1
Non-blocking
- The in-flight counter is per consumer, but there is one dispatcher per run.
messagesInFlightandcloseDispatcherWhenIdlebelong to the consumer, while every start creates a new dispatcher and a newCamelNatsMessageHandler. Restarting in place is supported, so suppose the route is started again while an exchange of the previous run is still in flight (only possible after the shutdown timeout forced the stop). The old dispatcher's close then waits until the new run is idle too, and a second stop before that overwritescloseDispatcherWhenIdle, so the first dispatcher is never closed and its thread stays parked on the connection's executor. Keeping the counter and the pending close on the handler instance (one per dispatcher), and keeping a reference to that handler next tothis.dispatcher, would remove the coupling. See the inline comment. This is a narrow case, so I'm fine with a follow-up.
Notes
- This also closes the dispatcher of a core NATS consumer (non-JetStream) on stop with a shared connection, which was leaked before. That's a good side effect; maybe say so in the PR description.
- No conflict with #28318: the two merge cleanly, and #28318's stop-race test already allows for the closed dispatcher. They also merge cleanly with #28204.
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:
🔬 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 12s total)Total reactor time: 4m 12s
Top 20 slowest modules:
|
… not per consumer Review of apache#28320 (optional point 1): the in-flight count and the pending close of the dispatcher were fields of the consumer, while each start makes a new dispatcher and a new CamelNatsMessageHandler. A consumer that is stopped and started again in place (as a route policy such as a throttling policy does with a consumer that is not Suspendable) while an exchange of the previous run is still in flight closed the old dispatcher only once the new run was idle too, and a second stop before that overwrote the pending close, so the first dispatcher was never closed. The count and the pending close now live on the CamelNatsMessageHandler, one per dispatcher. The consumer keeps the handler next to its dispatcher, and doStop asks that handler to close the dispatcher now or once its message in flight is done. Test: eachRunOfTheConsumerClosesItsOwnDispatcher (stop, start and stop the same consumer while a message of the first run is in flight) fails without the change: the dispatcher of the second run is not closed when it stops. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
… once the stop closes the dispatcher removesItsConnectionListenerWhenTheRouteStopsWhileItSubscribes waited for the keeper to register its listener. Merged with CAMEL-25847 (apache#28320), the stop closes the dispatcher, jnats refuses the subscribe on it and no keeper is started, so the wait timed out. The test now waits for either the keeper's listener or the refused subscribe, and then checks that no keeper listens to the connection. Checked on a local merge of both branches: camel-nats 21 tests, 0 failures (the previous version of this test failed there). On this branch alone it still fails without the keeper guard of the previous commit. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
|
@davsclaus thanks for the review. The optional point is done in 15aa72e: the in-flight state now lives on the per-dispatcher Tests (offline, Claude Code on behalf of allthingssecurity |
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
Thanks, that is what I had in mind. The in-flight count and the pending close now live on CamelNatsMessageHandler (NatsConsumer.java:363-379), and doStop asks this run's handler to close its own dispatcher (lines 126-130), so a second stop can no longer overwrite the first run's close. eachRunOfTheConsumerClosesItsOwnDispatcher models the in-place restart well.
One non-blocking point (from the first commit): when the route is stopped while the task is still in jetStream().subscribe(...), doStop closes the dispatcher the task already created, jnats refuses the subscribe with "Dispatcher is closed", and the catch-all in NatsConsumingTask.run (line 213) reports it as "Error during processing". A route stopped right after it starts would log a WARN it did not log before. Logging at DEBUG there when !isRunAllowed(), like the keeper of #28318 does in consumerDeleted, would keep the stop quiet.
LGTM once CI is green.
Reviewed with Claude Code on behalf of davsclaus. This review was generated by an AI agent and may contain inaccuracies; please verify all suggestions before applying.
|
@davsclaus done in 3fe5f7b. The catch-all in New test Tests (offline, The guard checks the consumer, not the run. If the same consumer is stopped and started in place while the first run's subscribe is still in progress, that refused subscribe is still logged at WARN. This is the same gap as your nit on #28318, and it is in the same follow-up draft. Claude Code on behalf of allthingssecurity |
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
The fix is well-scoped and correct. The in-flight tracking per-handler (CamelNatsMessageHandler) is the right design — one counter per handler avoids shared-state races when multiple handlers coexist on the same Connection.
One minor nit on closeWhenIdle:
This review was generated by an AI agent, Hermès on behalf of @gnodet.
… once the stop closes the dispatcher removesItsConnectionListenerWhenTheRouteStopsWhileItSubscribes waited for the keeper to register its listener. Merged with CAMEL-25847 (apache#28320), the stop closes the dispatcher, jnats refuses the subscribe on it and no keeper is started, so the wait timed out. The test now waits for either the keeper's listener or the refused subscribe, and then checks that no keeper listens to the connection. Checked on a local merge of both branches: camel-nats 21 tests, 0 failures (the previous version of this test failed there). On this branch alone it still fails without the keeper guard of the previous commit. 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.
TL;DR
The new commit is correct: a stop while the consumer subscribes is now logged at DEBUG, and real errors still go to the exception handler.
Blockers: None
Other findings: 0
Thanks @allthingssecurity, this covers the point from the earlier review.
NatsConsumer.java:213-222: the catch inNatsConsumingTask.runlogs at DEBUG only whenisRunAllowed()is false.isRunAllowed()is true from STARTING to SUSPENDED, so an error while the consumer starts or runs is still reported as before. The guard also covers the other ways a stop can break the setup: jnats refuses anulldispatcher afterdoStopcleared the field, and an owned connection is closed.NatsConsumerJetStreamPushStopTest.java:179-211: stopping the consumer from inside the fake subscribe hits the exact window. CI ran it, and all 4 tests pass.- CI for this commit ran before #28204 (CAMEL-25815) changed
NatsConsumer.javaon main. The file merges cleanly with today's main, and with #28318 in either order, with the same result. I also built camel-nats locally with today's main, #28318 and this PR merged together: 33 unit tests and 31 integration tests pass, including the new tests of all three PRs. - The open nit from the automated review (a comment before
close.run()at line 385) is optional.
Reviewed with Claude Code on behalf of @davsclaus. This review was generated by an AI agent and may contain inaccuracies; please verify all suggestions before applying.
… not per consumer Review of apache#28320 (optional point 1): the in-flight count and the pending close of the dispatcher were fields of the consumer, while each start makes a new dispatcher and a new CamelNatsMessageHandler. A consumer that is stopped and started again in place (as a route policy such as a throttling policy does with a consumer that is not Suspendable) while an exchange of the previous run is still in flight closed the old dispatcher only once the new run was idle too, and a second stop before that overwrote the pending close, so the first dispatcher was never closed. The count and the pending close now live on the CamelNatsMessageHandler, one per dispatcher. The consumer keeps the handler next to its dispatcher, and doStop asks that handler to close the dispatcher now or once its message in flight is done. Test: eachRunOfTheConsumerClosesItsOwnDispatcher (stop, start and stop the same consumer while a message of the first run is in flight) fails without the change: the dispatcher of the second run is not closed when it stops. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
3fe5f7b to
fbbb4f1
Compare
|
@davsclaus #28318 is merged, so as you asked this one is rebased onto main on top of it (fbbb4f1; the three commits applied without conflict) to get a fresh CI run. camel-nats unit tests on the rebased branch: 33 run, 0 failures, including the pull tests of #28204 and the push tests of #28318. Claude Code on behalf of allthingssecurity |
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Following up on my earlier review — the in-flight tracking nit remains open.
The concurrency design is correct: closeWhenIdle is called after the dispatcher is unsubscribed, so no new messages can arrive between the lock release and close.run() — the apparent window is not a real race. The two new tests (stoppedRouteGetsNoMessagesAndRestartedRouteGetsEachMessageOnce and stoppingTheRouteDoesNotInterruptTheExchangeInFlight) correctly validate both the unsubscription and the deferred-close scenarios. The rebase onto main (after #28318 merged) is clean.
One low-priority nit from the earlier review still stands — see inline.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
…bes through its dispatcher
A JetStream push subscription belongs to the dispatcher of the consumer.
On stop the consumer unsubscribed the topic from the dispatcher (the push
subscription is on its deliver subject, not on the topic) and then called
unsubscribe() on the subscription, which jnats refuses for a subscription
of a dispatcher ("cannot respond to unsubscribe directly"; only a WARN).
With a shared connection (connection option), which stays open, the
subscription stayed active: the stopped route kept getting messages,
which failed and were nak'ed and delivered again at once, and after the
route was started again every message was routed twice (the old and the
new subscription). The dispatcher and its thread were also never closed.
The consumer now unsubscribes the push subscription through its
dispatcher when it stops. It also closes the dispatcher it created,
which ends its thread, once no message is in flight on it: jnats
closes a dispatcher by cancelling its thread with an interrupt, and
that thread runs the route, which Camel stops before it waits for the
exchanges in flight; when a message is in flight, the dispatcher is
closed when that message is done.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
… not per consumer Review of apache#28320 (optional point 1): the in-flight count and the pending close of the dispatcher were fields of the consumer, while each start makes a new dispatcher and a new CamelNatsMessageHandler. A consumer that is stopped and started again in place (as a route policy such as a throttling policy does with a consumer that is not Suspendable) while an exchange of the previous run is still in flight closed the old dispatcher only once the new run was idle too, and a second stop before that overwrote the pending close, so the first dispatcher was never closed. The count and the pending close now live on the CamelNatsMessageHandler, one per dispatcher. The consumer keeps the handler next to its dispatcher, and doStop asks that handler to close the dispatcher now or once its message in flight is done. Test: eachRunOfTheConsumerClosesItsOwnDispatcher (stop, start and stop the same consumer while a message of the first run is in flight) fails without the change: the dispatcher of the second run is not closed when it stops. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
… is not reported as an error
When the consumer stops while its task is still in jetStream().subscribe(...), the stop closes the
dispatcher that the task passes to the subscribe, and jnats refuses it ("Dispatcher is closed").
The task reported that to the exception handler as "Error during processing", which logs a WARN for
a route that is only being stopped. When the consumer is no longer allowed to run, the task now logs
it at DEBUG instead; otherwise it is still reported as before.
New test stopWhileSubscribingIsNotReportedAsAnError: the consumer stops from inside the subscribe of
the fake server; nothing must reach the exception handler. Without the change it fails with
"reported: [java.lang.IllegalStateException: Dispatcher is closed]".
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…lock Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Signed-off-by: Claus Ibsen <claus.ibsen@gmail.com>
e1cde59 to
799c2f9
Compare
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Previous findings are all addressed. LGTM.
The comment line explaining why close.run() is called outside the synchronized block (to avoid blocking while holding the lock) is exactly what was needed. The concurrency design is sound: the in-flight tracking per CamelNatsMessageHandler ensures each dispatcher run is independently managed, and the deferred close correctly avoids interrupting the exchange in flight. All four tests validate the fix end-to-end.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
🔄 Backport BotThis bugfix targets
Add a Backport PRs will be created automatically on merge (once a port label is added). Comment ℹ️ If you push additional commits after |
Description
CAMEL-25847
A JetStream push subscription (
jetstreamEnabled=true,pullSubscription=false) belongs to the dispatcher the consumer creates. On stop the consumer unsubscribed the topic from that dispatcher (the push subscription is on its deliver subject, not on the topic) and then calledunsubscribe()on the subscription, which jnats refuses for a subscription of a dispatcher ("cannot respond to unsubscribe directly"), so only a WARN was logged. The dispatcher was never closed.When the consumer owns its connection, closing the connection ends everything. With a shared connection (the
connectionoption) the subscription stayed active: the stopped route kept getting messages, which failed and were nak'ed and delivered again at once, and after the route was started again every message was routed twice. In plain NATS mode the dispatcher thread of every stopped consumer was left behind.Change (
NatsConsumer.doStop):dispatcher.unsubscribe(jetStreamSubscription), which jnats accepts for a subscription of that dispatcher). This does not touch the dispatcher thread.connection.closeDispatcher, while the connection is open), which ends its thread, once no message is in flight on it. jnats closes a dispatcher withNatsDispatcher.stop(true), which cancels its thread withFuture.cancel(true), an interrupt (javap, jnats 2.26.3), and that thread runs the route: Camel stops a consumer that is notSuspendablebefore it waits for the exchanges in flight, so closing it right away would interrupt the step in flight (a bean waiting on I/O, a delay). The message handler of the dispatcher (one per start) counts its messages in flight; when one is in flight at the stop, the dispatcher is closed when it is done (on that thread, after the message). The count is kept per dispatcher, not per consumer (review): a consumer that a route policy stops and starts in place while a message of the earlier run is in flight closes each dispatcher of its own run.NatsConsumingTask.runlogs at DEBUG, instead of reporting "Error during processing" to the exception handler (a WARN), when the consumer is no longer allowed to run (review). A stop while the task is still injetStream().subscribe(...)closes the dispatcher that the subscribe uses, and jnats refuses it ("Dispatcher is closed"). Any other error is reported as before.Note: Camel itself fails an exchange that is in flight when its consumer stops at its next step (
RouteStoppingException, CAMEL-25502:DefaultConsumer.doStopstops the route's processors), on main as well. What the change keeps is that the step in progress is not interrupted.Side effect: the dispatcher of a core NATS consumer (
jetstreamEnabled=false) is now closed when the consumer stops with a shared connection; before, it was left behind with its thread.What users can notice: stopping a route with a JetStream push consumer on a shared connection really stops it, and no WARN "Error during unsubscribing" is logged. With a shared connection, a stopped core NATS consumer no longer leaves its dispatcher thread behind. No option changed, so no upgrade guide entry.
Tests:
NatsConsumerJetStreamPushStopTestwith a fake JetStream server behind the jnats interfaces the consumer calls (like jnats, it refusesunsubscribe()on a subscription of a dispatcher, unsubscribes by subscription, and closing a dispatcher ends its subscriptions and interrupts the thread that delivers its messages; a closed dispatcher refuses to subscribe or unsubscribe).stoppedRouteGetsNoMessagesAndRestartedRouteGetsEachMessageOncefails without the change: after a stop and start of the route one message is routed twice (Expected: <1> but was: <2>). With the change it passes, with no subscription left after the stop and the dispatcher of the first run closed.stoppingTheRouteDoesNotInterruptTheExchangeInFlight(found in review): the route stops while a step waits on a latch; the step must not be interrupted and complete, and the dispatcher is closed afterwards. On the first version of this branch (closing the dispatcher indoStop) it fails withthe step in flight must not be interrupted by the stop ==> expected: <false> but was: <true>.eachRunOfTheConsumerClosesItsOwnDispatcher(found in review): the route's consumer is stopped and started in place (as a route policy does with a consumer that is notSuspendable) while a message of the first run is in flight, then stopped again: the second run's dispatcher is closed at once and the first one when its message is done. With the in-flight count on the consumer it fails withthe dispatcher of the second run is closed when it stops ==> expected: <1> but was: <0>.stopWhileSubscribingIsNotReportedAsAnError(found in review): the consumer stops from inside the subscribe of the fake server; nothing must reach the exception handler. Without the change it fails withreported: [java.lang.IllegalStateException: Dispatcher is closed].What was not run: a real NATS server (the integration tests need a container); 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-nats. 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). Each commit carries a
Co-Authored-Bytrailer.Claude Code on behalf of allthingssecurity
🤖 Generated with Claude Code