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