shashank created CAMEL-26458:
--------------------------------
Summary: 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
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)
