[
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.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._
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). When an
ephemeral consumer is gone (404) 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). 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. No option changes.
A 4.24 upgrade-guide entry (like the one of CAMEL-25815) describes the
ephemeral case and that a durable consumer is not created again.
*Tests*: new {{NatsConsumerPushConsumerDeletedTest}} (6 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). camel-nats unit tests with the fix:
11 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.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._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
