[
https://issues.apache.org/jira/browse/CAMEL-25845?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen resolved CAMEL-25845.
---------------------------------
Resolution: Fixed
Fixed by https://github.com/apache/camel/pull/28318
> camel-nats - the JetStream push consumer stops consuming for good when the
> server deleted its ephemeral consumer (after a disconnect of more than 5 s or
> a server restart)
> --------------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-25845
> URL: https://issues.apache.org/jira/browse/CAMEL-25845
> Project: Camel
> Issue Type: Bug
> Components: camel-nats
> Reporter: shashank
> Assignee: shashank
> Priority: Major
> Fix For: 4.24.0
>
>
> With {{jetstreamEnabled=true}} and {{pullSubscription=false}} the consumer
> subscribes with a JetStream push subscription, and without {{durableName}}
> the JetStream consumer behind it is ephemeral
> ({{NatsConsumer.setupJetStreamConsumer}}, main 20cdb309f7d0, lines 240-262).
> The NATS server deletes an ephemeral push consumer when nobody is subscribed
> to its deliver subject for its inactive threshold (nats-server
> {{consumer.go}}: on lost interest an ephemeral consumer is deleted after
> {{JsDeleteWaitTimeDefault}}, 5 seconds by default). A connection outage of
> more than about 5 seconds does exactly that (jnats reconnects for about 2
> minutes by default), and a server restart loses an ephemeral consumer as well.
> jnats does not create a push consumer again. After the reconnect it
> subscribes the deliver subject (an inbox) again and fires {{RESUBSCRIBED}},
> but nothing delivers to that subject any more: {{PushMessageManager}} of
> jnats 2.26.3 has no recovery (only the ordered consumer recreates its
> consumer), and Camel sets no idle heartbeat, so not even a heartbeat alarm is
> raised. The route stays started, nothing is consumed again and nothing is
> logged. The messages stay in the stream until the route is restarted, which
> then creates a new ephemeral consumer from its deliver policy (by default the
> whole stream, so everything is routed again).
> This is the push twin of CAMEL-25815 (pull subscription). The fix for that
> issue only changes the pull loop.
> *Fix*: a small helper ({{NatsJetStreamPushConsumerKeeper}}) is the message
> handler of the push subscription 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 to {{null}}), so {{null}} counts
> as gone (as does a 404 error). When an ephemeral consumer is gone it
> subscribes again, starting at the first stream sequence delivered to the
> route and not acknowledged (it keeps the delivered messages and checks their
> {{lastAck()}}; a nak'ed message counts as not acknowledged), or after the
> last one delivered, so that nothing is skipped and the stream is not replayed
> from the deliver policy; the new configuration carries no start time (the
> server rejects a start time next to a start sequence). With
> {{AckPolicy.None}} the delivery is the acknowledgement: no message is kept,
> and the new consumer starts after the last one delivered. A message delivered
> {{maxDeliver}} times (when set) counts as settled even when it was nak'ed, as
> the server does not deliver it again ({{deliveredCount()}}), so a poison
> message does not hold the resume point back for the life of the route. 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, so the messages published during the outage are not skipped.
> Before subscribing again it looks the stream up: a stream with {{WorkQueue}}
> retention accepts only the deliver policy {{All}} (error 10101 otherwise) and
> has already removed the acknowledged messages, and a stream whose create time
> differs from the one seen at the subscribe was created again (its sequences
> start at 1 again, so a start sequence of the old stream would skip its
> messages); in both cases the consumer configuration is used as it is (for a
> re-created stream after clearing what was tracked, with a WARN). A durable
> consumer 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 keeper
> logs a WARN and keeps the subscription. 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
> (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 (jnats throws {{IllegalStateException}} for some, such as SUB-90007
> when the stream is gone) goes to the consumer's exception handler, the next
> ones are logged at DEBUG; a subscribe that fails because the consumer is
> stopping is not reported. The listener is removed when the consumer stops,
> and a subscription the keeper makes while the consumer is being stopped is
> ended again by the keeper. The checks run only after a reconnect, so an idle
> healthy consumer sends no extra requests. No option changes. A 4.24
> upgrade-guide entry (like the one of CAMEL-25815) describes the ephemeral
> case, {{ackPolicy=None}}, {{maxDeliver}}, {{New}}/{{Last}}, WorkQueue and
> re-created streams, the retries, and that a durable consumer is not created
> again.
> *Tests*: new {{NatsConsumerPushConsumerDeletedTest}} (12 tests) with a fake
> JetStream server behind the jnats interfaces the consumer calls
> ({{Connection}}, {{JetStream}}, {{Dispatcher}}, {{JetStreamSubscription}},
> {{Message}}): a push consumer delivers on its own thread, a disconnect
> deletes the ephemeral consumers and drops the acknowledgements sent to them,
> and a reconnect fires {{RESUBSCRIBED}} as jnats does.
> {{consumesAgainAfterTheServerDeletedTheEphemeralConsumer}} fails on main with
> {{mock://result Received message count. Expected: <2> but was: <0>}};
> {{redeliversTheMessageWhoseNakWasLostWithTheConsumer}} (the route fails a
> message while the connection drops, so its nak is lost with the consumer)
> fails on main with the same message. Both pass with the fix, and the message
> acknowledged before the deletion is not routed again.
> {{endsTheSubscriptionItMadeWhileTheRouteStopped}} stops the route while the
> consumer subscribes again; it fails without the stop check of the keeper
> ({{the new subscription is ended ==> expected: <true> but was: <false>}}) and
> passes with it. {{doesNotCreateAGoneDurableConsumerAgain}} ({{durableName}}
> set, the durable is deleted during the outage: no new consumer is created),
> {{reportsAFailedSubscribeAndTriesAgain}} (the first subscribe after the
> reconnect throws {{IllegalStateException}}: the endpoint's exception handler
> gets it and the next attempt routes the message) and
> {{resumesAnEphemeralConsumerThatStartedAtATime}}
> ({{DeliverPolicy.ByStartTime}}; the fake server rejects a start time next to
> {{ByStartSequence}}, as nats-server does). The fake server now answers like
> jnats ({{getConsumerInfo()}} returns {{null}} for a deleted consumer and the
> info for a live one); against the previous version of the branch 8 of the 9
> tests then fail, as nothing is routed after the reconnect.
> {{workQueueStreamIsSubscribedAgainWithDeliverPolicyAll}} (the fake refuses a
> consumer that is not deliver-all on a work-queue stream, 10101, and removes
> acknowledged messages), {{consumesAStreamCreatedAgainFromItsStart}} (the
> stream is created again during the outage, the next message is sequence 1)
> and {{ackPolicyNoneResumesAfterTheHighestDelivered}}
> ({{manualAck=true&ackPolicy=None}}, the route never acks) still fail with
> only the {{null}} fix ({{Expected: <1> but was: <0>}} twice, {{Expected:
> <order-3> but was: <order-1>}}) and pass with the change. Three more tests
> fail on the earlier version of the fix:
> {{aMessageDeliveredMaxDeliverTimesDoesNotHoldBackTheResume}}
> ({{maxDeliver=3}}, a poison message nak'ed three times; {{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; the check gave up after five)
> and
> {{deliverPolicyNewWithNothingDeliveredResumesAfterTheLastMessageAtTheSubscribe}}
> ({{Expected: <1> but was: <0>}}). The fake now throws 404 stream not found
> while the stream is gone, as jnats does. camel-nats unit tests with the fix:
> 17 run, 0 failures.
> *How it was found*: a variant hunt for the CAMEL-25815 pattern (a server-side
> object the client library does not recreate), then a JUnit test with a fake
> JetStream server. The server rule is that of nats-server {{consumer.go}}, the
> client behaviour that of jnats 2.26.3 (read from the classes). Not run
> against a real NATS server (the integration tests need a container).
> *Affected*: main, camel-4.23.x, camel-4.22.x and camel-4.18.x (the same push
> subscription without any recovery in {{setupJetStreamConsumer}}; read, not
> run).
> _Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
