[
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). When the
consumer is gone (404) it subscribes again: an ephemeral consumer is created
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;
a durable consumer is created again from its configuration. 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) says that a deleted
durable consumer is created again from its configuration, so with the default
{{DeliverPolicy.All}} the route consumes the stream again.
*Tests*: new {{NatsConsumerPushConsumerDeletedTest}} (3 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. camel-nats unit tests with the fix: 8 run, 0 failures; together with
the CAMEL-25815 and CAMEL-25847 fixes: 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._
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 the
consumer is gone (404) it subscribes again: an ephemeral consumer is created
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;
a durable consumer is created again from its configuration. The listener is
removed when the consumer stops. No option changes.
*Tests*: new {{NatsConsumerPushConsumerDeletedTest}} (2 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. camel-nats unit tests
with the fix: 7 run, 0 failures; together with the CAMEL-25815 fix: 9 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). When the consumer is gone (404) it subscribes again: an ephemeral
> consumer is created 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; a durable consumer is created again from
> its configuration. 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) says that a deleted durable consumer is created again
> from its configuration, so with the default {{DeliverPolicy.All}} the route
> consumes the stream again.
> *Tests*: new {{NatsConsumerPushConsumerDeletedTest}} (3 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. camel-nats unit tests with the fix: 8 run, 0 failures;
> together with the CAMEL-25815 and CAMEL-25847 fixes: 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._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
