allthingssecurity opened a new pull request, #29403: URL: https://github.com/apache/camel/pull/29403
# Description [CAMEL-26706](https://issues.apache.org/jira/browse/CAMEL-26706) When a group completes, `AggregateProcessor.onSubmitCompletion` books the aggregated exchange as in progress, then calls `aggregationStrategy.onCompletion(exchange)`, and only then adds the `AggregateOnCompletion` that undoes the booking. If `onCompletion` throws, nothing undoes it: - the group has already been removed from the repository; a recoverable repository has it in its completed store, but its recover task skips it on every run as "already in progress", so it is neither delivered nor recovered until the CamelContext restarts; - the in progress count stays above 0, so a stop with `forceCompletionOnStop` / `completeAllOnStop` waits for the shutdown timeout; - with `completionFromBatchConsumer`, the other groups of the batch are not submitted (`submitCompletions` only caught `RejectedExecutionException`), and stay marked as being completed, so they are lost the same way. The javadoc of `onCompletion` says it must not throw. Even so, CAMEL-26575 (#29200) made Multicast, Split and Recipient List fail the exchange in that case instead of hanging it, and CAMEL-26309 (#28709) already fails the aggregated exchange when the thread pool rejects it; this throw happens before the on completion is registered, so that fix does not cover it. Change (same handling as for the rejected task): - if `onCompletion` throws, the `AggregateOnCompletion` is added and the aggregated exchange is failed with the exception: its on completions run (it is no longer in progress and is not confirmed, so a recoverable repository recovers it), it goes to the exception handler, and the exception is rethrown, so the sender of the message that completed the group still fails as before. `onRejectedCompletion` becomes `failCompletion(exchange, cause, message)` for both cases. The `AggregateOnCompletion` is added in the catch rather than moved before `onCompletion`, so the order of the on completions on the normal path stays the same. - `submitCompletions` catches any `RuntimeException`, so the other groups of a batch are still sent before the first failure is thrown (an `Error` still stops the batch, as before). - the recover task catches a `RuntimeException` from resubmitting a recovered exchange (it has been failed and passed to the exception handler, and is recovered again on a later run) and goes on with the other exchanges. Otherwise an exchange whose `onCompletion` keeps throwing, with `maximumRedeliveries=0` (the JDBC default), would end every recover run before the exchanges after it. The path where `onCompletion` does not throw is unchanged. Tests: `AggregateOnCompletionThrowsTest` (strategy whose `onCompletion` throws once, in-memory recoverable `KeyValueAggregationRepository`, no database): - completion by size: the sender gets the exception, `A1+A2` is recovered and delivered once (`CamelRedelivered`), in progress back to 0; - memory repository: the aggregated exchange goes to the exception handler, its on completions run as failed, in progress is 0, and the next group is sent; - batch consumer: the group whose `onCompletion` throws is recovered and the other group of the batch is still sent; - recover task: `A1+A2` throws every time and is the first exchange the recover task finds (a test repository that scans in completion order); `B1+B2` is still recovered. On main all four tests fail (A1+A2 is never delivered, the exception handler is not called, the in progress count stays at 1); with the fix 4/4, and 80/80 over 20 runs. A mutant without the `submitCompletions` change fails only the batch test, and one without the recover-task catch only the recover test. camel-core tests of `processor/aggregator`, `processor/aggregate`, the Multicast, Idempotent Consumer and rollback tests: 392 run (offline, together with the other camel-core change of this batch, which touches different files), 0 failures; the camel-management aggregate tests (`ManagedAggregate*`, `BacklogTracerAggregate*`) pass. # 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) filed for the change (usually before you start working on it): [CAMEL-26706](https://issues.apache.org/jira/browse/CAMEL-26706). # 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 with the formatter and import-sort plugins and ran the related camel-core tests; 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]
