allthingssecurity opened a new pull request, #29038: URL: https://github.com/apache/camel/pull/29038
# Description [CAMEL-26458](https://issues.apache.org/jira/browse/CAMEL-26458) With `multipleConsumers=true` each consumer of a `disruptor` endpoint 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 they run when `consumed` calls `performSynchronization()`: the commit or rollback of the consumer the message came from (file, SQS, ...), and the `Synchronization`s the route added. ```java if (processedConsumers.incrementAndGet() == expectedConsumers || result.getException() != null && !resultHandled.getAndSet(true)) { ``` `&&` binds tighter than `||`, so `resultHandled` only guarded the failure branch and the last consumer always synchronized too: - a consumer fails while another one is still routing: the failure runs the on completions (`onFailure`), and when the last consumer is done it runs them again (`onFailure` a second time, as 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 runs them without the failure (`onComplete`), then the failing one with it (`onFailure` as well). So the rollback of the source consumer runs twice, or a commit and a rollback both run for one message. With a producer that waits (`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. ## The fix A consumer that failed sets a flag (`failed`) before it is counted. A failing consumer completes the exchange with its own result if it is the first to claim `resultHandled`; the last consumer completes it only when it claims `resultHandled` and no consumer failed. So the exchange is completed once, and with a failure if a consumer failed: setting the flag before counting makes the consumer counted last see the failure of any consumer counted before it (both are atomic operations, in that order). Each consumer only copies its own result into the published exchange, as before. The early completion on the first failure is kept: it is the existing behaviour, and a waiting producer gets the failure without waiting for the other consumers. camel-seda completes once all consumers are done; whether disruptor should do the same is part of the follow-up below. `cancelAndGetOriginalExchange` is unchanged. Related, not changed here (same on main): 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. Follow-up: CAMEL-26459. ## How it was found A SPIN (Promela) model of `consumed` with 2 and 3 consumers that finish in any order with success or failure, checking that the synchronization runs once and sees the failure if a consumer failed. SPIN reports both orders above on main; the fix holds for 2 to 5 consumers. A weaker 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 a consumer failed; SPIN finds that order, which is why the flag is set before counting (the ordering is stated in the code comment; the unit test below cannot hold a thread between the two steps). ## Tests New `DisruptorMultipleConsumersOnCompletionTest` (JUnit assertions as in the module, latches, no sleeps): a route sends InOnly to `disruptor:fanout?multipleConsumers=true` after adding a `Synchronization`; one consumer route fails at once, the other waits on a latch. - a failure, then the other consumer succeeds: `onFailure` once (twice on main); - both fail: `onFailure` once (twice on main); - success first, then the failure (control): `onFailure` once; - all succeed (control): `onComplete` once. New `MultipleConsumerSynchronizedExchangeTest` forces the second order of main on `consumed()` itself: the failing consumer is held when `consumed()` first looks at its exception, while the other consumer, which succeeded, is done: `onFailure` once (on main `onComplete` then `onFailure`). The two controls pass on main, the other three tests fail there. The test waits for both consumer exchanges to be done (`NotifyBuilder`), which happens after their on completions, before it checks the outcomes. A stress run outside the PR (two consumers that finish at the same moment through a barrier, one fails, 2000 messages) gave on main 918 double `onFailure` and 37 `onComplete` + `onFailure`, and with the fix 2000 single `onFailure`. camel-disruptor: 127 tests, 0 failures; the new tests 20 times in one JVM: 80/80 and 20/20. Backport: the same code is on camel-4.22.x, camel-4.18.x and camel-4.14.x (a plain bug fix, cherry-picks cleanly onto the file). _Claude Code on behalf of allthingssecurity_ 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
