shashank created CAMEL-24952:
--------------------------------
Summary: 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
{{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)
