[
https://issues.apache.org/jira/browse/CAMEL-25012?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18120102#comment-18120102
]
shashank commented on CAMEL-25012:
----------------------------------
Follow-up for the Wire Tap EIP, as requested in the review of #26870:
https://github.com/apache/camel/pull/26992 ({{WireTapProcessor}} had the same
leak: the tasks dropped by {{shutdownNow}} stayed counted as pending).
_Claude Code on behalf of allthingssecurity_
> OnCompletion EIP with parallelProcessing: a graceful shutdown drops the
> queued onCompletion tasks and interrupts the running ones
> ---------------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-25012
> URL: https://issues.apache.org/jira/browse/CAMEL-25012
> Project: Camel
> Issue Type: Bug
> Components: camel-core
> Reporter: shashank
> Priority: Minor
>
> With {{onCompletion().parallelProcessing()}} the {{OnCompletionProcessor}}
> synchronization submits the onCompletion work to its thread pool from the
> unit of work's done callback ({{OnCompletionProcessor.java:314-321}}
> onComplete, {{:345-354}} onFailure, {{:477-485}} BeforeConsumer mode). Right
> after that the exchange is removed from the inflight repository
> ({{DefaultUnitOfWork.java:272-275}}).
> The graceful shutdown ({{DefaultShutdownStrategy.ShutdownTask}},
> {{:667-716}}) waits for the route's inflight exchanges and for
> {{ShutdownAware.getPendingExchangesSize()}} of the route's services.
> {{OnCompletionProcessor}} is not {{ShutdownAware}}, and the onCompletion
> copies do not count as inflight for the route (they are not routed through
> the route's inflight advice), so the shutdown does not wait for them. It then
> shuts the route down, and {{OnCompletionProcessor.doShutdown()}} calls
> {{shutdownNow()}} on its pool ({{:107-111}}): queued onCompletion tasks are
> dropped and running ones are interrupted.
> So for exchanges that completed before the shutdown began, the onCompletion
> (for example sending a confirmation, releasing a reservation, committing an
> offset elsewhere) silently never runs, although the shutdown is graceful and
> reports no timeout.
> *Reproduction*: route
> {{from("direct:in").onCompletion().parallelProcessing().process(200 ms of
> work).end()...}}, 40 exchanges sent (all completed), then {{context.stop()}}:
> {noformat}
> [parallel] exchanges completed=40 inflight at stop=10
> (inflightRepository.size("orders")=0) graceful stop took 44ms
> [parallel] onCompletion tasks: started=10 finished=0 interrupted=10 never
> run=30
> [control] (without parallelProcessing) exchanges completed=5 ... started=5
> finished=5 interrupted=0 never run=0
> {noformat}
> The graceful stop took 44 ms although 10 onCompletion exchanges were in the
> inflight repository (none of them counted for the route, which is what the
> shutdown strategy asks for); 30 onCompletion tasks never ran and the 10
> running ones were interrupted.
> TLA+: {{EveryCompletionRuns}} is violated in 6 states (Arrive -> Done ->
> ShutdownBegin -> ShutdownWaitDone -> PoolDown drops the queued task) and
> {{NoInterrupted}} is violated. With the fix below ({{fix_all}},
> {{fix_all_w2}}) both hold and the shutdown terminates.
> *Proposed fix:* the same approach as CAMEL-24995 for the Wire Tap EIP:
> * {{OnCompletionProcessor}} implements {{ShutdownAware}} and counts the
> onCompletion tasks from submit until they finish (decrement on rejection), so
> the graceful shutdown waits for them within its timeout ({{deferShutdown}}
> returns false; {{getPendingExchangesSize}} returns the counter);
> * {{doShutdown}} shuts its own pool down gracefully ({{shutdownGraceful}})
> rather than with {{shutdownNow}}, so a forced shutdown after the timeout is
> the only case where work is dropped, and it is logged.
> _Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
