[
https://issues.apache.org/jira/browse/CAMEL-26457?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Work on CAMEL-26457 started by shashank.
----------------------------------------
> 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
> Priority: Major
>
> 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)
