[
https://issues.apache.org/jira/browse/CAMEL-25845?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
shashank updated CAMEL-25845:
-----------------------------
Description:
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._
was:
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.4 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 few attempts one second apart when the request fails). jnats 2.26.4
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. 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
subscribe (jnats throws {{IllegalStateException}} for some, such as SUB-90007
when the stream is gone) goes to the consumer's exception handler and is tried
again. 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}}, WorkQueue and
re-created streams, and that a durable consumer is not created again.
*Tests*: new {{NatsConsumerPushConsumerDeletedTest}} (9 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. camel-nats unit tests with the
fix: 14 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.4 (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._
> 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
>
> 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)
