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 805e314dde29248df6b4a899e9b71ddf33dba1ef Author: smjain <[email protected]> AuthorDate: Thu Sep 24 08:23:38 2026 +0530 CAMEL-24958: camel-core - Emit no exchange events for a recipient the Recipient List skipped MulticastProcessor.beforeSend() emitted the ExchangeSendingEvent before it called pair.begin(), where a pair released by the done Recipient List was found to be skipped. afterSend() then emitted the ExchangeSentEvent too. So a skipped recipient reported a send to an endpoint that was never called (tracing spans, metrics). The pair is now claimed in RecipientListProcessor.beforeSend(), before any event. A released pair is skipped without events: it is not begun, its processor does nothing, and done() does not release it again. The upgrade guide entry no longer lists "a rejected task" as a way the Recipient List completes early: a rejected sub-exchange task only completes it with CAMEL-24960, so the entry now gives stopOnException and timeout as examples. Co-Authored-By: Claude Opus 5.5 <[email protected]> --- .../camel/processor/RecipientListProcessor.java | 31 ++++-- .../RecipientListReleaseUnsentProducerTest.java | 108 ++++++++++++++++++++- .../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 2 +- 3 files changed, 133 insertions(+), 8 deletions(-) diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/RecipientListProcessor.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/RecipientListProcessor.java index 6f92ad45fc28..664dce0c6085 100644 --- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/RecipientListProcessor.java +++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/RecipientListProcessor.java @@ -48,6 +48,7 @@ import org.apache.camel.support.ExchangeHelper; import org.apache.camel.support.MessageHelper; import org.apache.camel.support.ObjectHelper; import org.apache.camel.support.service.ServiceHelper; +import org.apache.camel.util.StopWatch; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -137,14 +138,22 @@ public class RecipientListProcessor extends MulticastProcessor { return state.get() == RELEASED ? SKIP : prepared; } + /** + * Claims this pair to be sent, before any event is emitted for it. Fails when the recipient list is already + * done and has released the pair, as then it must not be sent anymore. + */ + boolean claim() { + if (state.compareAndSet(NEW, BEGUN)) { + return true; + } + LOG.trace("RecipientProcessorExchangePair #{} not sent as the recipient list is already done: {}", index, + exchange); + return false; + } + @Override public void begin() { - if (!state.compareAndSet(NEW, BEGUN)) { - LOG.trace("RecipientProcessorExchangePair #{} not sent as the recipient list is already done: {}", index, - exchange); - return; - } - // we have already acquired and prepare the producer + // the pair has been claimed (see beforeSend), and we have already acquired and prepare the producer LOG.trace("RecipientProcessorExchangePair #{} begin: {}", index, exchange); exchange.setProperty(ExchangePropertyKey.RECIPIENT_LIST_ENDPOINT, endpoint.getEndpointUri()); // ensure stream caching is reset @@ -374,6 +383,16 @@ public class RecipientListProcessor extends MulticastProcessor { index, producerCache, endpoint, producer, prepared, copy, pattern, prototypeEndpoint); } + @Override + protected StopWatch beforeSend(ProcessorExchangePair pair) { + if (pair instanceof RecipientProcessorExchangePair rpair && !rpair.claim()) { + // the recipient list is already done and has released this pair, so it is skipped: it is not begun, its + // processor does nothing, and no exchange sending or sent event is emitted as nothing is sent + return null; + } + return super.beforeSend(pair); + } + @Override protected void doDone( Exchange original, Exchange subExchange, Iterable<ProcessorExchangePair> pairs, diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/RecipientListReleaseUnsentProducerTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/RecipientListReleaseUnsentProducerTest.java index edd2e80a512d..d9de5f7fc217 100644 --- a/core/camel-core/src/test/java/org/apache/camel/processor/RecipientListReleaseUnsentProducerTest.java +++ b/core/camel-core/src/test/java/org/apache/camel/processor/RecipientListReleaseUnsentProducerTest.java @@ -16,8 +16,14 @@ */ package org.apache.camel.processor; +import java.util.List; import java.util.Map; +import java.util.concurrent.AbstractExecutorService; +import java.util.concurrent.BlockingDeque; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Future; +import java.util.concurrent.LinkedBlockingDeque; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -29,9 +35,13 @@ import org.apache.camel.Exchange; import org.apache.camel.Processor; import org.apache.camel.Producer; import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.spi.CamelEvent; +import org.apache.camel.spi.CamelEvent.ExchangeSendingEvent; +import org.apache.camel.spi.CamelEvent.ExchangeSentEvent; import org.apache.camel.support.DefaultComponent; import org.apache.camel.support.DefaultEndpoint; import org.apache.camel.support.DefaultProducer; +import org.apache.camel.support.EventNotifierSupport; import org.junit.jupiter.api.Test; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -41,7 +51,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue; /** * Tests that the recipient list releases the producers it acquired for recipients which it did not send to, because it - * was done before (stopOnException, timeout). + * was done before (stopOnException, timeout), and that it emits no exchange sending or sent events for them. * <p/> * The pooled component has no singleton producer, so a producer which is released is reused from the pool by the next * exchange, and a producer which is not released makes the next exchange create and start a new producer. @@ -53,11 +63,24 @@ class RecipientListReleaseUnsentProducerTest extends ContextTestSupport { private final AtomicInteger endpointsStarted = new AtomicInteger(); private final AtomicInteger endpointsStopped = new AtomicInteger(); private final CountDownLatch slowLatch = new CountDownLatch(1); + private final Map<String, AtomicInteger> sendingEvents = new ConcurrentHashMap<>(); + private final Map<String, AtomicInteger> sentEvents = new ConcurrentHashMap<>(); + private final ManualExecutorService manualExecutor = new ManualExecutorService(); @Override protected CamelContext createCamelContext() throws Exception { CamelContext context = super.createCamelContext(); context.addComponent("pooled", new PooledComponent()); + context.getManagementStrategy().addEventNotifier(new EventNotifierSupport() { + @Override + public void notify(CamelEvent event) { + if (event instanceof ExchangeSendingEvent sending) { + count(sendingEvents, sending.getEndpoint()); + } else if (event instanceof ExchangeSentEvent sent) { + count(sentEvents, sent.getEndpoint()); + } + } + }); return context; } @@ -85,6 +108,33 @@ class RecipientListReleaseUnsentProducerTest extends ContextTestSupport { assertEquals(1, producersStarted.get()); } + @Test + void testStopOnExceptionParallelSkippedRecipientHasNoEvents() throws Exception { + Future<Exchange> future + = template.asyncSend("direct:manual", e -> e.getIn().setHeader("to", "direct:boom,pooled:b")); + + // the recipient list task submits the task of direct:boom, and then schedules itself again + runTask(manualExecutor.tasks.pollFirst(10, TimeUnit.SECONDS)); + // the recipient list task runs again, before direct:boom, and submits the task of pooled:b + runTask(manualExecutor.tasks.pollLast()); + // direct:boom fails, so the recipient list is done, and releases pooled:b before its task has begun + runTask(manualExecutor.tasks.pollFirst()); + // the task of pooled:b, which must skip it + runTask(manualExecutor.tasks.pollFirst()); + assertTrue(manualExecutor.tasks.isEmpty()); + + Exchange out = future.get(10, TimeUnit.SECONDS); + assertNotNull(out.getException()); + assertEquals(0, producersSent.get()); + assertEquals(1, producersStarted.get()); + + assertEquals(1, eventCount(sendingEvents, "direct://boom")); + assertEquals(1, eventCount(sentEvents, "direct://boom")); + // pooled:b is not sent to, so there must be no events for it + assertEquals(0, eventCount(sendingEvents, "pooled://b")); + assertEquals(0, eventCount(sentEvents, "pooled://b")); + } + @Test void testPrototypeEndpointStopped() { for (int i = 0; i < 5; i++) { @@ -126,6 +176,10 @@ class RecipientListReleaseUnsentProducerTest extends ContextTestSupport { from("direct:parallel").recipientList(header("to")).stopOnException() .executorService(context.getExecutorServiceManager().newSingleThreadExecutor(this, "single")); + // the test runs the tasks of the pool itself, in the order it needs + from("direct:manual").recipientList(header("to")).parallelProcessing().stopOnException() + .executorService(manualExecutor); + from("direct:prototype").recipientList(header("to")).stopOnException().cacheSize(-1); // the single thread of the pool is blocked by direct:slow, so pooled:b is not sent before the timeout @@ -139,6 +193,58 @@ class RecipientListReleaseUnsentProducerTest extends ContextTestSupport { }; } + private static void runTask(Runnable task) { + assertNotNull(task); + task.run(); + } + + private static void count(Map<String, AtomicInteger> events, Endpoint endpoint) { + events.computeIfAbsent(endpoint.getEndpointUri(), k -> new AtomicInteger()).incrementAndGet(); + } + + private static int eventCount(Map<String, AtomicInteger> events, String uri) { + AtomicInteger count = events.get(uri); + return count != null ? count.get() : 0; + } + + /** + * Only queues the submitted tasks, which the test then runs itself. + */ + private static final class ManualExecutorService extends AbstractExecutorService { + + private final BlockingDeque<Runnable> tasks = new LinkedBlockingDeque<>(); + + @Override + public void execute(Runnable command) { + tasks.add(command); + } + + @Override + public void shutdown() { + // noop + } + + @Override + public List<Runnable> shutdownNow() { + return List.of(); + } + + @Override + public boolean isShutdown() { + return false; + } + + @Override + public boolean isTerminated() { + return false; + } + + @Override + public boolean awaitTermination(long timeout, TimeUnit unit) { + return true; + } + } + private final class PooledComponent extends DefaultComponent { @Override diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc index 613b43fc9e80..2bfd5cddf262 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc @@ -2716,7 +2716,7 @@ the commit or rollback of the consumer that received the message, now run as a f === camel-core - Recipient List releases the producers of recipients it did not send to The Recipient List acquires a producer for every recipient before it starts sending. When it completes before -it has sent to every recipient (`stopOnException`, `timeout`, or a rejected task), it now releases the +it has sent to every recipient (for example with `stopOnException` or a `timeout`), it now releases the producers of the recipients it did not send to. Previously these producers were never released: a pooled (non-singleton) producer was not returned to its pool, and with `cacheSize(-1)` the prototype endpoint and its producer were never stopped.
