shashank created CAMEL-25845:
--------------------------------
Summary: 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
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._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
