[
https://issues.apache.org/jira/browse/CAMEL-25847?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18131798#comment-18131798
]
Claus Ibsen commented on CAMEL-25847:
-------------------------------------
Fixed by https://github.com/apache/camel/pull/28320 (merged to main as
187bcde98c2c).
> 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
> Fix For: 4.24.0
>
>
> 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}} ends the push subscription through its dispatcher
> ({{dispatcher.unsubscribe(jetStreamSubscription)}}, which jnats accepts for a
> subscription of that dispatcher and which does not touch the dispatcher
> thread), and closes the dispatcher the consumer created
> ({{connection.closeDispatcher}}, while the connection is open), which ends
> its thread, once no message is in flight on it: jnats closes a dispatcher by
> cancelling its thread with an interrupt ({{NatsDispatcher.stop(true)}},
> {{Future.cancel(true)}}), and that thread runs the route, which Camel stops
> before it waits for the exchanges in flight. When a message is in flight at
> the stop, the dispatcher is closed when it is done. 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). {{stoppingTheRouteDoesNotInterruptTheExchangeInFlight}}
> stops the route while a step waits; with the dispatcher closed in {{doStop}}
> the step is interrupted ({{expected: <false> but was: <true>}}), with the fix
> it completes and the dispatcher is closed afterwards (the fake interrupts the
> delivering thread on {{closeDispatcher}}, as jnats does). 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: 7 run, 0 failures; merged with the CAMEL-25845 fix:
> 19 run, 0 failures.
> *How it was found*: while writing the CAMEL-25845 fix, reading the jnats
> classes (2.26.3, the version of Camel main)
> ({{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
(v10.3.26#10030026)
