CAMEL-25012: camel-core - OnCompletion EIP: make a graceful shutdown wait for the parallel onCompletion tasks

[allthingssecurity]

Description

CAMEL-25012

With onCompletion().parallelProcessing(), OnCompletionProcessor submits the onCompletion of an exchange to its thread pool when the exchange's unit of work is done (onComplete, onFailure, and onAfterRoute in BeforeConsumer mode). The exchange then leaves the inflight repository. The graceful shutdown waits for the route's inflight exchanges, and for ShutdownAware.getPendingExchangesSize() of the route's services. But OnCompletionProcessor was not ShutdownAware, so the shutdown did not wait for the onCompletion tasks. It then shut the route down, and OnCompletionProcessor.doShutdown() called shutdownNow() on its pool:

• queued onCompletion tasks were dropped;
• running ones were interrupted.

This happens when the context is stopped or the route is removed. A plain stopRoute does not shut the pool down, so the tasks keep running there, but the graceful wait is missing in that case too. For exchanges that had completed before the shutdown began, the onCompletion (sending a confirmation, releasing a reservation, ...) silently never ran. The shutdown was graceful and reported no timeout. In a reproduction with 40 completed exchanges and a 200 ms onCompletion, the graceful stop took 44 ms: 30 onCompletion tasks never ran, and the other 10 were interrupted.

This change makes OnCompletionProcessor ShutdownAware like WireTapProcessor, and counts its tasks from submit, as #26851 (CAMEL-24995) now does for the Wire Tap:

• it implements ShutdownAware, with getPendingExchangesSize() returning the number of onCompletion tasks from submit until they are done. deferShutdown returns true and prepareShutdown is a no-op, as in WireTapProcessor;
• the counter is decremented when the task ends, or when the pool rejects it.

The three submit sites now go through one helper. doShutdown still calls shutdownNow, which now only affects tasks that are still pending after the shutdown timeout, as for the Wire Tap.

Tests: new OnCompletionParallelProcessingShutdownTest. An exchange completes, and Camel is stopped while its parallel onCompletion is still running. The onCompletion is released once the context is stopping (latches and Awaitility, no sleeps), and must complete. The test also checks that the running task counts as a pending exchange, and that none is pending afterwards. Without the fix:

AssertionFailedError: The onCompletion should be done ==> expected: <1> but was: <0>

(the onCompletion was interrupted by shutdownNow). With the fix it passes. *OnCompletion*,*Shutdown* in camel-core: 116 tests, 0 failures.

Found with a TLA+ model of the onCompletion thread pool and the graceful shutdown, then reproduced against the real classes. With this change, "every completed exchange gets its onCompletion" and "a graceful shutdown does not interrupt an onCompletion" hold, and the shutdown terminates. The reproduction now gives 40 of 40 onCompletions finished and none interrupted, with a graceful stop of about 1 s.

Known remaining window: the task is counted from submit, and the onCompletion synchronizations run last (Ordered.LOWEST), after the exchange has already left the route's inflight count. If another synchronization of the same exchange is slow (for example a remote file move or a transaction commit), a shutdown that checks in that gap sees neither an inflight exchange nor a pending task, and goes ahead. The same holds for the Wire Tap counter. Closing it would mean counting from when the synchronization is registered in process() and decrementing on every skip path; I kept this change to the same scope as #26851, but can do that here if you prefer.

This does not conflict with #26851: that PR changes only WireTapProcessor, and both apply together. It is the same approach (count from submit, decrement on rejection), and the counter pattern could later be shared if the committers prefer.

Target

• I checked that the commit is targeting the correct branch (Camel 4 uses the main branch)

Tracking

• If this is a large change, bug fix, or code improvement, I checked there is a JIRA issue filed for the change (usually before you start working on it).

Apache Camel coding standards and style

• 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 and tested the affected modules, including the formatter and import-sort plugins. I did not run the full root build.)

AI-assisted contributions

• 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

[oscerd]

Traced the counter and the ShutdownAware wiring — the fix is correct.

• OnCompletionProcessor now implements ShutdownAware the same way WireTapProcessor does: deferShutdown returns true, prepareShutdown is a no-op, and getPendingExchangesSize() returns the LongAdder task count, so the graceful strategy now waits for in-flight onCompletion tasks before doShutdown() calls shutdownNow().
• The counter is leak-free across every path: increment() before submit; decrement() in the wrapped task's finally, so it covers both normal and exception completion; and decrement() + rethrow in the catch when the pool rejects the task. There's no double-decrement — when submit throws, the counted runnable never runs, so its finally doesn't fire and only the catch decrements; when submit succeeds, only the finally decrements.
• All three submit sites (onComplete, onFailure, onAfterRoute) now go through submitTask, so the counting is uniform, and doShutdown keeping shutdownNow is fine because tasks are drained (or timed out) before it runs.

This mirrors #26851's WireTap approach and, as you note, doesn't conflict with it (different processor). The count-from-submit gap you documented is the same one the Wire Tap counter has; matching the scope of #26851 here is reasonable — a shared counter/tightening can be a follow-up if the committers prefer. getPendingExchangesSize() returning intValue() without a clamp is fine here since the onCompletion task count stays small and non-negative (unlike the Loop EIP counter in #26860). LGTM.

(CI has not been triggered on this PR yet — fork PR awaiting a maintainer to approve the workflow run; I'll confirm green before it merges.)

This review was generated with AI assistance and reviewed/issued by the human operator. Claude Code on behalf of oscerd