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 4700e7f3c7aaa1a0f876d359ce882c32a14803c0 Author: smjain <[email protected]> AuthorDate: Wed Sep 23 19:49:19 2026 +0530 CAMEL-24959: camel-core - Do not complete a parallel streaming split before its parts when the iterator ends with a null Cause: an iterator may return true from hasNext() and then null from next() (CAMEL-9745, SplitIteratorNullTest). The Splitter skips such a null part. In parallel mode MulticastReactiveTask.run() re-schedules itself while hasNext() was true, so the last part sent is not known to be the last one. The next run then finds no pair and called doDone at once, while the parts already sent were still in progress. Sequential mode is not affected, as there the next run only starts after the previous part was aggregated. CAMEL-21114 fixed the same symptom for the transacted task. Effect: a split with streaming and parallelProcessing completed right away, while parts were still in progress: its result only had the parts aggregated so far (the original message when none was), and the remaining parts were processed after the route had moved on. Fix: when there are no more pairs, mark all as sent under the task lock, aggregate what is completed (aggregate() gives up when another thread holds the lock), and only be done if everything sent is aggregated. Otherwise aggregate() is done when the last part is aggregated. Co-Authored-By: Claude Opus 5.5 <[email protected]> --- .../apache/camel/processor/MulticastProcessor.java | 51 +++++++-- .../SplitParallelStreamingIteratorNullTest.java | 122 +++++++++++++++++++++ 2 files changed, 163 insertions(+), 10 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..cc0863fe8ddb 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 @@ -496,21 +496,52 @@ public class MulticastProcessor extends BaseProcessorSupport Lock lock = this.lock; if (lock.tryLock()) { try { - Exchange exchange; - while (!done.get() && (exchange = completion.poll()) != null) { - doAggregate(result, exchange, original); - if (nbAggregated.incrementAndGet() >= nbExchangeSent.get() && allSent.get()) { - doDone(result.get(), true); - } - } - } catch (Exception e) { - doFailed(e); + aggregateCompleted(); } finally { lock.unlock(); } } } + /** + * Aggregates the completed exchanges, and is done when all the exchanges are sent and aggregated. Must be + * called while holding the lock. + */ + private void aggregateCompleted() { + try { + Exchange exchange; + while (!done.get() && (exchange = completion.poll()) != null) { + doAggregate(result, exchange, original); + if (nbAggregated.incrementAndGet() >= nbExchangeSent.get() && allSent.get()) { + doDone(result.get(), true); + } + } + } catch (Exception e) { + doFailed(e); + } + } + + /** + * There are no more pairs to send, even though the last pair sent was not known to be the last one (some + * iterators return true from hasNext() and then null from next()). In parallel mode the exchanges sent may + * still be in progress, so mark all as sent, and only be done when they are all aggregated (the transacted task + * handles the same case since CAMEL-21114). + */ + protected void doDoneNoMorePairs() { + Lock lock = this.lock; + lock.lock(); + try { + allSent.set(true); + // aggregate() gives up when another thread holds the lock, so aggregate what is completed + aggregateCompleted(); + if (nbAggregated.get() >= nbExchangeSent.get()) { + doDone(result.get(), true); + } + } finally { + lock.unlock(); + } + } + protected void timeout() { // use lock() instead of tryLock() because timeout is a one-shot scheduled task // if tryLock fails (lock held by aggregate), the timeout would be silently lost @@ -630,7 +661,7 @@ public class MulticastProcessor extends BaseProcessorSupport // Get next processor exchange pair to sent, skipping null ones ProcessorExchangePair pair = getNextProcessorExchangePair(); if (pair == null) { - doDone(result.get(), true); + doDoneNoMorePairs(); return; } diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/SplitParallelStreamingIteratorNullTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/SplitParallelStreamingIteratorNullTest.java new file mode 100644 index 000000000000..dbd0b92025fe --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/processor/SplitParallelStreamingIteratorNullTest.java @@ -0,0 +1,122 @@ +/* + * 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.Arrays; +import java.util.Iterator; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.apache.camel.AggregationStrategy; +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.assertTrue; + +/** + * Tests that a streaming split waits for its parts and aggregates all of them when the iterator returns true from + * hasNext() and then null from next() (see {@link SplitIteratorNullTest}), also with parallel processing. + */ +public class SplitParallelStreamingIteratorNullTest extends ContextTestSupport { + + private final CountDownLatch lastHasNext = new CountDownLatch(1); + + @Test + public void testSplitStreamingParallel() { + String out = template.requestBody("direct:parallel", new MyIterator(lastHasNext), String.class); + + // the parts complete in any order + assertEquals("ABC", sorted(out), "The split should return the aggregated parts, but returned: " + out); + } + + @Test + public void testSplitStreaming() { + String out = template.requestBody("direct:sequential", new MyIterator(lastHasNext), String.class); + + assertEquals("ABC", out, "The split should return the aggregated parts, but returned: " + out); + } + + private static String sorted(String s) { + char[] chars = s.toCharArray(); + Arrays.sort(chars); + return new String(chars); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + AggregationStrategy concat = (oldExchange, newExchange) -> { + if (oldExchange == null) { + return newExchange; + } + oldExchange.getMessage().setBody( + oldExchange.getMessage().getBody(String.class) + newExchange.getMessage().getBody(String.class)); + return oldExchange; + }; + + from("direct:parallel") + .split(body(), concat).streaming().parallelProcessing() + // the parts wait until the splitter has found out there are no more parts + .process(e -> assertTrue(lastHasNext.await(10, TimeUnit.SECONDS))) + .end(); + + from("direct:sequential") + .split(body(), concat).streaming() + .log("${body}") + .end(); + } + }; + } + + private static class MyIterator implements Iterator<String> { + + private final CountDownLatch lastHasNext; + private int count = 4; + + MyIterator(CountDownLatch lastHasNext) { + this.lastHasNext = lastHasNext; + } + + @Override + public boolean hasNext() { + // we return true one extra time, and cause next to return null + boolean answer = count > 0; + if (!answer) { + lastHasNext.countDown(); + } + return answer; + } + + @Override + public String next() { + count--; + if (count == 0) { + return null; + } else if (count == 1) { + return "C"; + } else if (count == 2) { + return "B"; + } else { + return "A"; + } + } + } +}
