[allthingssecurity]

Description

CAMEL-25845

With jetstreamEnabled=true and pullSubscription=false the consumer uses a JetStream push subscription, and without durableName the JetStream consumer behind it is ephemeral. The NATS server deletes an ephemeral push consumer when nobody is subscribed to its deliver subject for its inactive threshold (5 seconds by default), for example while the client was disconnected for a few seconds before jnats reconnected, and a server restart loses it too. jnats subscribes the deliver subject again after the reconnect but does not create the consumer again. The route stayed started and consumed nothing again, without any log, until it was restarted.

This is the push twin of CAMEL-25815 (#28204), whose fix only changes the pull loop. It follows the reviews of #28204: only an ephemeral consumer is created again, a failed subscribe is reported and tried again, the new consumer carries no start time, a null consumer info counts as gone (that is how jnats reports a deleted consumer), and the resume looks at the stream first (WorkQueue retention, a stream created again). Both changes make NatsConsumer.jetStreamSubscription volatile and replace streamExists with the same lookupStream. #28204 merged first, and this branch is rebased onto main after it (the conflict was the Status import and the field next to jetStreamSubscription; the lookupStream lines were identical and merged cleanly).

Changes (NatsConsumer, new package-private NatsJetStreamPushConsumerKeeper, upgrade guide, nats-component.adoc):

• The keeper is the message handler of the push subscription (it passes every message on to the existing handler) and a connection listener. On RESUBSCRIBED it asks the server for the consumer (getConsumerInfo, on the consumer's executor). A reconnect does not queue a second check while one waits.
• jnats (2.26.4, the version of Camel main; 2.26.3 does the same) answers getConsumerInfo() with null when the server has no such consumer (lookupConsumerInfo maps api error 10014, consumer not found, to null) and throws only for other errors, such as 10059 stream not found. The keeper treats null as gone (and still a 404 error); a check that waited for an exception never saw a deleted consumer.
• When an ephemeral consumer is gone it subscribes again, starting at the first stream sequence delivered to the route and not acknowledged (it checks lastAck() of the delivered messages, so a nak'ed message counts as not acknowledged), or after the last one delivered: nothing is skipped, and the stream is not replayed from the deliver policy. The new configuration has no start time (with DeliverPolicy.ByStartTime the copied start time next to the start sequence would be rejected by the server). A message delivered maxDeliver times (when set, by the endpoint option or the consumerConfiguration) counts as settled even when it was nak'ed, as the server does not deliver it again (deliveredCount() of its metadata): a poison message no longer holds the resume point back for the life of the route, which would replay everything after it at the next deletion. A message that is not acknowledged and can still be delivered keeps the old start (at-least-once). With AckPolicy.None the delivery is the acknowledgement: the keeper does not keep the delivered messages (a route with manualAck=true may never call ack(), which would keep them for the life of the route) and the new consumer starts after the last one delivered.
• When nothing was delivered yet, a consumer with the deliver policy New (or Last) starts after (or at) the last stream sequence seen at the subscribe (NatsConsumer passes it from the stream lookup), so the messages published during the outage are not skipped (the configuration as it is would start at the new messages again).
• Before it subscribes again the keeper looks the stream up (getStreamInfo). A stream with WorkQueue retention refuses any deliver policy other than All (400, error 10101) and has already removed the acknowledged messages, so the consumer configuration is used as it is. A stream whose create time differs from the one seen at the subscribe was created again (for example a memory stream after a server restart, or by an administrator); its sequences start at 1 again, and a start sequence of the old stream would make the server skip its messages without an error. The keeper then clears what it tracked, logs a WARN and uses the consumer configuration. NatsConsumer passes the create time it saw when it looked the stream up at start (streamExists becomes lookupStream, which returns the StreamInfo).
• A durable consumer (a durableName, or a durable name in the consumerConfiguration) is not created again: it is gone only when an administrator deleted it or after its inactiveThreshold, and creating it again from its configuration would replay the stream from its deliver policy (the whole stream with the default DeliverPolicy.All). The keeper logs a WARN and keeps the subscription; messages arrive again when the durable is created again with the same deliver subject.
• A failed check or subscribe is tried again while the route runs: five attempts a second apart, then with a delay that doubles up to 30 seconds, with one WARN after the fifth attempt that says the route receives nothing until it succeeds (JetStream not ready yet after a server restart, or a stream that is gone until it is created again: jnats throws 404 stream not found from getConsumerInfo and getStreamInfo). The first failed subscribe of a reconnect goes to the consumer's exception handler (jnats throws IllegalStateException for some, for example SUB-90007 when the stream is gone), the next ones are logged at DEBUG. A subscribe that fails because the consumer is stopping (with CAMEL-25847 the stop closes the dispatcher, and jnats refuses a subscribe on a closed dispatcher) is not reported. The WARNs name why the consumer counts as gone (consumer not found, or the 404 error such as stream not found).
• The listener is removed when the consumer stops, also when the route is stopped while the consumer makes its first subscription (found in review: doStop did not see the keeper yet; the task now stops it when the consumer may no longer run, and pushConsumerKeeper is volatile). A subscription made while the consumer was being stopped is ended by the keeper. After a resubscribe only the messages of the old subscription are dropped from tracking, so a message the new consumer delivers meanwhile stays tracked.
• Upgrade guide (4.24): a short migration note (the ephemeral consumer is created again after a reconnect from the first unacknowledged message; a durable one is not, WARN; a consumer deleted while connected is not detected). The details (ackPolicy=None, maxDeliver, New/Last, WorkQueue and re-created streams, the retries) are in a new section of nats-component.adoc (and its catalog copy), as asked in review.

What users can notice: after an outage of more than about 5 seconds or a server restart, a push consumer without durableName consumes again (a WARN says it subscribed again and from which stream sequence). A durable consumer that is gone now gets a WARN after a reconnect. No option changed.

Traffic: the keeper asks the server for the consumer only after a reconnect (RESUBSCRIBED): one CONSUMER.INFO request per reconnect (more, at most every 30 seconds, while it fails), plus one STREAM.INFO when it subscribes again. An idle healthy push consumer sends nothing extra (the pull twin had to limit its periodic check; the push keeper has none).

Stopping: the keeper only checks while the consumer runs, and when the consumer was stopped while it subscribed again, it ends the subscription it just made (the stop may have ended only the previous one). NatsConsumer.jetStreamSubscription is now volatile, as the keeper sets it on another thread than the one that stops the consumer.

Not changed (review): the consumer is checked only after a reconnect, so a consumer deleted while the connection stays up (nats consumer rm) is not detected, unlike the pull consumer of CAMEL-25815; an idle heartbeat is not allowed with a queue group and its alarm goes to the connection's ErrorListener. This is stated in the keeper javadoc, the guide and the component page, and a follow-up JIRA is drafted. A shared resume helper for pull and push (and a shared FakeServer test helper) is a second drafted follow-up. The keeper javadoc also says that Message.lastAck() can be read stale, which makes the consumer resume earlier (redelivery), never skip.

Tests:

• New NatsConsumerPushConsumerDeletedTest (13 tests) with a fake JetStream server behind the jnats interfaces the consumer calls. It answers like jnats: getConsumerInfo() returns null for a deleted consumer and the info for a live one; a WorkQueue stream removes acknowledged messages and refuses a consumer whose deliver policy is not All (10101); a stream created again starts at sequence 1 with a new create time; a start time next to ByStartSequence is rejected, as nats-server does; a nak'ed message is not delivered again after maxDeliver deliveries, and the delivered count is in the reply subject; while the stream is gone, getStreamInfo and getConsumerInfo throw 404 stream not found (10059) and a subscribe throws SUB-90007; DeliverPolicy.New starts after the last message; a new dispatcher per createDispatcher, and a closed dispatcher refuses a subscribe, as jnats does.
• consumesAgainAfterTheServerDeletedTheEphemeralConsumer, redeliversTheMessageWhoseNakWasLostWithTheConsumer: the messages published during the outage are routed, a message whose nak was lost with the deleted consumer is delivered again, and the message acknowledged before the deletion is not routed again. endsTheSubscriptionItMadeWhileTheRouteStopped stops the route while the consumer subscribes again; the new subscription is ended (or, merged with CAMEL-25847, refused by the closed dispatcher), and nothing goes to the exception handler. doesNotCreateAGoneDurableConsumerAgain (durableName=orders): no new consumer after the reconnect (asserted for 3 s). reportsAFailedSubscribeAndTriesAgain: the first subscribe after the reconnect throws IllegalStateException (SUB-90007); the endpoint's exceptionHandler gets it and the next attempt routes the message. resumesAnEphemeralConsumerThatStartedAtATime (DeliverPolicy.ByStartTime).
• workQueueStreamIsSubscribedAgainWithDeliverPolicyAll, consumesAStreamCreatedAgainFromItsStart (the stream is created again during the outage; order-3 is sequence 1 of the new stream) and ackPolicyNoneResumesAfterTheHighestDelivered (manualAck=true&ackPolicy=None, the route never acks; the new consumer starts at sequence 3).
• Against the previous version of this branch (ffc2d94, with the fake answering like jnats) 8 of the 9 fail: nothing is routed after the reconnect (Expected: <2> but was: <0>, Expected: <1> but was: <0>), because a null consumer info never counted as gone; doesNotCreateAGoneDurableConsumerAgain passes there for the same reason. With only the null fix applied, the three new tests still fail: WorkQueue Expected: <1> but was: <0> (every resubscribe rejected with 10101), re-created stream Expected: <1> but was: <0> (the new consumer starts at sequence 3 of the old stream), ackPolicy=None Body of message: 0. Expected: <order-3> but was: <order-1> (the never-acked messages are replayed). With the change all 9 pass.
• Three tests from review, which fail on the previous version of this branch (57555db): aMessageDeliveredMaxDeliverTimesDoesNotHoldBackTheResume (maxDeliver=3, a poison message nak'ed three times, then two messages routed, then the consumer deleted) with Body of message: 0. Expected: <order-4> but was: <order-2>; keepsCheckingUntilTheStreamIsCreatedAgain (the stream is gone after the reconnect and created again after six checks) with the consumer keeps checking, 5 ==> expected: <true> but was: <false> within 30 seconds; deliverPolicyNewWithNothingDeliveredResumesAfterTheLastMessageAtTheSubscribe with Received message count. Expected: <1> but was: <0>.
• removesItsConnectionListenerWhenTheRouteStopsWhileItSubscribes (found in review): the route is stopped from inside its first subscribe; no keeper may stay registered on the connection. Without the guard it fails with no keeper listens to the connection after the stop ==> expected: <[]> but was: <[...NatsJetStreamPushConsumerKeeper@...]>. It also accepts a subscribe refused on the closed dispatcher, which is what happens once merged with CAMEL-25847 (CAMEL-25847: camel-nats - a stopped JetStream push consumer unsubscribes through its dispatcher #28320, d0c258b).
• Merged locally with CAMEL-25847 (stop unsubscribes, CAMEL-25847: camel-nats - a stopped JetStream push consumer unsubscribes through its dispatcher #28320 at 3fe5f7b) on top of main: 33 run, 0 failures; without the "not reported while stopping" check, endsTheSubscriptionItMadeWhileTheRouteStopped fails there with expected: <[]> but was: <[java.lang.IllegalStateException: Dispatcher is closed]>.
• With the change the camel-nats unit tests pass: 29 run, 0 failures, including the pull tests of CAMEL-25815: camel-nats - the JetStream pull consumer subscribes again when the server deleted its consumer #28204 (NatsConsumerPullConsumerDeletedTest, NatsConsumerPullConsumerGoneTest) (formatter and import sort in the build). Rebased onto main after CAMEL-25815: camel-nats - the JetStream pull consumer subscribes again when the server deleted its consumer #28204 merged; it still merges cleanly with CAMEL-25847: camel-nats - a stopped JetStream push consumer unsubscribes through its dispatcher #28320.

What was not run: a real NATS server (the integration tests need a container), so an IT against nats-server (jsm.deleteConsumer(...) plus forceReconnect()), suggested in review, is not part of this PR; the full root build.

Target

• I checked that the commit is targeting the correct branch (Camel 4 uses the main branch)

Tracking

• If this is a large change, bug fix, or code improvement, I checked there is a JIRA issue filed for the change (usually before you start working on it): CAMEL-25845.

Apache Camel coding standards and style

• 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; the upgrade guide has a new entry. I did not run the full root build.)

AI-assisted contributions

• 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). Each commit carries a Co-Authored-By trailer.

Claude Code on behalf of allthingssecurity

🤖 Generated with Claude Code

[github-actions]

🌟 Thank you for your contribution to the Apache Camel project! 🌟
🤖 CI automation will test this PR automatically.

🐫 Apache Camel Committers, please review the following items:

• First-time contributors require MANUAL approval for the GitHub Actions to run
• You can use the command /component-test (camel-)component-name1 (camel-)component-name2.. to request a test from the test bot although they are normally detected and executed by CI.
• You can label PRs using skip-tests and test-dependents to fine-tune the checks executed by this PR.
• Build and test logs are available in the summary page. Only Apache Camel committers have access to the summary.

⚠️ Be careful when sharing logs. Review their contents before sharing them publicly.

[davsclaus]

Thanks Shashank, this is a careful push twin of CAMEL-25815 (#28204). Pulling the logic into its own NatsJetStreamPushConsumerKeeper keeps NatsConsumer readable. The resume rules match #28204: an ephemeral consumer only, start at the first unacked sequence, startTime(null), WorkQueue and a re-created stream use the configuration as it is. The tests fail without the fix. The fake is large but every behaviour it models is used by a test, and it follows the style of the #28204 fake, so I don't think it is over-engineered.

TL;DR

Looks good, no bugs found. A scope question and some consistency points with #28204 below. CI is still running.

Blockers: None

Other findings: 4

Non-blocking

1. The check runs only after a reconnect. A consumer that is deleted while the connection stays up (for example nats consumer rm, which is exactly what NatsJetstreamConsumerPullDeletedIT in #28204 does) is never noticed, and the route silently stops consuming. The pull consumer does notice this case through its empty-fetch check, so the two modes now differ. I think the reconnect-only scope is fine for this JIRA. An idle heartbeat cannot be used with a queue group (JsSubFcHbNotValidQueue), and its alarm goes to the connection's ErrorListener, which Camel does not own when the connection is shared. Could you state the limitation in the keeper javadoc and the upgrade-guide entry ("only after a reconnect; a consumer deleted while connected is not detected")? See the inline comment.

2. Two ways to track unacknowledged messages. #28204 removes a sequence from its set through explicit callbacks (acknowledged(msg) in onComplete/onFailure, plus DefaultNatsManualAck's onAcknowledged). This PR instead scans Message.lastAck() on every delivery. Both work, and if lastAck is stale (the field is not volatile in jnats) the consumer resumes earlier and redelivers rather than skipping, so it fails safe. Still, the resume code (stream re-created, WorkQueue, start-sequence calculation) is now in the codebase twice. They also already differ: push handles DeliverPolicy.New/Last when nothing was delivered yet, and pull does not (resume = !workQueue && highestFetched.get() > 0). Not for this PR, but a follow-up JIRA to share one resume helper between the two modes would keep them from drifting further.

3. Merge conflict with #28204. Merging both onto main gives a small textual conflict in NatsConsumer.java: the io.nats.client.support.Status import and the volatile JetStreamSubscription jetStreamSubscription field (this PR adds a comment and the pushConsumerKeeper field next to it). The identical streamExists -> lookupStream change and the upgrade-guide entries merge cleanly. Whichever PR goes in second needs a rebase. This PR merges cleanly with #28320, and the two work together: after #28320 closes the dispatcher on stop, a keeper subscribe that races the stop is refused with "Dispatcher is closed" and swallowed because isRunAllowed() is false. endsTheSubscriptionItMadeWhileTheRouteStopped already allows for that.

4. Keeper registered after a stop that came during startup. See the inline comment. This is a narrow window, and the subscription itself has the same race on main today.

Suggestions

• An IT against a real nats-server would check the jnats behaviour the fake assumes: getConsumerInfo() returns null for 10014, and RESUBSCRIBED fires after the reconnect. With a shared connection, jsm.deleteConsumer(...) followed by connection.forceReconnect() should exercise the whole path, in the same way as the pull IT in #28204.
• The upgrade-guide entry is about 20 lines and reads more like reference documentation. The upgrade guide is meant for migration notes, so maybe keep two or three sentences there (the push consumer now creates a deleted ephemeral consumer again after a reconnect, resuming at the first unacked message, and a durable is not created again) and move the details to nats-component.adoc. That is optional, since #28204 used the same style.
• Three test classes (#28204, this PR, #28320) now each have their own copy of the FakeServer/proxy(...) scaffolding. A shared test helper in a later cleanup would save about 100 lines per copy.

Claude Code on behalf of @davsclaus. This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.

[github-actions]

🧪 CI tested the following changed modules:

• catalog/camel-catalog
• components/camel-nats
• docs

🔁 2 tests passed only after a retry on JDK 21 (3 retried attempts)

Recovered flaky tests on JDK 21 (2)

Module	Test	Failed attempts	First failure
components/camel-nats	NatsConsumerPushConsumerDeletedTest.removesItsConnectionListenerWhenTheRouteStopsWhileItSubscribes	2	Condition with Lambda expression in org.apache.camel.component.nats.NatsConsumerPushConsumerDeletedTest was not fulfilled within 20 seconds.
dsl/camel-jbang/camel-jbang-plugin-tui	AiPanelAcpTest.unknownSlashCommandIsForwardedToTheAgentWithoutThePreamble	1	Index 1 out of bounds for length 1

ℹ️ These tests did not fail the build. Surefire retried them and they passed.
Retries are enabled project-wide by surefire.rerunFailingTestsCount in parent/pom.xml.

---

🔬 Scalpel shadow comparison — Scalpel: 8 of 707 tested, 25 compile-only — current: 8 all tested

Maveniverse 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)

• camel-jbang-mcp ← downstream of org.apache.camel:camel-catalog
• camel-jbang-plugin-mcp ← downstream of org.apache.camel:camel-jbang-core
• camel-jbang-plugin-tui ← downstream of org.apache.camel:camel-catalog
• camel-jbang-plugin-validate ← downstream of org.apache.camel:camel-yaml-dsl-validator
• camel-launcher-container ← downstream of org.apache.camel:camel-launcher
• camel-nats ← components/camel-nats/src/main/docs/nats-component.adoc, components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java, components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsJetStreamPushConsumerKeeper.java, components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPushConsumerDeletedTest.java
• camel-yaml-dsl-validator ← downstream of org.apache.camel:camel-catalog
• camel-yaml-dsl-validator-maven-plugin ← downstream of org.apache.camel:camel-yaml-dsl-validator

Modules with tests skipped (25)

• apache-camel
• camel-allcomponents
• camel-catalog-console
• camel-catalog-maven
• camel-catalog-suggest
• camel-componentdsl
• camel-endpointdsl
• camel-endpointdsl-support
• camel-itest
• camel-jbang-core
• camel-jbang-it
• camel-jbang-main
• camel-jbang-plugin-edit
• camel-jbang-plugin-generate
• camel-jbang-plugin-kubernetes
• camel-jbang-plugin-test
• camel-kamelet-main
• camel-launcher
• camel-report-maven-plugin
• camel-route-parser
• camel-yaml-dsl
• camel-yaml-dsl-deserializers
• camel-yaml-dsl-maven-plugin
• coverage
• dummy-component

ℹ️ Shadow mode — Scalpel observes but does not affect test execution. Learn more

All tested modules (35 modules, 5m 39s total)

Total reactor time: 5m 39s

Module	Duration	Status
Camel :: Launcher	50.5s	SUCCESS
Camel :: JBang :: MCP	42.6s	SUCCESS
Camel :: JBang :: Plugin :: Validate	28.6s	SUCCESS
Camel :: Component DSL	25.8s	SUCCESS
Camel :: YAML DSL	25.3s	SUCCESS
Camel :: Catalog :: Camel Catalog	24.3s	SUCCESS
Camel :: YAML DSL :: Validator	21.6s	SUCCESS
Camel :: Docs	15.1s	SUCCESS
Camel :: JBang :: Plugin :: Testing	14.0s	SUCCESS
Camel :: Kamelet Main	13.5s	SUCCESS
Camel :: JBang :: Plugin :: Kubernetes	12.9s	SUCCESS
Camel :: YAML DSL :: Deserializers	10.5s	SUCCESS
Camel :: Catalog :: Camel Route Parser	9.2s	SUCCESS
Camel :: Catalog :: Camel Report Maven Plugin	7.5s	SUCCESS
Camel :: YAML DSL :: Validator Maven Plugin	6.6s	SUCCESS
Camel :: Assembly	6.6s	SUCCESS
Camel :: All Components Sync point	5.4s	SUCCESS
Camel :: YAML DSL :: Maven Plugins	3.7s	SUCCESS
Camel :: Catalog :: Maven	2.7s	SUCCESS
Camel :: Catalog :: Suggest (deprecated)	2.2s	SUCCESS
Camel :: Coverage	1.6s	SUCCESS
Camel :: JBang :: Integration tests	1.3s	SUCCESS
Camel :: JBang :: Plugin :: Edit	1.2s	SUCCESS
Camel :: Catalog :: Dummy Component	1.1s	SUCCESS
Camel :: JBang :: Plugin :: Generate	1.1s	SUCCESS
Camel :: Endpoint DSL :: Support	1.1s	SUCCESS
Camel :: Catalog :: Console	0.9s	SUCCESS
Camel :: JBang :: Main	0.9s	SUCCESS
Camel :: JBang :: Plugin :: MCP	0.8s	SUCCESS
Camel :: Launcher :: Container	0.8s	SUCCESS
Camel :: Endpoint DSL	n/a
Camel :: Integration Tests	n/a
Camel :: JBang :: Core	n/a
Camel :: JBang :: Plugin :: TUI	n/a
Camel :: Nats	n/a

Top 20 slowest modules:

• Camel :: Launcher (50.5s)
• Camel :: JBang :: MCP (42.6s)
• Camel :: JBang :: Plugin :: Validate (28.6s)
• Camel :: Component DSL (25.8s)
• Camel :: YAML DSL (25.3s)
• Camel :: Catalog :: Camel Catalog (24.3s)
• Camel :: YAML DSL :: Validator (21.6s)
• Camel :: Docs (15.1s)
• Camel :: JBang :: Plugin :: Testing (14.0s)
• Camel :: Kamelet Main (13.5s)
• Camel :: JBang :: Plugin :: Kubernetes (12.9s)
• Camel :: YAML DSL :: Deserializers (10.5s)
• Camel :: Catalog :: Camel Route Parser (9.2s)
• Camel :: Catalog :: Camel Report Maven Plugin (7.5s)
• Camel :: YAML DSL :: Validator Maven Plugin (6.6s)
• Camel :: Assembly (6.6s)
• Camel :: All Components Sync point (5.4s)
• Camel :: YAML DSL :: Maven Plugins (3.7s)
• Camel :: Catalog :: Maven (2.7s)
• Camel :: Catalog :: Suggest (deprecated) (2.2s)

---

⚙️ View full build and test results

[allthingssecurity]

@davsclaus thanks, addressed in a9dbaf2:

1. Reconnect-only check: kept as the scope of this JIRA. The limitation is now stated in the keeper javadoc, the upgrade guide and nats-component.adoc. I drafted a follow-up JIRA to detect a consumer deleted while connected.
2. lastAck() vs explicit callbacks: a javadoc line on the stale read (it fails safe, by redelivering). I drafted a follow-up JIRA to share one resume helper between pull and push (and the FakeServer test scaffolding), which would also remove the New/Last difference.
3. Conflict with CAMEL-25815: camel-nats - the JetStream pull consumer subscribes again when the server deleted its consumer #28204 in NatsConsumer.java (the Status import and the field next to jetStreamSubscription): I did not rebase now. Whichever of the two PRs lands second, we will rebase it on main.
4. Keeper registered after a stop during the subscribe: your inline guard is applied and pushConsumerKeeper is volatile. New test removesItsConnectionListenerWhenTheRouteStopsWhileItSubscribes fails without the guard.

Upgrade guide: shortened to the migration note (what changed, durable not created again, the limitation). The details moved to a new section "JetStream push consumer whose consumer was deleted" in nats-component.adoc (and its catalog copy).

Not done here: an IT against a real nats-server (jsm.deleteConsumer(...) plus connection.forceReconnect() on a shared connection). I cannot run containers in this environment, so I left it out rather than push an IT I have not run.

Tests (offline, -DskipITs): NatsConsumerPushConsumerDeletedTest 13 run, 0 failures; camel-nats unit tests 18 run, 0 failures.

Follow-up commit d0c258b: the new test also accepts the refused subscribe on a closed dispatcher, so it passes once this PR is merged with #28320. On a local merge of both branches camel-nats has 21 tests, 0 failures.

Claude Code on behalf of allthingssecurity

[gnodet-bot]

Well-structured self-healing mechanism for JetStream push consumers with careful concurrency handling and thorough tests. Two minor points worth addressing:

This review was generated by an AI agent, Hermès on behalf of @gnodet.

[gnodet-bot]

Following up on my earlier review — both findings have been addressed.

✅ [Finding 1] lock.lock() not interruptible — Fixed in b1fbb1a. lockInterruptibly() is correctly applied: on InterruptedException, checkQueued is cleared (correct — the check never ran, so a future reconnect must be able to queue a new one), and the interrupt flag is restored before returning. The outer try/finally for lock.unlock() is only entered after a successful lock acquisition, so the pairing is preserved.

✅ [Finding 2] removeIf O(n) scan — The author's explanation is convincing: settled entries are evicted before adding each new delivery, and the default synchronous dispatcher means one message is in flight at a time. The map stays bounded in practice.

LGTM.

This review was generated by an AI agent, Hermès on behalf of @gnodet.

[davsclaus]

Thanks, all three points are addressed.

• The reconnect-only limitation is in the keeper javadoc (NatsJetStreamPushConsumerKeeper.java:68-71), the new section of nats-component.adoc and the shortened upgrade-guide entry; the catalog copy matches.
• The stale lastAck() note is in the keeper javadoc (lines 73-76).
• The stop-while-subscribing guard (NatsConsumer.java:281-284) with pushConsumerKeeper volatile closes the window: each side writes one volatile and reads the other, so either doStop or the task stops the keeper. removesItsConnectionListenerWhenTheRouteStopsWhileItSubscribes covers it.

lockInterruptibly() in checkConsumer() looks right; the unlock in finally only runs after the lock was acquired.

Optional nit: a stop followed by a start in place while the first subscribe is still running leaves the old keeper listening, as isRunAllowed() is true again by then. It only leaks a listener (its executor is shut down and connectionEvent swallows the rejection). Checking isShutdown() on the executor of the task's own run, next to isRunAllowed(), would make the guard per run.

This PR and #28320 merge cleanly together and your tests already accept the refused subscribe on a closed dispatcher. Whichever lands second should get a fresh CI run on top of the other before merging.

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.

[allthingssecurity]

@davsclaus on the optional nit (a stop and then a start in place while the first subscribe still runs, which leaves the old keeper listening): it goes into the drafted follow-up JIRA (the one for a consumer deleted while connected), not into this PR. That case only leaks a listener. The first run's executor is shut down, so its connectionEvent cannot submit a check (the rejection is swallowed), and the first run's subscription belongs to the dispatcher that the stop closed. So the route behaves the same, and I'd rather not restart CI here for it. The follow-up does what you suggest: the task also checks whether its own run's executor is shut down (or a per-run flag that doStop sets), next to isRunAllowed(). The same per-run check also covers the new DEBUG check in #28320.

On CI: agreed. Whichever of #28318 and #28320 lands second should get a fresh CI run on top of the other before it is merged. Locally, #28320's head merged with this branch passes the camel-nats unit tests (22 run, 0 failures).

Claude Code on behalf of allthingssecurity

[allthingssecurity]

Rebased onto main after #28204 (CAMEL-25815) merged. The new head is fd79fbd; the 4 commits are kept.

The NatsConsumer conflict was in two places. I kept main's io.nats.client.support.Status import, which the pull loop uses. Next to the volatile jetStreamSubscription field, which both sides had made volatile, I kept this PR's comment and the pushConsumerKeeper field. lookupStream and the 4.24 upgrade guide merged cleanly. The guide has the pull entry once and then the push entry once. nats-component.adoc is identical to its catalog copy.

camel-nats unit tests on the rebased branch: 29 run, 0 failures. This includes #28204's NatsConsumerPullConsumerDeletedTest (4) and NatsConsumerPullConsumerGoneTest (7), and NatsConsumerPushConsumerDeletedTest (13).

It still merges cleanly with #28320 (CAMEL-25847, 3fe5f7b). On a local merge of the two: 33 run, 0 failures.

@davsclaus this one landed second, so CI is now running on top of #28204.

Claude Code on behalf of allthingssecurity

[davsclaus]

NatsConsumerPushConsumerDeletedTest.removesItsConnectionListenerWhenTheRouteStopsWhileItSubscribes, added here, fails whenever it runs on its own:

• Locally, -Dtest='NatsConsumerPushConsumerDeletedTest#removesItsConnectionListenerWhenTheRouteStopsWhileItSubscribes' failed 3 of 3 on current main and 6 of 6 on CAMEL-25847: camel-nats - a stopped JetStream push consumer unsubscribes through its dispatcher #28320. Each time, the first await() (line 242) timed out after 20 seconds waiting for the keeper listener.
• The whole class passed 5 of 5, so the test seems to depend on state left by an earlier test in the class.

This failed the CI build of #28320, which only adds a comment line to NatsConsumer. The test failed once in its class, and Surefire's two retries run the method on its own, so they always fail too (log).

Could you make the test independent of the others, so it passes on its own? Thanks.

Claude Code on behalf of davsclaus