allthingssecurity opened a new pull request, #28709: URL: https://github.com/apache/camel/pull/28709
# Description [CAMEL-26309](https://issues.apache.org/jira/browse/CAMEL-26309) 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. `AggregateProcessor.onSubmitCompletion` books the aggregated exchange as in progress (`inProgressCount`, `inProgressCompleteExchanges`), calls `AggregationStrategy.onCompletion`, adds the `AggregateOnCompletion` that undoes the booking, and then calls `executorService.execute(...)`. By then `onCompletion` has removed the group from the aggregation repository. If `execute` throws `RejectedExecutionException` (an `executorService` or a thread pool profile with `rejectedPolicy=Abort` and a bounded queue: Camel's Abort handler throws for a task that is not `Rejectable`, and this task is a lambda), nothing runs the on completions of the aggregated exchange: 1. The in progress booking is never undone. `getInProgressCompleteExchanges()` stays above 0, and with `forceCompletionOnStop` or `completeAllOnStop` a later stop waits in `inProgressCount.await()`, which has no timeout, until the shutdown strategy gives up. 2. With a recoverable repository the group stays in the completed store, but the recover task skips it because its id stays in `inProgressCompleteExchanges`, so it is not recovered until the application restarts. 3. With the memory repository the group is gone. The only sign is the `RejectedExecutionException` of the exchange that completed the group, which does not mention the group or its other messages. 4. On completions that the aggregation strategy added to the aggregated exchange never run (the zip and tar strategies delete their temporary file there), and neither does the release of the spooled stream caches the group held. The same applies when the group is completed by the completion timeout, the completion interval, a force completion or the recover task. When the batch consumer completes several groups at once (`completionFromBatchConsumer`), the groups after the rejected one were not even submitted: they had been removed from the repository as well and were lost the same way, and with a recoverable repository they stayed marked as being completed, so the recover task skipped them too. Changes: - `onSubmitCompletion` catches the `RejectedExecutionException` of `execute` and fails the aggregated exchange as if its processing had failed: the exception is set on it, its on completions run (`AggregateOnCompletion.onFailure` undoes the booking and does not confirm the exchange, so a recoverable repository recovers it later), and it is passed to the aggregator's exception handler (by default a WARN log with the aggregated exchange). Then the exception is thrown again, so the caller fails as before. The existing `AggregateParallelThreadPoolAbortTest` expects that. - When the batch consumer completes several groups at once, each aggregated exchange is still submitted after a rejection (and failed the same way if it is rejected too), and the first rejection is thrown afterwards. - A recovered exchange that the pool rejects again is recovered by a later scan of the recover task. That attempt counts as a redelivery, so `maximumRedeliveries` and `deadLetterUri` apply. - Nothing changes when the pool accepts the task, or for the default synchronous executor, which never rejects. With Camel's `Block` policy `execute` throws only when the pool is shut down or the waiting caller is interrupted, and that is handled the same way. Camel's `CallerRuns` is the JDK `CallerRunsPolicy`: it runs the task in the caller, but drops it without an exception when the pool is shut down. This change does not cover that case. Tests: - New `AggregateRejectedCompletionTest` (camel-core; latches, no sleeps). The setup is a thread pool profile with 1 thread, a queue of 1 and `rejectedPolicy=Abort`, and `completionSize(2)` (or `completionFromBatchConsumer()` in the last two tests). Group A blocks the thread, group B waits in the queue, and the task of group C is rejected. - `testRejectedAggregatedExchangeIsFailed`: after the thread is released, C1+C2 is passed to the exception handler with the `RejectedExecutionException`, the strategy's on completion runs `onFailure`, `getInProgressCompleteExchanges()` is back to 0, and the exchange that completed the group still fails with the rejection. On main the handler is not called (expected 1, was 0) and the count stays at 1. - `testRejectedAggregatedExchangeIsRecovered`: with a `KeyValueAggregationRepository` that has recovery enabled (100 ms), C1+C2 is recovered and delivered. On main it never is (mock got 2 of 3). - `testRejectedBatchCompletionFailsEveryGroup`: a batch of three messages completes the groups C, D and E at once, and the pool rejects all three. C1, D1 and E1 are passed to the exception handler, their on completions run `onFailure`, and `getInProgressCompleteExchanges()` is back to 0. On main the handler gets none of them. - `testRejectedBatchCompletionIsRecovered`: the same batch with the recoverable repository. C1 and D1 are recovered and delivered (E1 completed its group on arrival, so it was never stored and its sender gets the rejection). On main neither is recovered. - All four tests fail on main (3 of 3 surefire runs) and pass with the change. - camel-core, all tests under `org/apache/camel/processor` (the aggregator, multicast, splitter, threads and wire tap tests among them) together with the aggregator tests elsewhere (`*Aggregat*Test`): 2723 tests including the new ones, 0 failures, 17 skipped. That includes `AggregateParallelThreadPoolAbortTest` and `AggregateParallelThreadPoolCallerRunsTest`. The formatter and import-sort plugins ran. How it was found: by reading the EIPs that hand work to a thread pool (Threads, WireTap and Multicast handle a rejection, the aggregator did 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. The test above reproduced it on HotSpot. 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 with the `onSubmitCompletion` of this change states that the exchange is failed and handed to the exception handler. Both run a forced completion of one group; the batch case is covered by the tests above. The schedule search found no other outcome. # Target - [x] I checked that the commit is targeting the correct branch (Camel 4 uses the `main` branch) # Tracking - [x] If this is a large change, bug fix, or code improvement, I checked there is a [JIRA issue](https://issues.apache.org/jira/browse/CAMEL-26309) filed for the change (usually before you start working on it): [CAMEL-26309](https://issues.apache.org/jira/browse/CAMEL-26309). # Apache Camel coding standards and style - [x] I checked that each commit in the pull request has a meaningful subject line and body. - [ ] I have run `mvn clean install -DskipTests` locally from root folder and I have committed all auto-generated changes. (I built the core modules and ran the tests above, including the formatter and import-sort plugins. No generated file changes. I did not run the full root build.) # AI-assisted contributions - [x] If this PR includes AI-generated code, commits have proper co-authorship attribution (e.g., `Co-authored-by` trailers) and the PR description identifies the AI tool used. This PR was prepared with Claude Code (Claude Opus 5.5). The commit carries a `Co-Authored-By` trailer. _Claude Code on behalf of allthingssecurity_ 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
