[jira] [Work started] (CAMEL-25847) camel-nats - a JetStream push consumer on a shared connection is not unsubscribed when the route stops: the stopped route keeps getting messages and after a restart every message is routed twice

["shashank (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/CAMEL-25847?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Work on CAMEL-25847 started by shashank.
----------------------------------------
> camel-nats - a JetStream push consumer on a shared connection is not unsubscribed when the route stops: the stopped route keeps getting messages and after a restart every message is routed twice
> --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
>                 Key: CAMEL-25847
>                 URL: https://issues.apache.org/jira/browse/CAMEL-25847
>             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 that belongs to a dispatcher it creates ({{NatsConsumer.setupJetStreamConsumer}}, main 20cdb309f7d0, lines 240-262). {{doStop}} (lines 93-133) then does two things that do not end that subscription:
> * {{dispatcher.unsubscribe(configuration.getTopic())}}: a push subscription is on its deliver subject (an inbox), not on the topic, so nothing is unsubscribed;
> * {{jetStreamSubscription.unsubscribe()}}: jnats refuses it for a subscription that belongs to a dispatcher ({{IllegalStateException}} "Subscriptions that belong to a dispatcher cannot respond to unsubscribe directly."), which only ends in a WARN "Error during unsubscribing".
> The dispatcher is never closed. When the consumer owns its connection the connection is closed right after, which ends everything. With a shared connection (the {{connection}} endpoint option, an already instantiated {{Connection}}), the connection stays open and the subscription stays active:
> * while the route is stopped, messages keep being delivered to the stopped consumer; the exchanges fail and are nak'ed, and the server delivers them again at once (a busy loop for as long as the route is stopped; with {{maxDeliver}} set the messages are then dropped, with {{ackPolicy=None}} they are acknowledged and lost);
> * when the route is started again, a second subscription is made, and every message is routed twice for as long as the application runs (with an ephemeral consumer both consumers get every message).
> In plain NATS mode with a shared connection the topic is unsubscribed, but the dispatcher (one thread) of every stopped consumer is left behind.
> *Fix*: {{doStop}} closes the dispatcher the consumer created ({{connection.closeDispatcher}}) when the connection is not closed. That unsubscribes all its subscriptions, the JetStream push subscription included, and ends its thread. The existing unsubscribe calls are kept. No option changes.
> *Tests*: new {{NatsConsumerJetStreamPushStopTest}} with a fake JetStream server behind the jnats interfaces the consumer calls, which, like jnats, refuses {{unsubscribe()}} on a subscription of a dispatcher and ends the subscriptions of a closed dispatcher. {{stoppedRouteGetsNoMessagesAndRestartedRouteGetsEachMessageOnce}} (shared connection, push, stop and start the route, publish one message) fails on main with {{mock://result Received message count. Expected: <1> but was: <2>}}, and passes with the fix (no subscription left after the stop, one after the start). A probe on main with the same fake, publishing while the route is stopped, saw the stopped route fail and nak the message over a million times in 2 seconds. camel-nats unit tests with the fix: 6 run, 0 failures; merged with the CAMEL-25815 and CAMEL-25845 fixes: 10 run, 0 failures.
> *How it was found*: while writing the CAMEL-25845 fix, reading the jnats 2.26.4 classes ({{NatsSubscription.unsubscribe}}), then a JUnit test with a fake JetStream server. Not run against a real NATS server.
> *Affected*: main, camel-4.23.x, camel-4.22.x and camel-4.18.x (the same {{doStop}}; read, not run).
> _Filed with Claude Code on behalf of allthingssecurity._



--
This message was sent by Atlassian Jira
(v8.20.10#820010)