allthingssecurity opened a new pull request, #28320: URL: https://github.com/apache/camel/pull/28320
# Description [CAMEL-25847](https://issues.apache.org/jira/browse/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 called `unsubscribe()` 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 `connection` option) 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`): - End the push subscription through its dispatcher (`dispatcher.unsubscribe(jetStreamSubscription)`, which jnats accepts for a subscription of that dispatcher). This does not touch the dispatcher thread. - Close the dispatcher the consumer created (`connection.closeDispatcher`, while the connection is open), which ends its thread, once no message is in flight on it. jnats closes a dispatcher with `NatsDispatcher.stop(true)`, which cancels its thread with `Future.cancel(true)`, an interrupt (javap, jnats 2.26.3), and that thread runs the route: Camel stops a consumer that is not `Suspendable` before 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 handler counts the 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 existing unsubscribe calls are kept. Note: Camel itself fails an exchange that is in flight when its consumer stops at its next step (`RouteStoppingException`, CAMEL-25502: `DefaultConsumer.doStop` stops the route's processors), on main as well. What the change keeps is that the step in progress is not interrupted. 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. No option changed, so no upgrade guide entry. Tests: - New `NatsConsumerJetStreamPushStopTest` with a fake JetStream server behind the jnats interfaces the consumer calls (like jnats, it refuses `unsubscribe()` 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). `stoppedRouteGetsNoMessagesAndRestartedRouteGetsEachMessageOnce` fails 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 in `doStop`) it fails with `the step in flight must not be interrupted by the stop ==> expected: <false> but was: <true>`. - With the change the camel-nats unit tests pass: 7 run, 0 failures. Merged with the CAMEL-25845 branch: 19 run, 0 failures. Merge-tree clean with the CAMEL-25815 branch. What was not run: a real NATS server (the integration tests need a container); the full root build. # Target - [x] I checked that the commit is targeting the correct branch (Camel 4 uses the `main` branch) # Tracking - [x] If this is a large change, bug fix, or code improvement, I checked there is a [JIRA issue](https://issues.apache.org/jira/browse/CAMEL) filed for the change (usually before you start working on it): [CAMEL-25847](https://issues.apache.org/jira/browse/CAMEL-25847). # Apache Camel coding standards and style - [x] I checked that each commit in the pull request has a meaningful subject line and body. - [ ] I have run `mvn clean install -DskipTests` locally 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 - [x] If this PR includes AI-generated code, commits have proper co-authorship attribution (e.g., `Co-authored-by` trailers) 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-By` trailer. _Claude Code on behalf of allthingssecurity_ 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
