[
https://issues.apache.org/jira/browse/CAMEL-26575?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Work on CAMEL-26575 started by shashank.
----------------------------------------
> camel-core - Multicast, Split and RecipientList: an AggregationStrategy whose
> onCompletion throws hangs the exchange
> --------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-26575
> URL: https://issues.apache.org/jira/browse/CAMEL-26575
> Project: Camel
> Issue Type: Bug
> Components: camel-core
> Reporter: shashank
> Assignee: shashank
> Priority: Minor
>
> Seen while fixing CAMEL-26464.
> {{MulticastTask.doDone}} (and {{doTimeoutDone}}) first sets the done flag
> ({{done.compareAndSet(false, true)}}) and then calls
> {{MulticastProcessor.doDone}}, which calls
> {{AggregationStrategy.onCompletion(subExchange, original)}} before it
> schedules the callback of the exchange
> ({{reactiveExecutor.schedule(callback)}}, its last statement). When
> {{onCompletion}} throws, the callback is never scheduled. The exception goes
> to the catch of {{aggregateCompleted}} (or of {{timeout}}), whose
> {{doFailed}} / {{doTimeoutFailed}} finds the task done already and only logs
> it at DEBUG ({{Exception occurred after the exchange was already
> completed...}}). With {{stopOnException}}, where the multicast is done from
> the callback of the failed sub exchange, it goes up to the reactive executor,
> which logs and drops it. Either way the exchange hangs: a synchronous caller
> ({{template.send}}, a consumer thread, a {{synchronous()}} multicast) blocks
> for good, the exchange stays inflight and a graceful shutdown waits for it.
> A SPIN model of the completion (the aggregator, the callback of a failed sub
> exchange with {{stopOnException}}, and the timeout, all competing for the
> done flag) confirms it: it is not a race, it hangs every time: once
> {{onCompletion}} throws, no interleaving schedules the callback; with the
> change below it is scheduled exactly once in every interleaving.
> {code:java}
> AggregationStrategy strategy = new AggregationStrategy() {
> public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
> return newExchange; }
> public void onCompletion(Exchange exchange, Exchange inputExchange) {
> throw new IllegalStateException("onCompletion"); }
> };
> from("direct:start").multicast(strategy).to("mock:a",
> "mock:b").end().to("mock:result");
> template.send("direct:start", e -> e.getMessage().setBody("Hello")); //
> never returns
> {code}
> On main the send hangs for the multicast (sequential, {{parallelProcessing}},
> {{synchronous}}, {{stopOnException}}, {{timeout}}), the split (also
> {{streaming}}, also in a {{doTry}}: the {{doCatch}} never runs) and the
> recipient list. The same strategy throwing from {{aggregate}} fails the
> exchange with the exception, as expected. {{onCompletion}} is user code (a
> custom strategy, the completion mixin of {{FlexibleAggregationStrategy}}).
> Its javadoc says it must not throw, and none of Camel's own strategies do, so
> this needs a strategy that breaks that contract; it should still fail the
> exchange, as when {{aggregate}} throws, instead of hanging it with only a
> DEBUG log.
> *Proposal*
> * Catch an exception from {{onCompletion}} in {{MulticastProcessor.doDone}},
> do the rest of the done work and schedule the callback as now, with the
> exchange failed with that exception (set after the results are copied, as
> {{ExchangeHelper.copyResults}} would overwrite it) and not marked as
> exhausted, so the error handler of the route handles it (redelivery,
> {{onException}}, dead letter channel) as when the strategy throws from
> {{aggregate}}. An exchange that has failed already (such as with
> {{stopOnException}}) keeps its exception, which tells where it failed, with
> the new one added as suppressed.
> * {{doDone}} runs once per exchange (it is guarded by the done flag), so the
> callback is still scheduled exactly once; nothing changes on the success path.
> * camel-4.18.x has the same bug.
> _Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
