allthingssecurity opened a new pull request, #29042: URL: https://github.com/apache/camel/pull/29042
# Description [CAMEL-26462](https://issues.apache.org/jira/browse/CAMEL-26462) Follow-up of CAMEL-26366 and CAMEL-25847. While working on the flaky push consumer tests of CAMEL-26366 (fixed by #28614 with test changes only), the test `stopWhileSubscribingIsNotReportedAsAnError` also showed a race in the consumer itself, which is still on main. This PR fixes only that race; the tests of CAMEL-26366 are kept as they are on main. The JetStream push consumer subscribes on its own executor (`NatsConsumingTask`), after `doStart` returned. Since CAMEL-25847, `doStop` ends the push subscription through the dispatcher of the consumer and closes the dispatcher (once no message is in flight), which matters when the connection is shared (`connection=#conn`) and stays open after the stop. But `doStop` only sees the dispatcher once the task has set it in the field `dispatcher` (which was not `volatile` either). When the consumer stops after the task created the dispatcher and before it set it, `doStop` finds nothing to close, shuts the executor down (which interrupts the task) and returns. The task then goes on: - The push subscribe asks the server first (jnats `NatsJetStreamImpl.makeRequestResponseRequired`, which turns an `InterruptedException` of the request into an `IOException`), so on the interrupted thread it fails. The task logs that at DEBUG (the consumer is stopping), and the dispatcher it created stays open, with its thread, on the shared connection until the connection closes: one leaked dispatcher per such stop. - If the subscribe is made anyway (`doStop` reads the dispatcher before it shuts the executor down, so the task can set the dispatcher and subscribe in between), the task sees `isRunAllowed() == false` and stops the keeper of CAMEL-25845 (the comment said "stopped while subscribing: doStop did not see the keeper"), but not the dispatcher: the subscription stays active, and the stopped route keeps getting messages from the shared connection. The task also marked the consumer active (which matters only for a first run: the flag is never reset to false). ## Changes - Right after the task created the dispatcher of its run and set it in the field (and before it subscribes), it checks the state of the consumer (`stoppedBeforeSubscribe`). When the consumer is stopping, the task closes that dispatcher, which has no subscription yet, and returns (so it does not subscribe or start a keeper). The same check is in the push path and in the consumer without JetStream (`setupStandardNatsConsumer`), which had the same window: its dispatcher would get the messages of its subject from the shared connection after the stop. - `NatsConsumer.dispatcher` is `volatile`. The task writes it before it reads the state of the consumer (`BaseService.status`, volatile), and a stop sets the state before `doStop` reads the field. So the task sees the stop and closes its dispatcher, or `doStop` sees the dispatcher and closes it on the stopping thread (a subscribe on the closed dispatcher then fails: jnats "Dispatcher is closed", as in `stopWhileSubscribingIsNotReportedAsAnError`), or both. There is no order in which neither closes it. - Both happen when the stop begins shortly before the task sets its dispatcher: `doStop` flushes the connection before it reads the field, so the task can set the field, see the stop and close its dispatcher meanwhile. `doStop` then finds a closed dispatcher, whose unsubscribe fails ("Dispatcher is closed"): it now reports a failed unsubscribe only when the dispatcher is still active (`Dispatcher.isActive()`, the `running` flag of jnats, false once the dispatcher is closed and never true again, so checking it after the failure is not racy), and logs at DEBUG otherwise (also when the close of the connection closed the dispatcher, which before was reported as well); its close of the dispatcher is refused by jnats ("Dispatcher is already closed.") and logged at DEBUG by `closeDispatcher`, as before. Without this, the unsubscribe on the closed dispatcher would report "Dispatcher is closed" to the exception handler (a WARN with a stack trace on a normal stop). - Why before the subscribe and not after it (an earlier version of this branch closed the dispatcher after the subscribe): `doStop` interrupts the task (it shuts its executor down), and jnats queues the UNSUB of a closed dispatcher with `WriterMessageQueue.push`, which drops the message when the thread is interrupted (`tryLock` throws, `push` returns false; checked in the 2.26.4 jar and with a probe against a protocol-logging server). A close on the task thread after its subscribe would end the subscription only on the client, and the server would keep it: a queue group would lose the share of the dead member, an ephemeral JetStream consumer would not go away, and a durable one would stay bound (a restart fails with "Consumer is already bound to a subscription"). Before the subscribe there is nothing to unsubscribe. - The push path keeps the check after its subscribe, as on main, to stop the keeper that `doStop` did not see (it saw the dispatcher, and ended the subscription through it); it now returns there before `setActive(true)`. - `doStop` reads the dispatcher field once into a local; the two unsubscribe failures go through one helper (`unsubscribeFailed`). - The standard path uses the local dispatcher instead of reassigning the field with the return value of `Dispatcher.subscribe` (jnats returns the same dispatcher). Not changed (all on main as well; follow-up JIRA draft to file, one issue for the stop of the NATS consumer): - The JetStream pull subscription: a pull subscription made after a stop that `doStop` did not see gets no messages (the fetch loop does not run when the consumer is stopping), but it is not unsubscribed either. - A quick stop then start: `isRunAllowed()` is the state of the consumer, not of a run, so the task of the stopped run can see the consumer running again, subscribe and overwrite the fields of the new run (the dispatcher of one run is then never closed). That needs a token per run. - After the task gave up, the fields still point at the closed dispatcher of that run until the next run sets them (a stop of the next run before its task sets them now finds it inactive and logs at DEBUG). - `doStop` flushes the connection first; when the connection is disconnected or closed, `flush` throws a `TimeoutException`, and `doStop` ends there, before it closes the dispatcher or shuts the executor down. Backport: main only. The older branches (4.18.x, 4.14.x) have neither CAMEL-25847 (`closeWhenIdle`) nor CAMEL-25845 (the keeper), so a backport is not a cherry-pick. No upgrade guide entry: a stopped consumer that keeps a dispatcher or a subscription is a bug, and no option changes. ## Tests Three new tests in `NatsConsumerJetStreamPushStopTest`, with the fake JetStream server of the class (no container). The fake gets a hook after it created a dispatcher (before the consumer gets it), and it behaves as jnats on an interrupted thread: a push subscribe fails with an `IOException` (the request that jnats makes to the server first), and a close leaves the subscriptions of the dispatcher on the server (the dropped UNSUB). Closing a dispatcher twice is refused, and a closed dispatcher answers `isActive()` with false, as in jnats; a hook in its `flush` lets a test hold `doStop` there. - `stopBeforeTheRunSetsItsDispatcherClosesThatDispatcherBeforeItSubscribes`: the consumer stops right after its task created the dispatcher (the stop runs inside the task, so it interrupts it, as `doStop` does). The stopped run does not try to subscribe, the dispatcher of each run is closed (2), no subscription is left, nothing is reported to the exception handler. - `stopThatSeesTheDispatcherClosedByTheRunIsNotReportedAsAnError`: the stop runs on another thread and waits in its flush (a hook of the fake) until the task has closed its dispatcher; then `doStop` finds the closed dispatcher. The consumer stops, the dispatcher is closed once, the stopped run does not subscribe, nothing is reported. (On main this order does not exist, as the task never closes its dispatcher; the test fails there by the timeout of its own wait in the flush.) - `stopBeforeAStandardConsumerSetsItsDispatcherClosesThatDispatcher`: the same for a consumer without JetStream on the shared connection (`nats:plain?connection=#conn`): its dispatcher is closed, nothing is reported. Each test waits for the task of the stopped run to end (the stop shut its executor down, so its thread ends with the task; the recorded thread is reset before the start) before it asserts, so no assertion runs before the path it checks. Without the fix, the three new tests fail (the stopped run tries to subscribe on the interrupted thread and its dispatcher stays open; the standard dispatcher is not closed; the third as said above); the four other tests of the class pass. Mutants: no check at all fails the three; no check in the push path fails the two push tests; no check in the standard path fails the standard test only; reporting every unsubscribe failure as before fails the third test ("Dispatcher is closed" reported). - `NatsConsumerJetStreamPushStopTest` 20 times in one JVM: 140 tests, 0 failures. - camel-nats tests (all the unit tests of the module; the `*IT` tests need a NATS server in a container and were not run locally, CI runs them): 40 tests, 0 failures. - Formatter and import-sort: no changes. # 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-26462) filed for the change (usually before you start working on it): [CAMEL-26462](https://issues.apache.org/jira/browse/CAMEL-26462). # 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 camel-nats and ran its tests, including the formatter and import-sort plugins. No generated file changes. 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]
