shashank created CAMEL-26309:
--------------------------------
Summary: camel-core - Aggregate EIP: when the thread pool rejects
an aggregated exchange, the completed group is lost (not recovered, never
confirmed) and stays counted as in progress
Key: CAMEL-26309
URL: https://issues.apache.org/jira/browse/CAMEL-26309
Project: Camel
Issue Type: Bug
Components: camel-core
Reporter: shashank
Assignee: shashank
When the thread pool of the Aggregate EIP rejects the task that sends an
aggregated exchange, the completed group is lost and the aggregator keeps
counting it as in progress forever.
{{AggregateProcessor.onSubmitCompletion}} (camel-core-processor, main
c536c99caaaa) first books the aggregated exchange as in progress
({{inProgressCount.increment()}}, {{inProgressCompleteExchanges.add(id)}}),
calls {{AggregationStrategy.onCompletion}}, adds the {{AggregateOnCompletion}}
that undoes the booking, and only then calls {{executorService.execute(...)}}.
The group has already been removed from the aggregation repository
({{onCompletion}}). If {{execute}} throws {{RejectedExecutionException}} (an
{{executorService}} or thread pool profile with {{rejectedPolicy=Abort}} and a
bounded queue; the Abort handler throws for a task that is not {{Rejectable}},
and the aggregator's task is a lambda), nothing runs the on completions of the
aggregated exchange:
# the in progress booking is never undone: {{getInProgressCompleteExchanges()}}
stays above 0, and with {{forceCompletionOnStop}} or {{completeAllOnStop}} a
later route or CamelContext stop waits in {{inProgressCount.await()}} (no
timeout) until the shutdown strategy times out;
# with a recoverable repository (JDBC, LevelDB, Cassandra, Infinispan, the
key-value one ...) the group stays in the completed store, but the recover task
skips it because its id stays in {{inProgressCompleteExchanges}}: it is never
recovered until the application restarts;
# with the memory repository the group (all its messages but the last) is gone,
and nothing says so: the exchange that completed the group fails with a
{{RejectedExecutionException}} that does not mention the group;
# on completions that the aggregation strategy added to the aggregated exchange
(the zip and tar strategies delete their temporary file there), and the release
of spooled stream caches the group held, never run.
The same happens when the group is completed by the completion timeout, the
interval, a force completion or the recover task (the exception is logged or
retried there, but the group is already removed). When the batch consumer
completes several groups at once ({{completionFromBatchConsumer}}), the groups
after the rejected one are not even submitted: they have been removed from the
repository too and are lost the same way, and with a recoverable repository
they stay marked as being completed, so the recover task skips them as well.
Reproduced on main with a route test (latches, no sleeps):
{{aggregate(header("id"),
strategy).completionSize(2).executorService("oneThread")}} with a profile of 1
thread, queue size 1, {{rejectedPolicy=Abort}}; group A blocks the thread,
group B waits in the queue, group C is rejected. After the thread is released,
A and B are delivered; C is neither delivered nor passed to the exception
handler, its on completion does not run, and
{{getInProgressCompleteExchanges()}} stays 1. With a
{{KeyValueAggregationRepository}} with recovery every 100 ms, C is never
recovered. With {{completionFromBatchConsumer}} and a batch that completes the
groups C, D and E at once, none of them is passed to the exception handler, and
with the recoverable repository neither C nor D is recovered.
The existing {{AggregateParallelThreadPoolAbortTest}} expects the caller to
fail when the pool rejects; that is kept.
*Fix*: when {{execute}} is rejected, fail the aggregated exchange as if its
processing had failed: set the {{RejectedExecutionException}} on it, run its on
completions ({{AggregateOnCompletion.onFailure}} undoes the in progress booking
and does not confirm the exchange, so a recoverable repository recovers it; the
strategy's and the stream cache on completions run), pass it to the
aggregator's exception handler (logged at WARN with the aggregated exchange by
default), and rethrow so the caller fails as before. The other groups of a
batch are still sent (or failed the same way when rejected too) before the
first rejection is rethrown. A recovered exchange that the pool rejects again
is recovered by a later scan; the attempt counts as a redelivery, so
{{maximumRedeliveries}} and {{deadLetterUri}} apply.
Not changed: the default (synchronous) executor never rejects. With Camel's
{{Block}} policy {{execute}} throws only when the pool is shut down or the
waiting caller is interrupted (handled the same way). The JDK
{{CallerRunsPolicy}} behind Camel's {{CallerRuns}} runs the task in the caller,
but drops it without an exception when the pool is shut down; that case is not
covered here.
*Tests*: new {{AggregateRejectedCompletionTest}} (camel-core; latches, no
sleeps; the setup above): {{testRejectedAggregatedExchangeIsFailed}},
{{testRejectedAggregatedExchangeIsRecovered}},
{{testRejectedBatchCompletionFailsEveryGroup}} and
{{testRejectedBatchCompletionIsRecovered}} fail on main (3 of 3 runs) and pass
with the fix. The camel-core tests under {{org/apache/camel/processor}} and the
other aggregator tests ({{*Aggregat*Test}}): 2723 tests, 0 failures, including
{{AggregateParallelThreadPoolAbortTest}} and
{{AggregateParallelThreadPoolCallerRunsTest}}.
*How it was found*: by reading the EIPs that hand work to a thread pool
(Threads, WireTap and Multicast handle a rejection, the aggregator does not)
while writing drivers for a JVM model in Rocq that runs the real class files
with a thread layer and {{ThreadPoolExecutor}} with its rejection policies;
reproduced on HotSpot with the test above. A kernel-checked theorem on the
class files of main states the leak (after a rejected completion and a drained
pool the aggregator still counts the exchange as in progress), and one on the
fix states that the exchange is failed and passed to the exception handler.
*Affected*: main, camel-4.22.x, camel-4.18.x and camel-4.14.x have the same
{{onSubmitCompletion}} (read, not run).
_Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
