shashank created CAMEL-26457:
--------------------------------
Summary: camel-disruptor - disruptor-vm: CamelContexts that create
or stop their endpoints of a queue at the same time can end up on different
ring buffers, so the messages never reach the consumer
Key: CAMEL-26457
URL: https://issues.apache.org/jira/browse/CAMEL-26457
Project: Camel
Issue Type: Bug
Components: camel-disruptor
Reporter: shashank
Assignee: shashank
Found with a TLA+ model of the disruptor-vm queue map shared by the
CamelContexts of a JVM (two CamelContexts, one of them stopping and starting
again).
The {{disruptor-vm}} component keeps the queues of all the CamelContexts of the
JVM in one static {{java.util.HashMap}} ({{DisruptorVmComponent.DISRUPTORS}}).
But it guards it with the lock of each component instance
({{DisruptorComponent.getOrCreateDisruptor}} uses the {{BaseService}} lock),
and each CamelContext has its own component instance, so the CamelContexts do
not exclude each other:
# *Concurrent creation.* Two CamelContexts that create their endpoint of the
same queue at the same time both find no queue in the map, both create one and
both put it (the second put wins; the map is a plain {{HashMap}}). Short trace:
C1 get -> null, C2 get -> null, C1 put R1, C2 put R2: the producer of C1
publishes to R1 and the consumer of C2 consumes from R2.
# *Stop of the component.* {{DisruptorVmComponent.doStop}} clears the whole map
when the last *started* vm component stops. A CamelContext that is not started
yet (or is starting) has its endpoints already created on a queue of the map,
but its component is not started, so the queue is dropped under it; the next
endpoint of the queue (another CamelContext, or a redeployed one) gets a new
ring buffer.
# *Shutdown of an endpoint.* {{DisruptorComponent.onShutdownEndpoint}} looks up
the queue by key and removes it without a lock when its endpoint count is 0.
The count only includes initialized endpoints, so an endpoint that is being
created in another CamelContext (it already took the queue from the map) does
not keep it, and if the queue is created again meanwhile the two endpoints use
different ones.
A hook-free probe that creates the endpoint of a new queue in two CamelContexts
at the same moment got two ring buffers in 1998 of 2000 runs (the window is the
creation of the disruptor itself).
In every case the producer and the consumer of the queue use different ring
buffers: the messages are never consumed (they are lost when the CamelContext
stops; with {{blockWhenFull}} the producer blocks once the ring buffer is
full). An InOnly producer gets no error; an InOut one times out.
Model: {{SameRef}} (every running endpoint of a key uses the same ring buffer)
is violated in 9 steps (concurrent creation) and 18 steps (a CamelContext stops
and starts again while another one creates its endpoint); with the change below
it holds in the model, and the tests below pass.
Reproduced with the real classes (no broker, plain CamelContexts):
* an endpoint created in a CamelContext before it is started, then the only
other CamelContext stops, then a third one starts: the producer and the
consumer are on different disruptors, the message is not received;
* two CamelContexts create their endpoint at the same time (a test subclass
pauses the first one after its lookup found nothing, until the second one
stored its queue): different disruptors;
* a CamelContext creates its endpoint while the CamelContext of the only other
endpoint stops: different disruptors.
*Proposal*
* Access the queue map only under a lock that all the components sharing it
share (a static one for disruptor-vm): lookup and creation, endpoint shutdown,
component stop.
* When an endpoint is initialized (it registers on its reference there
already), use the reference of the key: put its own back when the key has none,
take the one of the key when another CamelContext created a new one meanwhile.
* Remove the reference of the endpoint being shut down ({{remove(key, ref)}})
when no endpoint is registered or being registered on it, and on the stop of
the last vm component remove only the references no endpoint uses. Read those
counts without the lock of the reference, which a reconfiguration holds while
it waits for the exchanges in flight.
_Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
