shashank created CAMEL-26462:
--------------------------------
Summary: camel-nats - a consumer that stops while its task
subscribes can leave its dispatcher (and its subscription) open on a shared
connection
Key: CAMEL-26462
URL: https://issues.apache.org/jira/browse/CAMEL-26462
Project: Camel
Issue Type: Bug
Components: camel-nats
Reporter: shashank
Assignee: shashank
Follow-up of CAMEL-26366 and CAMEL-25847, found while working on the flaky
tests of CAMEL-26366.
The JetStream push consumer subscribes on its own executor
({{NatsConsumingTask}}), after {{doStart}} returned. Since CAMEL-25847
{{doStop}} ends the push subscription through the dispatcher of the consumer
and closes that dispatcher (once no message is in flight), which matters when
the connection is shared ({{connection=#conn}}) and stays open. But {{doStop}}
only sees the dispatcher once the task has set it ({{NatsConsumer.dispatcher}},
which is not volatile either). When the consumer stops after the task created
the dispatcher and before it set it, {{doStop}} finds nothing to close, shuts
the executor down (which interrupts the task) and returns. The task then goes
on:
* the push subscribe asks the server first (jnats
{{NatsJetStreamImpl.makeRequestResponseRequired}}), so on the interrupted
thread it fails with an {{IOException}}; the task logs it at DEBUG (the
consumer is stopping) and the dispatcher it created stays open, with its
thread, on the shared connection, until the connection closes. Each such stop
leaks one.
* if the subscribe is made anyway (the interrupt of {{doStop}} comes after it:
{{doStop}} reads the dispatcher before it shuts the executor down), the task
sees {{isRunAllowed() == false}} and stops the keeper of CAMEL-25845
({noformat}stopped while subscribing: doStop did not see the keeper{noformat}),
but not the dispatcher: the subscription stays active and the stopped route
keeps getting messages from the shared connection; the task also marks the
consumer active (which matters for a first run only).
The consumer without JetStream ({{setupStandardNatsConsumer}}) has the same
window: its subscribe sends no request, so it is made after such a stop, and
its dispatcher gets the messages of the subject from the shared connection.
*Proposal*
* Right after the task set the dispatcher of its run, and before it subscribes,
check the state of the consumer; when it is stopping, close that dispatcher (no
subscription yet) and return. With the field {{NatsConsumer.dispatcher}}
volatile, the task sees the stop, or {{doStop}} sees the dispatcher and closes
it on the stopping thread (a subscribe on the closed dispatcher then fails), or
both; in that last case {{doStop}} must not report the failed unsubscribe of
the closed dispatcher ({{Dispatcher.isActive()}} is false). The same in the
consumer without JetStream.
* Do not close the dispatcher from the task after its subscribe: {{doStop}}
interrupts the task, and jnats drops the UNSUB that a close queues on an
interrupted thread ({{WriterMessageQueue.push}}), so the server would keep the
subscription.
* Tests in {{NatsConsumerJetStreamPushStopTest}} (fake server, no container,
faithful to jnats on an interrupted thread): the consumer stops right after its
task created the dispatcher, for the push consumer and a consumer without
JetStream, and a stop that waits in its flush until the task closed its
dispatcher; the stopped run does not subscribe, its dispatcher is closed,
nothing is reported to the exception handler.
Not part of this: a JetStream pull subscription made after such a stop gets no
messages (the fetch loop does not run), but is not unsubscribed; and a quick
stop then start, where the task of the stopped run sees the consumer running
again (that needs a token per run).
_Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
