This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-24990 in repository https://gitbox.apache.org/repos/asf/camel.git
commit c4cde83a39d8114bb096df03980aadce58513995 Author: Claus Ibsen <[email protected]> AuthorDate: Thu Sep 24 12:58:17 2026 +0200 CAMEL-24990: camel-core - Multicast, Split and Recipient List EIPs: fix bugs found in a deep review - A streaming split that finished early (stopOnException or timeout) read the rest of the input in doDone to release the exchanges, and ran onPrepare on every remaining part. Only a collection of pairs is released now. - A transacted recipient list shared one transaction context data map across all transactions. It is now created per exchange. - When a recipient could not be resolved, the producers already acquired for the recipients before it were never released. Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../apache/camel/processor/MulticastProcessor.java | 4 +- .../camel/processor/RecipientListProcessor.java | 90 ++++++++++----------- .../java/org/apache/camel/processor/Splitter.java | 7 +- .../RecipientListInvalidEndpointReleaseTest.java | 94 ++++++++++++++++++++++ .../RecipientListTransactedContextDataTest.java | 64 +++++++++++++++ ...itterStreamingStopOnExceptionReadAheadTest.java | 78 ++++++++++++++++++ 6 files changed, 287 insertions(+), 50 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 6e9584466a11..d172b5fa2e8e 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 @@ -950,8 +950,10 @@ public class MulticastProcessor extends BaseProcessorSupport } } - if (processorExchangeFactory != null && pairs != null) { + if (processorExchangeFactory != null && pairs instanceof Collection) { // the exchanges on the pairs was created with a factory, so they should be released + // (only when the pairs are a collection, as iterating a streaming iterable would read, prepare and create + // exchanges for all the remaining parts, such as after stopOnException or a timeout) try { for (ProcessorExchangePair pair : pairs) { processorExchangeFactory.release(pair.getExchange()); 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 664dce0c6085..8f3a65b31cfd 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 @@ -16,7 +16,6 @@ */ package org.apache.camel.processor; -import java.lang.reflect.Array; import java.util.ArrayList; import java.util.Collection; import java.util.Iterator; @@ -74,7 +73,6 @@ public class RecipientListProcessor extends MulticastProcessor { private final String delimiter; private final ProducerCache producerCache; private int cacheSize; - private Map<String, Object> txData; /** * Class that represent each step in the recipient list to do @@ -253,52 +251,50 @@ public class RecipientListProcessor extends MulticastProcessor { recipientList = expression.evaluate(exchange, Object.class); } - // optimize for recipient without need for using delimiter - // (if its list/collection/array type) - if (recipientList instanceof List<?> col) { - int size = col.size(); - List<ProcessorExchangePair> result = new ArrayList<>(size); - int index = 0; - for (Object recipient : col) { - index = doCreateProcessorExchangePairs(exchange, recipient, result, index); - } - return result; - } else if (recipientList instanceof Collection<?> col) { - int size = col.size(); - List<ProcessorExchangePair> result = new ArrayList<>(size); + // each exchange (transaction) has its own transaction context data, shared by its copies + Map<String, Object> txData = exchange.isTransacted() ? new ConcurrentHashMap<>() : null; + + List<ProcessorExchangePair> result = new ArrayList<>(); + try { int index = 0; - for (Object recipient : col) { - index = doCreateProcessorExchangePairs(exchange, recipient, result, index); + // optimize for recipient without need for using delimiter + // (if its collection/array type) + if (recipientList instanceof Collection<?> col) { + for (Object recipient : col) { + index = doCreateProcessorExchangePairs(exchange, recipient, result, index, txData); + } + } else if (recipientList != null && recipientList.getClass().isArray()) { + for (Object recipient : (Object[]) recipientList) { + index = doCreateProcessorExchangePairs(exchange, recipient, result, index, txData); + } + } else { + // okay we have to use iterator based separated by delimiter + Iterator<?> iter; + if (delimiter != null && delimiter.equalsIgnoreCase(IGNORE_DELIMITER_MARKER)) { + iter = ObjectHelper.createIterator(recipientList, null); + } else { + iter = ObjectHelper.createIterator(recipientList, delimiter); + } + while (iter.hasNext()) { + index = doCreateProcessorExchangePairs(exchange, iter.next(), result, index, txData); + } } - return result; - } else if (recipientList != null && recipientList.getClass().isArray()) { - Object[] arr = (Object[]) recipientList; - int size = Array.getLength(recipientList); - List<ProcessorExchangePair> result = new ArrayList<>(size); - int index = 0; - for (Object recipient : arr) { - index = doCreateProcessorExchangePairs(exchange, recipient, result, index); + } catch (Exception e) { + // a recipient could not be resolved, so release the producers acquired for the recipients before it, + // as the recipient list is not sent to any of them + for (ProcessorExchangePair pair : result) { + if (pair instanceof RecipientProcessorExchangePair rpair) { + rpair.releaseIfNotBegun(); + } } - return result; - } - - // okay we have to use iterator based separated by delimiter - Iterator<?> iter; - if (delimiter != null && delimiter.equalsIgnoreCase(IGNORE_DELIMITER_MARKER)) { - iter = ObjectHelper.createIterator(recipientList, null); - } else { - iter = ObjectHelper.createIterator(recipientList, delimiter); - } - List<ProcessorExchangePair> result = new ArrayList<>(); - int index = 0; - while (iter.hasNext()) { - index = doCreateProcessorExchangePairs(exchange, iter.next(), result, index); + throw e; } return result; } private int doCreateProcessorExchangePairs( - Exchange exchange, Object recipient, List<ProcessorExchangePair> result, int index) + Exchange exchange, Object recipient, List<ProcessorExchangePair> result, int index, + Map<String, Object> txData) throws NoTypeConversionAvailableException { boolean prototype = cacheSize < 0; @@ -330,7 +326,7 @@ public class RecipientListProcessor extends MulticastProcessor { } // then create the exchange pair - result.add(createProcessorExchangePair(index++, endpoint, producer, exchange, pattern, prototype)); + result.add(createProcessorExchangePair(index++, endpoint, producer, exchange, pattern, prototype, txData)); return index; } @@ -340,6 +336,13 @@ public class RecipientListProcessor extends MulticastProcessor { protected ProcessorExchangePair createProcessorExchangePair( int index, Endpoint endpoint, Producer producer, Exchange exchange, ExchangePattern pattern, boolean prototypeEndpoint) { + return createProcessorExchangePair(index, endpoint, producer, exchange, pattern, prototypeEndpoint, + exchange.isTransacted() ? new ConcurrentHashMap<>() : null); + } + + private ProcessorExchangePair createProcessorExchangePair( + int index, Endpoint endpoint, Producer producer, + Exchange exchange, ExchangePattern pattern, boolean prototypeEndpoint, Map<String, Object> txData) { // copy exchange, and do not share the unit of work Exchange copy = processorExchangeFactory.createCorrelatedCopy(exchange, false); copy.getExchangeExtension().setTransacted(exchange.isTransacted()); @@ -350,10 +353,7 @@ public class RecipientListProcessor extends MulticastProcessor { // If we are in a transaction, set TRANSACTION_CONTEXT_DATA property for new exchanges to share txData // during the transaction. - if (exchange.isTransacted() && copy.getProperty(Exchange.TRANSACTION_CONTEXT_DATA) == null) { - if (txData == null) { - txData = new ConcurrentHashMap<>(); - } + if (txData != null && copy.getProperty(Exchange.TRANSACTION_CONTEXT_DATA) == null) { copy.setProperty(Exchange.TRANSACTION_CONTEXT_DATA, txData); } diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/Splitter.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/Splitter.java index 050872386d75..c00f2c13045f 100644 --- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/Splitter.java +++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/Splitter.java @@ -292,8 +292,7 @@ public class Splitter extends MulticastProcessor { // tracks individual (raw) item count, independent of grouping private final AtomicInteger rawItemCount = new AtomicInteger(); // tracks whether the primary (processing) iterator has been created; - // subsequent iterators (e.g. the drain in MulticastProcessor.doDone) - // must not update the watermark count (CAMEL-24139) + // any subsequent iterator must not update the watermark count (CAMEL-24139) private boolean primaryIteratorCreated; private SplitterIterable(Exchange exchange, Object value) { @@ -363,7 +362,7 @@ public class Splitter extends MulticastProcessor { @Override public Iterator<ProcessorExchangePair> iterator() { // only the first (primary) iterator tracks watermark count; - // subsequent iterators (drain in doDone) must not inflate it (CAMEL-24139) + // subsequent iterators must not inflate it (CAMEL-24139) boolean isPrimary = !primaryIteratorCreated; primaryIteratorCreated = true; @@ -440,7 +439,7 @@ public class Splitter extends MulticastProcessor { } } // eagerly update watermark count for items actually routed (primary iterator only) - // so the drain loop in MulticastProcessor.doDone cannot inflate it (CAMEL-24139) + // so a subsequent iterator cannot inflate it (CAMEL-24139) if (isPrimary && resumeStrategy != null && watermarkKey != null && watermarkExpression == null) { original.setProperty(SPLIT_WATERMARK_COUNT, rawItemCount.get()); } diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/RecipientListInvalidEndpointReleaseTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/RecipientListInvalidEndpointReleaseTest.java new file mode 100644 index 000000000000..301c176d4bd8 --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/processor/RecipientListInvalidEndpointReleaseTest.java @@ -0,0 +1,94 @@ +/* + * 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.atomic.AtomicInteger; + +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.assertThrows; + +/** + * When a recipient cannot be resolved, the producers already acquired for the recipients before it must be released. + */ +public class RecipientListInvalidEndpointReleaseTest extends ContextTestSupport { + + private final AtomicInteger started = new AtomicInteger(); + private final AtomicInteger stopped = new AtomicInteger(); + + @Test + public void testProducersReleasedWhenLaterRecipientIsInvalid() { + assertThrows(Exception.class, () -> template.sendBody("direct:start", "Hello")); + + assertEquals(1, started.get()); + assertEquals(1, stopped.get(), "the producer of the prototype recipient should be released and stopped"); + } + + @Override + protected RouteBuilder createRouteBuilder() { + context.addComponent("track", new DefaultComponent() { + @Override + protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) { + return new DefaultEndpoint(uri, this) { + @Override + public Producer createProducer() { + return new DefaultProducer(this) { + @Override + public void process(Exchange exchange) { + // noop + } + + @Override + protected void doStart() { + started.incrementAndGet(); + } + + @Override + protected void doStop() { + stopped.incrementAndGet(); + } + }; + } + + @Override + public Consumer createConsumer(Processor processor) { + throw new UnsupportedOperationException(); + } + }; + } + }); + + return new RouteBuilder() { + @Override + public void configure() { + from("direct:start").recipientList(constant("track:a,unknownxyz:b")).cacheSize(-1); + } + }; + } +} diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/RecipientListTransactedContextDataTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/RecipientListTransactedContextDataTest.java new file mode 100644 index 000000000000..b47c479b1e9b --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/processor/RecipientListTransactedContextDataTest.java @@ -0,0 +1,64 @@ +/* + * 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.ArrayList; +import java.util.List; +import java.util.Map; + +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Exchange; +import org.apache.camel.builder.RouteBuilder; +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.assertNotSame; + +/** + * Each transaction must get its own transaction context data when a transacted exchange goes through a recipient list. + */ +public class RecipientListTransactedContextDataTest extends ContextTestSupport { + + private final List<Map<?, ?>> data = new ArrayList<>(); + + @Test + public void testTransactionContextDataIsNotShared() { + template.sendBody("direct:start", "A"); + template.sendBody("direct:start", "B"); + + assertEquals(2, data.size()); + assertNotNull(data.get(0)); + assertNotNull(data.get(1)); + assertNotSame(data.get(0), data.get(1), "two transactions must not share the transaction context data"); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("direct:start") + .process(e -> e.getExchangeExtension().setTransacted(true)) + .recipientList(constant("direct:a")); + + from("direct:a") + .process(e -> data.add(e.getProperty(Exchange.TRANSACTION_CONTEXT_DATA, Map.class))); + } + }; + } +} diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/SplitterStreamingStopOnExceptionReadAheadTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterStreamingStopOnExceptionReadAheadTest.java new file mode 100644 index 000000000000..349f1d7ffee5 --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterStreamingStopOnExceptionReadAheadTest.java @@ -0,0 +1,78 @@ +/* + * 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.concurrent.atomic.AtomicInteger; + +import org.apache.camel.ContextTestSupport; +import org.apache.camel.builder.RouteBuilder; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +/** + * A streaming split that stops on an exception must not read, prepare or create exchanges for the parts after the + * failure. + */ +public class SplitterStreamingStopOnExceptionReadAheadTest extends ContextTestSupport { + + private final AtomicInteger read = new AtomicInteger(); + private final AtomicInteger prepared = new AtomicInteger(); + + @Test + public void testStopOnExceptionDoesNotReadRemainingParts() { + Iterator<Integer> parts = new Iterator<>() { + private int next; + + @Override + public boolean hasNext() { + return next < 100; + } + + @Override + public Integer next() { + read.incrementAndGet(); + return next++; + } + }; + + assertThrows(Exception.class, () -> template.sendBody("direct:start", parts)); + + // part 0 is routed, part 1 fails and stops the split + assertEquals(2, prepared.get(), "onPrepare should only be called for the parts that are routed"); + assertEquals(2, read.get(), "the parts after the failure should not be read"); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("direct:start") + .split(body()).streaming().stopOnException().onPrepare(e -> prepared.incrementAndGet()) + .process(e -> { + if (e.getMessage().getBody(Integer.class) == 1) { + throw new IllegalArgumentException("Forced"); + } + }) + .end(); + } + }; + } +}
