[
https://issues.apache.org/jira/browse/CAMEL-25845?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Work on CAMEL-25845 started by shashank.
----------------------------------------
> 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. 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)
