[
https://issues.apache.org/jira/browse/CAMEL-26458?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen resolved CAMEL-26458.
---------------------------------
Resolution: Fixed
> camel-disruptor - multipleConsumers: when a consumer fails, the on
> completions of the exchange run twice, or as a success and as a failure
> ------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-26458
> URL: https://issues.apache.org/jira/browse/CAMEL-26458
> Project: Camel
> Issue Type: Bug
> Components: camel-disruptor
> Reporter: shashank
> Assignee: shashank
> Priority: Major
> Fix For: 4.24.0
>
>
> Found with a SPIN (Promela) model of the completion of a {{disruptor}}
> exchange by several consumers ({{multipleConsumers=true}}), and reproduced on
> the real classes.
> *What happens*
> With {{multipleConsumers=true}} each consumer routes its own copy of the
> published exchange and then calls
> {{MultipleConsumerSynchronizedExchange.consumed(result)}}. For an InOnly
> exchange the producer hands the on completions of the caller's exchange over
> to the published copy ({{prepareCopy(exchange, true)}}), so these on
> completions (the rollback or commit of the consumer the message came from,
> such as a file or SQS consumer, the {{Synchronization}}s added by the route)
> run when {{consumed}} calls {{performSynchronization()}}:
> {code:java}
> if (processedConsumers.incrementAndGet() == expectedConsumers ||
> result.getException() != null
> && !resultHandled.getAndSet(true)) {
> if (result.getException() != null) {
> ExchangeHelper.copyResults(getExchange(), result);
> }
> performSynchronization();
> }
> {code}
> The {{resultHandled}} guard only applies to the failure branch ({{&&}} binds
> tighter than {{||}}), so the last consumer always synchronizes, also when a
> consumer that failed earlier already did:
> # a consumer fails while another one is still routing: the failure
> synchronizes (onFailure), then the last consumer is done and synchronizes
> again: the on completions run twice ({{onFailure}} twice, the copy still has
> the exception), also when both fail;
> # a consumer fails and the last consumer is counted before the failing one
> reaches its {{getAndSet}}: the last consumer synchronizes without the failure
> ({{onComplete}}, e.g. the file is moved to {{.camel}}), then the failing one
> synchronizes with it ({{onFailure}} as well).
> Effects: the rollback of the consumer runs twice, or a commit and a rollback
> both run for one message (a file consumer may release the read lock and
> in-progress mark of a file that is picked up again meanwhile, an
> {{onCompletion().onFailureOnly()}} runs twice, custom on completions see two
> outcomes). With a waiting producer ({{waitForTaskToComplete}}, InOut) the
> producer's own on completion is on the copy: it ignores the second call, but
> in the second order it runs first without the failure, so the caller gets a
> successful reply although a consumer failed.
> *SPIN model* (consumers C = 2 and 3, each finishes with success or failure in
> any order; properties: on completions run once, and with the failure if a
> consumer failed). OnceSync is violated (2 consumers, shortest trail at depth
> 33; the 3-consumer run finds it too):
> {noformat}
> c1: routes its copy, fails c2: routes its copy, succeeds
> c1: incrementAndGet() = 1 (not last) -> failure branch
> c2: incrementAndGet() = 2 = expected -> synchronize (resultHandled not looked
> at)
> c1: resultHandled.getAndSet(true) = false -> copyResults(failure) ->
> synchronize
> c2: performSynchronization #1 c1: performSynchronization #2 ->
> violation
> {noformat}
> The fix below holds both properties for 2 to 5 consumers (5 consumers: 22
> million states, 8 s; also without partial order reduction for 3 and 4
> consumers); a witness shows the failure path and the all-success path are
> reachable.
> *Reproducer* (JUnit, latches, no sleeps): {{from("direct:start")}} adds a
> {{Synchronization}} and sends InOnly to
> {{disruptor:fanout?multipleConsumers=true}}; one consumer route throws at
> once, the other waits on a latch: after the latch is released, {{onFailure}}
> has run twice on main (once with the fix); also when both consumers fail. A
> consumer that fails after the other one is done, and all consumers
> succeeding, synchronize once (unchanged). The second order is forced in a
> unit test of {{consumed()}}, where the failing consumer is held at its first
> look at the exception while the other consumer is done: {{onComplete}} then
> {{onFailure}} on main. A stress run (two consumers that finish at the same
> moment, one fails, 2000 messages) gave on main 918 double {{onFailure}} and
> 37 {{onComplete}} + {{onFailure}}; with the fix 2000 single {{onFailure}}.
> *Proposal*
> A consumer that failed sets a flag before it is counted, and synchronizes
> with its own result if it is the first to claim {{resultHandled}}; the last
> consumer synchronizes only when it claims {{resultHandled}} and no consumer
> failed. The early completion on the first failure stays as it is (the waiting
> producer gets the failure without waiting for the other consumers). Setting
> the flag before counting matters: a fix that only adds the once-guard, or
> sets the flag after counting, still lets the last consumer complete the
> exchange as a success while another consumer failed (SPIN finds that order on
> such a fix).
> Related, not part of this: on the first failure the failure is copied into
> the shared published exchange, so a consumer that gets to that message later
> routes an already failed copy and never processes it (a separate follow-up).
> _Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
