[
https://issues.apache.org/jira/browse/CAMEL-26459?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Work on CAMEL-26459 started by shashank.
----------------------------------------
> camel-disruptor - multipleConsumers: when a consumer fails, a consumer that
> gets to the message later routes an already failed copy and never processes it
> ----------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-26459
> URL: https://issues.apache.org/jira/browse/CAMEL-26459
> Project: Camel
> Issue Type: Bug
> Components: camel-disruptor
> Reporter: shashank
> Assignee: shashank
> Priority: Major
>
> Found in the review of CAMEL-26458 (camel-disruptor, {{multipleConsumers}}:
> on completions run twice).
> With {{multipleConsumers=true}} each consumer routes its own copy of the
> published exchange: {{DisruptorConsumer.prepareExchange}} copies the
> published exchange when that consumer gets to the event. When a consumer
> fails, {{MultipleConsumerSynchronizedExchange.consumed}} completes the
> exchange right away (early completion) and copies the failure into the shared
> published exchange first ({{ExchangeHelper.copyResults(getExchange(),
> result)}}). A consumer that has not started on that event yet (its thread is
> still busy with an earlier event) then copies an exchange that already has
> the exception, so its route stops after the first step: that consumer never
> processes the message, and nothing is logged as a failure of that consumer.
> {code:java}
> from("disruptor:fan?multipleConsumers=true") // consumer 1: fails m2 at
> once
> .filter(body().isEqualTo("m2")).throwException(new
> IllegalArgumentException("Forced"));
> from("disruptor:fan?multipleConsumers=true") // consumer 2: busy with m1
> for a while
> .process(e -> { /* slow for m1 */ })
> .to("mock:c2"); // never gets m2
> {code}
> Reproduced with latches (no sleeps): consumer 2 waits on a latch while
> routing m1; consumer 1 fails m2 and the caller's on completion reports the
> failure of m2; then consumer 2 is released: it routes m1 and never m2 (its
> copy of m2 already has the exception, so its route stops before {{mock:c2}}).
> A SPIN model of the two consumers (the published exchange written by the
> early completion, copied later by the other consumer) shows the same order,
> and that running the early completion on a copy of the published exchange
> keeps the copies clean.
> The early completion also runs the on completions of the caller's exchange
> (such as the rollback of a file consumer: read lock and in-progress mark
> released) while the other consumers are still routing the message, so the
> source may hand the same message out again meanwhile. camel-seda completes
> the exchange of {{multipleConsumers}} once, after all consumers are done.
> With a waiting producer (InOut) the producer has already been answered by
> then, so a late consumer skips the message (the copy is marked to be ignored)
> instead of routing a failed copy.
> Same on main before and after CAMEL-26458.
> *Proposal* (to discuss): either complete the exchange once when all consumers
> are done (as seda; a waiting producer then gets the failure when all
> consumers are done, bounded by its timeout), or keep the early completion but
> run it on a copy of the published exchange, so that the published exchange is
> not written after it was published and later consumers copy the message as it
> was sent. Tests: a consumer that is busy with an earlier message when another
> consumer fails the next one still routes it.
> _Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
