[
https://issues.apache.org/jira/browse/CAMEL-26462?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen resolved CAMEL-26462.
---------------------------------
Resolution: Fixed
> 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
> Priority: Minor
> Fix For: 4.24.0
>
>
> 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)
