[
https://issues.apache.org/jira/browse/CAMEL-25847?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
shashank updated CAMEL-25847:
-----------------------------
Description:
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._
was:
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: 11 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._
> 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}} 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)
