[
https://issues.apache.org/jira/browse/CAMEL-24952?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen updated CAMEL-24952:
--------------------------------
Fix Version/s: 4.23.0
> camel-seda - virtualThreadPerTask consumer: graceful stop does not wait for
> polled exchanges whose task has not started yet
> ---------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-24952
> URL: https://issues.apache.org/jira/browse/CAMEL-24952
> Project: Camel
> Issue Type: Bug
> Components: camel-seda
> Reporter: shashank
> Priority: Minor
> Fix For: 4.23.0
>
>
> {{ThreadPerTaskSedaConsumer.processPolledExchange}} hands each polled
> exchange to the task executor. {{activeTasks}} is only incremented when the
> task starts running (ThreadPerTaskSedaConsumer.java:133). An exchange that
> was polled but whose task has not started yet is:
> * not in the queue;
> * not inflight, because it has not entered the route;
> * not counted in {{activeTasks}}.
> So both the {{DefaultShutdownStrategy}} wait loop and {{prepareShutdown}}
> (which skips waiting when {{activeTasks.sum() == 0}}, :168-169) consider the
> consumer idle. The route is stopped. When the task finally runs, the stopped
> error handler rejects the exchange ({{RejectedExecutionException}} from
> {{RedeliveryErrorHandler.SimpleTask.runNotAllowed}}). The exception is only
> logged, so the message is lost. The classic {{SedaConsumer}} does not have
> this problem, because it processes the exchange on the polling thread that
> {{prepareShutdown}} waits for.
> The window is the delay between {{taskExecutor.execute}} and the task
> starting. It only matters when that delay is longer than the coordinator's
> next empty poll ({{pollTimeout}}), e.g. with starved/pinned virtual-thread
> carriers. Reproduced against 4.23.0-SNAPSHOT with a {{ThreadPoolFactory}}
> that delays the start of cached-pool tasks by 2 s; Camel code unchanged:
> {noformat}
> t=120 ms stopRoute(v) returned, route status=Stopped
> WARN Error processing exchange. Exchange[]. Caused by:
> [java.util.concurrent.RejectedExecutionException - null] (logged twice)
> {noformat}
> Proposed fix: increment {{activeTasks}} before {{taskExecutor.execute(...)}}.
> Decrement it (and release the concurrency permit) in the task's {{finally}}
> block, and also when {{execute}} throws (e.g. {{RejectedExecutionException}};
> previously the permit acquired in {{beforePoll}} leaked in that case).
> A PR with the fix follows, with regression test
> {{ThreadPerTaskSedaConsumerStopTest}} (a {{ThreadPoolFactory}} whose task
> executor holds dispatched tasks until the stop reaches {{awaitTermination}}).
> The duplicate exception-handler report below is left for a separate change.
> Minor, same class: a failed exchange is reported to the exception handler
> twice, once by {{onProcessingDone(exchange, prepared)}} and again in the
> callback's {{finally}} block (:143-146), because {{prepared == exchange}}.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
