shashank created CAMEL-24960:
--------------------------------
Summary: Multicast/Split/RecipientList with parallelProcessing
hangs forever when the thread pool rejects a sub-exchange task
Key: CAMEL-24960
URL: https://issues.apache.org/jira/browse/CAMEL-24960
Project: Camel
Issue Type: Bug
Components: camel-core
Reporter: shashank
When a parallel Multicast, Split or Recipient List runs on a thread pool that
rejects tasks (for example {{rejectedPolicy(Abort)}} with a bounded queue), a
rejected sub-exchange is silently dropped. The exchange then never completes:
the caller waits forever and the exchange stays in the inflight repository.
{{MulticastProcessor.schedule(Runnable, boolean)}} catches
{{RejectedExecutionException}} and only calls {{reject()}} when the runnable is
{{Rejectable}}. That handles the {{MulticastTask}} itself (CAMEL-16829). The
sub-exchange task submitted by {{completion.submit(...)}} in
{{MulticastReactiveTask.run()}} is an {{AsyncCompletionService.Task}}, which is
not {{Rejectable}}, so the rejection is swallowed. The sub-exchange was already
counted in {{nbExchangeSent}}, so {{aggregate()}} never reaches {{nbAggregated
>= nbExchangeSent}} and {{doDone}} is never called. Without a {{timeout}}
nothing ever completes the exchange.
Reproduced against 4.23.0-SNAPSHOT with the thread pool from
{{SplitParallelThreadPoolAbortTest}} (poolSize 1, maxPoolSize 1, maxQueueSize
0, Abort). The task occupies the only thread, so the sub-exchange submission is
rejected:
{noformat}
split(body()).executorService(pool) body=[a, b] -> completed with
RejectedExecutionException (the task's own reschedule was rejected; part a was
dropped)
split(body()).executorService(pool) body=[a] -> not completed after 5
s, inflight=1
multicast().executorService(pool).to(..) -> not completed after 5
s, inflight=2
recipientList(body()).executorService(pool) "mock:z" -> not completed after 5
s, inflight=3
{noformat}
{{SplitParallelThreadPoolAbortTest}} ("Tests that the EIP does not hang-threads
due to thread-pools being exhausted and rejects new tasks") only passes because
its lists have 2 elements, so the task's own re-schedule is rejected too.
Proposed fix: fail the multicast when a sub-exchange task is rejected.
{{schedule()}} rethrows the {{RejectedExecutionException}} for a runnable that
is not {{Rejectable}}. It propagates out of {{completion.submit(...)}} into the
catch of {{MulticastReactiveTask.run()}}, which fails the exchange with it, the
same way {{MulticastTask.reject()}} does for the task itself. The exception is
the executor's own {{RejectedExecutionException}} (it names the pool), not the
"Task rejected executing from ExecutorService" one of {{reject()}}. The
alternative, completing the rejected sub-exchange as a failed sub-exchange so
that {{stopOnException}} and the aggregation strategy see it, needs the task
runner to run without an executor and is a larger change; the multicast task's
own rejection already fails the whole exchange, so the fix does the same.
Only the parallel path is affected: {{schedule()}} only submits to the executor
with {{parallelProcessing}}, and the other runnables it gets there (the
multicast task itself) are {{Rejectable}}.
With a Recipient List, the producer of a recipient whose task was rejected is
not released by this fix alone; the fix for CAMEL-24958 releases it.
A PR with the fix follows, with regression test
{{MulticastParallelSubTaskRejectedTest}} (split with one element, multicast,
recipient list, each on a pool with one thread, no queue and {{Abort}}).
Without the fix every test fails with {{java.util.concurrent.TimeoutException}}
after 5 s (the callback is never invoked); with it they complete with
{{RejectedExecutionException}} and nothing stays inflight.
Found with a TLA+ model of the multicast task, then reproduced against
4.23.0-SNAPSHOT.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
