[
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)
