allthingssecurity opened a new pull request, #28318: URL: https://github.com/apache/camel/pull/28318
# Description [CAMEL-25845](https://issues.apache.org/jira/browse/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`, so whichever merges second needs a small rebase (the volatile field line and one import line conflict; the `lookupStream` lines are identical on both sides and merge cleanly). Changes (`NatsConsumer`, new package-private `NatsJetStreamPushConsumerKeeper`, upgrade guide): - 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.3, the version of Camel main) 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. 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): an entry like the one of CAMEL-25815 for the pull consumer, about the ephemeral consumer, `ackPolicy=None`, `maxDeliver`, `New`/`Last`, WorkQueue and re-created streams, the retries, and that a durable one is not created again (WARN). It sits in another place of the guide than the CAMEL-25815 entry. 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. Tests: - New `NatsConsumerPushConsumerDeletedTest` (12 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 (ffc2d94d1d, 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 (57555dbbbd): `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>`. - Merged locally with CAMEL-25847 (stop unsubscribes): 19 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: 17 run, 0 failures (formatter and import sort in the build). It conflicts with the current pull twin (#28204, after its review commits) in `NatsConsumer`; whichever merges second needs a rebase. 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-25845](https://issues.apache.org/jira/browse/CAMEL-25845). # 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; the upgrade guide has a new entry. 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]
