This is an automated email from the ASF dual-hosted git repository. apupier pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/camel.git
commit 3e05fa644910f950863315871e1a685b004f519a Author: smjain <[email protected]> AuthorDate: Wed Sep 23 19:35:17 2026 +0530 CAMEL-24960: camel-core - Fail a parallel multicast when the thread pool rejects a sub-exchange task Cause: MulticastProcessor.schedule() catches the RejectedExecutionException of the executor and only rejects a Rejectable runnable. That covers the multicast task itself (CAMEL-16829), but not the sub-exchange task submitted through the AsyncCompletionService, which is not Rejectable. The rejected sub-exchange was silently dropped although it was already counted as sent. Before CAMEL-16829 a rejected sub-task submission propagated out of completion.submit into run()'s catch; CAMEL-16829 added the catch in schedule() that swallowed it for non-Rejectable tasks. Effect: a parallel Multicast, Split or Recipient List whose thread pool rejects a sub-exchange task (for example rejectedPolicy Abort with a bounded queue) never completes: the callback is never invoked, the caller waits forever and the exchange stays inflight. SplitParallelThreadPoolAbortTest only passed because its lists have two elements, so the task's own re-schedule is rejected too. Fix: rethrow the RejectedExecutionException for a runnable that is not Rejectable. MulticastReactiveTask.run() then fails the exchange with it, the same way a rejected multicast task does. Co-Authored-By: Claude Opus 5.5 <[email protected]> --- .../apache/camel/processor/MulticastProcessor.java | 5 + .../MulticastParallelSubTaskRejectedTest.java | 103 +++++++++++++++++++++ 2 files changed, 108 insertions(+) diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java index e71bcf68053e..367296282eda 100644 --- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java +++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java @@ -420,6 +420,11 @@ public class MulticastProcessor extends BaseProcessorSupport } catch (RejectedExecutionException e) { if (runnable instanceof Rejectable rej) { rej.reject(); + } else { + // a sub-exchange task (submitted via the completion service) is not rejectable, + // so rethrow to let the multicast task fail the exchange instead of silently + // dropping the sub-exchange (which would otherwise never complete) + throw e; } } } else if (transacted) { diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/MulticastParallelSubTaskRejectedTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/MulticastParallelSubTaskRejectedTest.java new file mode 100644 index 000000000000..9828976658c2 --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/processor/MulticastParallelSubTaskRejectedTest.java @@ -0,0 +1,103 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.processor; + +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeUnit; + +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Exchange; +import org.apache.camel.RoutesBuilder; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.builder.ThreadPoolBuilder; +import org.apache.camel.util.concurrent.ThreadPoolRejectedPolicy; +import org.awaitility.Awaitility; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.parallel.Isolated; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; + +/** + * Tests that a parallel EIP completes, with the rejection as exception, when the thread pool rejects the task of a + * sub-exchange (and not only when it rejects the task of the EIP itself, see {@link SplitParallelThreadPoolAbortTest}). + * <p/> + * Each thread pool has a single thread and no queue. The EIP task itself runs on that thread, so the submission of the + * first sub-exchange task is rejected. + */ +@Isolated +@Timeout(30) +public class MulticastParallelSubTaskRejectedTest extends ContextTestSupport { + + @Test + public void testSplitSingleElement() throws Exception { + assertRejected("direct:split", List.of("a")); + } + + @Test + public void testMulticast() throws Exception { + assertRejected("direct:multicast", "Hello World"); + } + + @Test + public void testRecipientList() throws Exception { + assertRejected("direct:recipients", "mock:z"); + } + + private void assertRejected(String uri, Object body) throws Exception { + Future<Exchange> future = template.asyncSend(uri, e -> e.getIn().setBody(body)); + + // the exchange must complete and not hang forever + Exchange out = future.get(5, TimeUnit.SECONDS); + assertInstanceOf(RejectedExecutionException.class, out.getException()); + + Awaitility.await().atMost(5, TimeUnit.SECONDS) + .untilAsserted(() -> assertEquals(0, context.getInflightRepository().size())); + } + + @Override + protected RoutesBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() throws Exception { + from("direct:split") + .split(body()).executorService(newSingleThreadPool("split")) + .to("mock:x"); + + from("direct:multicast") + .multicast().executorService(newSingleThreadPool("multicast")) + .to("mock:y"); + + from("direct:recipients") + .recipientList(body()).executorService(newSingleThreadPool("recipients")); + } + + private ExecutorService newSingleThreadPool(String name) throws Exception { + return new ThreadPoolBuilder(getContext()) + .poolSize(1) + .maxPoolSize(1) + .maxQueueSize(0) + .rejectedPolicy(ThreadPoolRejectedPolicy.Abort) + .build(name); + } + }; + } +}
