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 48eb7c074a74c79579e1666f6b3dbe7428234b88 Author: smjain <[email protected]> AuthorDate: Thu Sep 24 08:11:46 2026 +0530 CAMEL-24960: camel-core - Do not fail a multicast exchange that has already been completed The catches in MulticastReactiveTask.run(), MulticastTransactedTask.run(), aggregate() and timeout(), and reject(), set the exception on the original exchange before calling doDone, which does nothing when the task is already done. So a failure that happens after the exchange was completed, for example by the timeout or by stopOnException, was written onto an exchange the caller already had back. With the rethrown RejectedExecutionException of a sub-exchange task this can happen when the task is rejected after a sub-exchange failed and completed the EIP. The new doFailed and doTimeoutFailed only set the exception when they complete the task, and the timeout cancelling is shared with doDone. Co-Authored-By: Claude Opus 5.5 <[email protected]> --- .../apache/camel/processor/MulticastProcessor.java | 63 ++++++---- ...ticastParallelSubTaskRejectedAfterDoneTest.java | 131 +++++++++++++++++++++ 2 files changed, 172 insertions(+), 22 deletions(-) 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 367296282eda..8c1b3a0c7873 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 @@ -504,9 +504,7 @@ public class MulticastProcessor extends BaseProcessorSupport } } } catch (Exception e) { - original.setException(e); - // and do the done work - doDone(null, false); + doFailed(e); } finally { lock.unlock(); } @@ -541,9 +539,7 @@ public class MulticastProcessor extends BaseProcessorSupport } doTimeoutDone(result.get(), true); } catch (Exception e) { - original.setException(e); - // and do the done work - doTimeoutDone(null, false); + doTimeoutFailed(e); } finally { lock.unlock(); } @@ -557,24 +553,49 @@ public class MulticastProcessor extends BaseProcessorSupport protected void doDone(Exchange exchange, boolean forceExhaust) { if (done.compareAndSet(false, true)) { - // cancel timeout if we are done normally (we cannot cancel if called via onTimeout) - if (timeoutTask != null) { - try { - timeoutTask.cancel(true); - } catch (Exception e) { - // ignore - LOG.debug("Cancel timeout task caused an exception. This exception is ignored.", e); - } - } + cancelTimeoutTask(); MulticastProcessor.this.doDone(original, exchange, pairs, callback, false, forceExhaust); } } + /** + * Fails the original exchange with the given exception and does the done work, unless this task is already + * done. The exception is only set by the thread that completes the task, so an exchange that has already been + * completed (for example by the timeout), and handed back to the caller, is not changed afterwards. + */ + protected void doFailed(Exception e) { + if (done.compareAndSet(false, true)) { + cancelTimeoutTask(); + original.setException(e); + MulticastProcessor.this.doDone(original, null, pairs, callback, false, false); + } + } + + /** + * Same as {@link #doFailed(Exception)} but called from the timeout task, which must not cancel itself. + */ + protected void doTimeoutFailed(Exception e) { + if (done.compareAndSet(false, true)) { + original.setException(e); + MulticastProcessor.this.doDone(original, null, pairs, callback, false, false); + } + } + + private void cancelTimeoutTask() { + // cancel timeout if we are done normally (we cannot cancel if called via onTimeout) + if (timeoutTask != null) { + try { + timeoutTask.cancel(true); + } catch (Exception e) { + // ignore + LOG.debug("Cancel timeout task caused an exception. This exception is ignored.", e); + } + } + } + @Override public void reject() { - original.setException(new RejectedExecutionException("Task rejected executing from ExecutorService")); - // and do the done work - doDone(null, false); + doFailed(new RejectedExecutionException("Task rejected executing from ExecutorService")); } } @@ -673,8 +694,7 @@ public class MulticastProcessor extends BaseProcessorSupport schedule(this); } } catch (Exception e) { - original.setException(e); - doDone(null, false); + doFailed(e); } } @@ -707,8 +727,7 @@ public class MulticastProcessor extends BaseProcessorSupport try { next = doRun(); } catch (Exception e) { - original.setException(e); - doDone(null, false); + doFailed(e); return; } } diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/MulticastParallelSubTaskRejectedAfterDoneTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/MulticastParallelSubTaskRejectedAfterDoneTest.java new file mode 100644 index 000000000000..d747da1f1c23 --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/processor/MulticastParallelSubTaskRejectedAfterDoneTest.java @@ -0,0 +1,131 @@ +/* + * 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.Iterator; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.apache.camel.CamelExchangeException; +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Exchange; +import org.apache.camel.RoutesBuilder; +import org.apache.camel.RuntimeCamelException; +import org.apache.camel.builder.RouteBuilder; +import org.awaitility.Awaitility; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests that a sub-exchange task rejected by the thread pool after the parallel EIP has already completed (here by + * stopOnException) does not change the exchange that was handed back to the caller. + * <p/> + * The EIP task is busy with the streaming iterator while the first sub-exchange fails and completes the EIP. The thread + * pool is then shut down, so the submission of the next sub-exchange task is rejected. + */ +@Timeout(30) +class MulticastParallelSubTaskRejectedAfterDoneTest extends ContextTestSupport { + + private final CountDownLatch iteratorBlocked = new CountDownLatch(1); + private final CountDownLatch iteratorRelease = new CountDownLatch(1); + private ExecutorService pool; + + @Test + void testRejectedAfterStopOnException() throws Exception { + Future<Exchange> future = template.asyncSend("direct:start", e -> e.getIn().setBody(new BlockingIterator())); + + // the first sub-exchange fails, which completes the split while its task is blocked in the iterator + Exchange out = future.get(10, TimeUnit.SECONDS); + assertEquals(0, iteratorBlocked.getCount()); + assertInstanceOf(CamelExchangeException.class, out.getException()); + + // let the task continue, its submission of the last sub-exchange task is rejected, + // and wait until it is finished + pool.shutdown(); + iteratorRelease.countDown(); + assertTrue(pool.awaitTermination(10, TimeUnit.SECONDS)); + + // the late rejection must not replace the exception of the exchange the caller already has + assertInstanceOf(CamelExchangeException.class, out.getException()); + + Awaitility.await().atMost(5, TimeUnit.SECONDS) + .untilAsserted(() -> assertEquals(0, context.getInflightRepository().size())); + } + + @AfterEach + void shutdownPool() { + if (pool != null) { + pool.shutdownNow(); + } + } + + @Override + protected RoutesBuilder createRouteBuilder() { + pool = Executors.newCachedThreadPool(); + return new RouteBuilder() { + @Override + public void configure() { + from("direct:start") + .split(body()).streaming().parallelProcessing().stopOnException().executorService(pool) + .process(e -> { + // only fail once the task is blocked in the iterator + iteratorBlocked.await(10, TimeUnit.SECONDS); + throw new IllegalArgumentException("Forced"); + }) + .end(); + } + }; + } + + /** + * Iterates over two elements, and blocks when the end is reached until the test releases it. + */ + private final class BlockingIterator implements Iterator<String> { + + private final Iterator<String> delegate = List.of("a", "b").iterator(); + private final AtomicBoolean blocked = new AtomicBoolean(); + + @Override + public boolean hasNext() { + if (!delegate.hasNext() && blocked.compareAndSet(false, true)) { + iteratorBlocked.countDown(); + try { + iteratorRelease.await(10, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeCamelException(e); + } + } + return delegate.hasNext(); + } + + @Override + public String next() { + return delegate.next(); + } + } +}
