[
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}} 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._
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: 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._
> 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: 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._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
