[
https://issues.apache.org/jira/browse/CAMEL-26309?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Work on CAMEL-26309 started by shashank.
----------------------------------------
> 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
> Priority: Major
>
> 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)
