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 69c06886d5be4f73224fde94bd026c8a886353a4 Author: smjain <[email protected]> AuthorDate: Wed Sep 23 19:43:04 2026 +0530 CAMEL-24958: camel-core - Release the producers of recipients the Recipient List did not send to Cause: RecipientListProcessor acquires a producer for every recipient up front, and releases it (and stops a prototype endpoint) only in RecipientProcessorExchangePair.done(), which MulticastProcessor calls after a recipient was sent to. When the Recipient List completes before it sent to every recipient (stopOnException, timeout, a rejected task), the producers of the remaining recipients were never released. Effect: a pooled (non-singleton) producer, such as ftp/sftp, smb or ssh, is not returned to its pool, so every such exchange creates and starts a new producer which is never stopped. With cacheSize(-1) the prototype endpoint and its producer are never stopped, not even when the CamelContext stops. Fix: when the Recipient List is done, release the producer of every pair that was not begun. A pair is begun or released exactly once (compare-and-set on the pair state), so a pair whose task starts after the release is not sent, and done() is idempotent. Co-Authored-By: Claude Opus 5.5 <[email protected]> --- .../camel/processor/RecipientListProcessor.java | 65 ++++++- .../RecipientListReleaseUnsentProducerTest.java | 193 +++++++++++++++++++++ .../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 11 ++ 3 files changed, 264 insertions(+), 5 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 52afcf27ca18..6f92ad45fc28 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 @@ -25,8 +25,10 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; +import java.util.concurrent.atomic.AtomicInteger; import org.apache.camel.AggregationStrategy; +import org.apache.camel.AsyncCallback; import org.apache.camel.AsyncProducer; import org.apache.camel.CamelContext; import org.apache.camel.Endpoint; @@ -80,6 +82,16 @@ public class RecipientListProcessor extends MulticastProcessor { * using it. */ static final class RecipientProcessorExchangePair implements ProcessorExchangePair { + private static final int NEW = 0; + private static final int BEGUN = 1; + private static final int DONE = 2; + private static final int RELEASED = 3; + // used instead of the prepared processor when the pair was released before it could begin + private static final Processor SKIP = exchange -> { + // noop + }; + + private final AtomicInteger state = new AtomicInteger(NEW); private final int index; private final Endpoint endpoint; private final AsyncProducer producer; @@ -120,11 +132,18 @@ public class RecipientListProcessor extends MulticastProcessor { @Override public Processor getProcessor() { - return prepared; + // the recipient list completed (and released the producer) before this pair could begin, + // so it must not be sent anymore + return state.get() == RELEASED ? SKIP : prepared; } @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 LOG.trace("RecipientProcessorExchangePair #{} begin: {}", index, exchange); exchange.setProperty(ExchangePropertyKey.RECIPIENT_LIST_ENDPOINT, endpoint.getEndpointUri()); @@ -140,12 +159,32 @@ public class RecipientListProcessor extends MulticastProcessor { @Override public void done() { + if (!state.compareAndSet(BEGUN, DONE)) { + // not begun (released already), or done already + return; + } LOG.trace("RecipientProcessorExchangePair #{} done: {}", index, exchange); + // preserve original MEP + if (originalPattern != null) { + exchange.setPattern(originalPattern); + } + releaseProducer(); + } + + /** + * Releases the producer of this pair when the recipient list is done before the pair was begun (such as + * stopOnException or timeout), as then {@link #done()} is not called. If the pair has not begun yet, it will + * not be sent anymore. + */ + void releaseIfNotBegun() { + if (state.compareAndSet(NEW, RELEASED)) { + LOG.trace("RecipientProcessorExchangePair #{} released as not sent: {}", index, exchange); + releaseProducer(); + } + } + + private void releaseProducer() { try { - // preserve original MEP - if (originalPattern != null) { - exchange.setPattern(originalPattern); - } // when we are done we should release back in pool producerCache.releaseProducer(endpoint, producer); // and stop prototype endpoints @@ -335,6 +374,22 @@ public class RecipientListProcessor extends MulticastProcessor { index, producerCache, endpoint, producer, prepared, copy, pattern, prototypeEndpoint); } + @Override + protected void doDone( + Exchange original, Exchange subExchange, Iterable<ProcessorExchangePair> pairs, + AsyncCallback callback, boolean doneSync, boolean forceExhaust) { + if (pairs != null) { + // the producers are acquired up front for all recipients, so release the producers of the recipients + // that were not sent to, as the recipient list may be done before (such as stopOnException or timeout) + for (ProcessorExchangePair pair : pairs) { + if (pair instanceof RecipientProcessorExchangePair rpair) { + rpair.releaseIfNotBegun(); + } + } + } + super.doDone(original, subExchange, pairs, callback, doneSync, forceExhaust); + } + protected static Object prepareRecipient(Exchange exchange, Object recipient) throws NoTypeConversionAvailableException { return ProcessorHelper.prepareRecipient(exchange, recipient); } 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 new file mode 100644 index 000000000000..145395bd06de --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/processor/RecipientListReleaseUnsentProducerTest.java @@ -0,0 +1,193 @@ +/* + * 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.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.camel.CamelContext; +import org.apache.camel.Consumer; +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Endpoint; +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.support.DefaultComponent; +import org.apache.camel.support.DefaultEndpoint; +import org.apache.camel.support.DefaultProducer; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +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). + * <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. + */ +public class RecipientListReleaseUnsentProducerTest extends ContextTestSupport { + + private final AtomicInteger producersStarted = new AtomicInteger(); + private final AtomicInteger producersSent = new AtomicInteger(); + private final AtomicInteger endpointsStarted = new AtomicInteger(); + private final AtomicInteger endpointsStopped = new AtomicInteger(); + private final CountDownLatch slowLatch = new CountDownLatch(1); + + @Override + protected CamelContext createCamelContext() throws Exception { + CamelContext context = super.createCamelContext(); + context.addComponent("pooled", new PooledComponent()); + return context; + } + + @Test + public void testStopOnException() { + for (int i = 0; i < 5; i++) { + Exchange out = template.send("direct:stop", e -> e.getIn().setHeader("to", "direct:boom,pooled:b")); + assertNotNull(out.getException()); + } + + assertEquals(0, producersSent.get()); + // the producer of pooled:b is released after each exchange, so it is reused from the pool + assertEquals(1, producersStarted.get()); + } + + @Test + public void testStopOnExceptionParallel() { + for (int i = 0; i < 5; i++) { + Exchange out = template.send("direct:parallel", e -> e.getIn().setHeader("to", "direct:boom,pooled:b")); + assertNotNull(out.getException()); + } + + // the single thread of the pool fails direct:boom before the recipient list task gets to send to pooled:b + assertEquals(0, producersSent.get()); + assertEquals(1, producersStarted.get()); + } + + @Test + public void testPrototypeEndpointStopped() { + for (int i = 0; i < 5; i++) { + final int n = i; + Exchange out = template.send("direct:prototype", e -> e.getIn().setHeader("to", "direct:boom,pooled:c" + n)); + assertNotNull(out.getException()); + } + + assertEquals(5, endpointsStarted.get()); + // every prototype endpoint must be stopped, also when it was not sent to + assertEquals(5, endpointsStopped.get()); + } + + @Test + public void testTimeout() { + try { + for (int i = 0; i < 2; i++) { + Exchange out = template.send("direct:timeout", + e -> e.getIn().setHeader("to", "direct:slow,pooled:b")); + assertNull(out.getException()); + } + } finally { + slowLatch.countDown(); + } + + assertEquals(0, producersSent.get()); + assertEquals(1, producersStarted.get()); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + errorHandler(noErrorHandler()); + + from("direct:stop").recipientList(header("to")).stopOnException(); + + from("direct:parallel").recipientList(header("to")).stopOnException() + .executorService(context.getExecutorServiceManager().newSingleThreadExecutor(this, "single")); + + 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 + from("direct:timeout").recipientList(header("to")).parallelProcessing().timeout(100) + .executorService(context.getExecutorServiceManager().newSingleThreadExecutor(this, "slow")); + + from("direct:boom").throwException(new IllegalArgumentException("Forced")); + + from("direct:slow").process(e -> assertTrue(slowLatch.await(20, TimeUnit.SECONDS))); + } + }; + } + + private final class PooledComponent extends DefaultComponent { + + @Override + protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) { + return new PooledEndpoint(uri, this); + } + } + + private final class PooledEndpoint extends DefaultEndpoint { + + private PooledEndpoint(String uri, PooledComponent component) { + super(uri, component); + } + + @Override + public boolean isSingletonProducer() { + return false; + } + + @Override + public Producer createProducer() { + return new DefaultProducer(this) { + @Override + public void process(Exchange exchange) { + producersSent.incrementAndGet(); + } + + @Override + protected void doStart() { + producersStarted.incrementAndGet(); + } + }; + } + + @Override + public Consumer createConsumer(Processor processor) { + throw new UnsupportedOperationException("Consumer not supported"); + } + + @Override + protected void doStart() throws Exception { + endpointsStarted.incrementAndGet(); + super.doStart(); + } + + @Override + protected void doStop() throws Exception { + endpointsStopped.incrementAndGet(); + super.doStop(); + } + } +} 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 f5e04dcb608e..613b43fc9e80 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 @@ -2712,3 +2712,14 @@ are now failed with a `RejectedExecutionException` and their on completions are reply of a discarded exchange (`waitForTaskToComplete`) is released with that exception, instead of waiting until its `timeout`, or forever when the timeout is disabled. On completions handed over to a discarded InOnly exchange, such as the commit or rollback of the consumer that received the message, now run as a failure, where previously they never ran. + +=== 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 +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. + +With `parallelProcessing`, a recipient whose task had not started yet when the Recipient List completed is now +skipped instead of being sent to afterwards. As before, recipients that had already started keep running.
