shashank created CAMEL-25849:
--------------------------------
Summary: 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
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
(v8.20.10#820010)
