allthingssecurity opened a new pull request, #29788: URL: https://github.com/apache/camel/pull/29788
# Description [CAMEL-26884](https://issues.apache.org/jira/browse/CAMEL-26884) Follow-up of #29042 (CAMEL-26462, a consumer that stops before its task subscribes): two of the three gaps listed in its review, and the same flush failure in `NatsProducer.doStop`. - *Flush on a disconnected or closed connection.* `NatsConsumer.doStop` first flushes the connection (`flushConnection=true` by default). jnats fails the flush of a connection that is not connected (the server is down or restarting) with a `TimeoutException`, and `doStop` ended there: the dispatcher was not unsubscribed or closed, the JetStream subscription not ended, the executor of the consumer not shut down, and a connection the consumer created not closed. The consumer is then marked failed, so a later stop does not retry the cleanup. With a shared connection (`connection` option) the stopped route kept its subscription when the server came back, and a restart of the route added a second one (each message routed twice); a connection the consumer created kept reconnecting (by default 60 attempts, 2 seconds apart), and when the server came back meanwhile its dispatcher subscribed again. The route stop itself did not fail, as the shutdown strategy logs the failure of a consumer s top at WARN. - *A JetStream pull subscription made while the consumer stops.* The pull subscribe runs on the task of the consumer. When the subscribe completed just before the stop, and the task stored the subscription after `doStop` had looked for it, the subscription was never unsubscribed (the fetch loop does not run, as the consumer is stopping): the connection and the server kept it, with a shared connection until that connection closes. Ending it does not delete the JetStream consumer on the server, before or after this change. The window is narrow (a subscribe in flight during the stop usually fails on the interrupt of the stop). Change: - the flush in `NatsConsumer.doStop` is best effort: a failure is logged at WARN (the stack trace at DEBUG) and the stop goes on; an interrupt during the flush is kept until the end of the stop, so the unsubscribes of the stop do not run on an interrupted thread (jnats drops a message to the server queued there); - the same in `NatsProducer.doStop`, which flushes the connection it created before it closes it: on main a failed flush skipped the close, so the connection kept reconnecting; - after the pull subscribe, the task ends the subscription itself when the consumer is stopping (if it is still active; `doStop` may have ended it meanwhile), with the interrupt of the stop cleared for the unsubscribe and set again after it, as jnats drops a message to the server that it queues on an interrupted thread (the same care as #29042); - `doStop` reports a failed unsubscribe of the JetStream subscription only while it is still active (the task may have ended it first). Not changed (left for later): a quick stop and start while the task of the first run still subscribes, and the fields of a run that gave up, which need a token per run of the consumer (I will file a follow-up JIRA for it, linked from CAMEL-26462). Tests, new `NatsConsumerStopGapsTest` (fake connections, no server, as the other `NatsConsumer*Test` classes): - `stopWhenTheFlushFailsEndsTheSubscription`: the flush fails with the `TimeoutException` jnats throws; the route stop ends the push subscription and closes the dispatcher; - `producerStopWhenTheFlushFailsClosesItsConnection`: a producer with its own connection (the endpoint returns a fake connection whose flush fails): the stop closes the connection; - `pullSubscriptionMadeWhileStoppingIsEnded`: the pull subscribe completes just before the stop and the task gets the subscription after `doStop` looked for it; the subscription is ended, not on an interrupted thread. Without the change the three fail (the subscription stays: 1 instead of 0; the pull subscription is never unsubscribed, and the producer stop fails without closing its connection). With it the camel-nats unit tests pass (43 run, 0 failures). The ITs need a NATS container and were not run locally (CI runs them). No documentation or upgrade guide change. Backport: the flush in `doStop` is the same on 4.18.x and 4.14.x; the pull-subscribe code differs there, and the test reuses fakes that are on main only (CAMEL-26462 and earlier), so both need adapting. # 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-26884](https://issues.apache.org/jira/browse/CAMEL-26884). # 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 the core modules, camel-test-infra-nats and camel-nats with the formatter and import-sort plugins and ran the camel-nats unit tests. I did not run the full root build. The ITs, which need a NATS container, were not run locally; CI runs them.) # 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]
