[
https://issues.apache.org/jira/browse/CAMEL-25849?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen updated CAMEL-25849:
--------------------------------
Fix Version/s: 4.24.0
> camel-disruptor - a disruptor-vm consumer stops consuming for good when the
> CamelContext that first used the queue is stopped
> -----------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-25849
> URL: https://issues.apache.org/jira/browse/CAMEL-25849
> Project: Camel
> Issue Type: Bug
> Components: camel-disruptor
> Reporter: shashank
> Assignee: shashank
> Priority: Major
> Fix For: 4.24.0
>
>
> A {{disruptor-vm}} queue is shared by all CamelContexts of the JVM
> ({{DisruptorVmComponent.DISRUPTORS}} is static), but the threads that run its
> event handlers are always created by the {{ExecutorServiceManager}} of the
> CamelContext whose component created the queue:
> {{DisruptorReference.resizeThreadPoolExecutor}} (main 20cdb309f7d0, lines
> 259-291) uses {{component.getCamelContext().getExecutorServiceManager()}},
> and {{component}} is the component that first resolved the queue name.
> When that CamelContext stops while another CamelContext still consumes from
> the queue, the stopping CamelContext shuts the thread pool down as part of
> its own shutdown ({{BaseExecutorServiceManager}}: "Forcing shutdown of
> ExecutorService ... [disruptor-vm://foo] due first await termination
> elapsed", after 10 seconds, then "Forced shutdown of 1 ExecutorService's
> which has not been shutdown properly (acting as fail-safe)"). The queue
> itself stays registered, as the other CamelContext still has endpoints on it.
> From then on the consumer of the other CamelContext never consumes again: its
> route stays Started, nothing is logged in that CamelContext, and every
> message sent to the queue (from any CamelContext) stays in the ring buffer as
> an inflight exchange. When the ring buffer is full, producers block
> ({{blockWhenFull}}) or fail; when the consuming CamelContext stops, it waits
> the whole shutdown timeout for those inflight exchanges and then drops them.
> A typical case: CamelContext A has
> {{from("direct:in").to("disruptor-vm:orders")}} and is created first,
> CamelContext B has {{from("disruptor-vm:orders")}}. A is stopped or
> redeployed: B's consumer is dead for good. The existing test
> {{DisruptorVmMultipleContextsStartStopTest}} covers a restart of the
> consuming CamelContext, not a stop of the one that created the queue.
> *Fix*: {{DisruptorReference}} creates the thread pool with the CamelContext
> of an endpoint that uses the queue, and remembers it. When the last endpoint
> of that CamelContext is removed (the endpoints are stopped before the
> {{ExecutorServiceManager}}) while endpoints of another CamelContext remain,
> it moves the event handlers to a new thread pool of one of those
> CamelContexts, the same way a reconfiguration does (the backlog is kept and
> published again). With a single CamelContext nothing changes. No option
> changes.
> *Tests*: new {{DisruptorVmFirstContextStopTest}}: CamelContext 1 sends to
> {{disruptor-vm:foo}} (it creates the queue), CamelContext 2 consumes from it
> and also has a route that sends to it. After CamelContext 1 is stopped, a
> message sent from CamelContext 2 must reach CamelContext 2's consumer. On
> main it fails ({{mock://result Received message count. Expected: <1> but was:
> <0>}}; the stop of CamelContext 1 takes 10 s and the stop of CamelContext 2
> waits 45 s for the stuck exchange); with the fix it passes in 0.2 s.
> camel-disruptor unit tests with the fix: 118 run, 0 failures.
> *How it was found*: variant analysis of CAMEL-25846 (an object kept in a
> static, JVM-wide map that depends on a lifecycle owned by one instance), then
> the JUnit test. Not run in an application server with several deployments.
> *Affected*: main, camel-4.23.x, camel-4.22.x and camel-4.18.x (the same
> executor creation in {{DisruptorReference}}; read, not run).
> _Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
