This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-24142 in repository https://gitbox.apache.org/repos/asf/camel.git
commit a666c9376108ff1d0ea825323d304bf5503d5e92 Author: Claus Ibsen <[email protected]> AuthorDate: Fri Jul 17 08:19:09 2026 +0200 CAMEL-24142: Fix Multicast/Split timeout silently lost when fired during aggregation Co-Authored-By: Claude Opus 4.6 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../apache/camel/processor/MulticastProcessor.java | 51 +++++++----- ...ticastParallelTimeoutDuringAggregationTest.java | 97 ++++++++++++++++++++++ 2 files changed, 125 insertions(+), 23 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 5a970717ccca..6ccea3d52aab 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 @@ -505,33 +505,38 @@ public class MulticastProcessor extends BaseProcessorSupport } 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 + // and the exchange would hang forever (CAMEL-24142) Lock lock = this.lock; - if (lock.tryLock()) { - try { - while (nbAggregated.get() < nbExchangeSent.get()) { - Exchange exchange = completion.pollUnordered(); - int index = exchange != null ? getExchangeIndex(exchange) : nbExchangeSent.get(); - while (nbAggregated.get() < index) { - int idx = nbAggregated.getAndIncrement(); - AggregationStrategy strategy = getAggregationStrategy(null); - if (strategy != null) { - strategy.timeout(result.get() != null ? result.get() : original, - idx, nbExchangeSent.get(), timeout); - } - } - if (exchange != null) { - doAggregate(result, exchange, original); - nbAggregated.incrementAndGet(); + lock.lock(); + try { + if (done.get()) { + return; + } + while (nbAggregated.get() < nbExchangeSent.get()) { + Exchange exchange = completion.pollUnordered(); + int index = exchange != null ? getExchangeIndex(exchange) : nbExchangeSent.get(); + while (nbAggregated.get() < index) { + int idx = nbAggregated.getAndIncrement(); + AggregationStrategy strategy = getAggregationStrategy(null); + if (strategy != null) { + strategy.timeout(result.get() != null ? result.get() : original, + idx, nbExchangeSent.get(), timeout); } } - doTimeoutDone(result.get(), true); - } catch (Exception e) { - original.setException(e); - // and do the done work - doTimeoutDone(null, false); - } finally { - lock.unlock(); + if (exchange != null) { + doAggregate(result, exchange, original); + nbAggregated.incrementAndGet(); + } } + doTimeoutDone(result.get(), true); + } catch (Exception e) { + original.setException(e); + // and do the done work + doTimeoutDone(null, false); + } finally { + lock.unlock(); } } diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/MulticastParallelTimeoutDuringAggregationTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/MulticastParallelTimeoutDuringAggregationTest.java new file mode 100644 index 000000000000..b4a8bdea6ae4 --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/processor/MulticastParallelTimeoutDuringAggregationTest.java @@ -0,0 +1,97 @@ +/* + * 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.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.apache.camel.AggregationStrategy; +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Exchange; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.parallel.Isolated; + +import static org.junit.jupiter.api.Assertions.assertNotNull; + +/** + * CAMEL-24142: Verifies that a multicast timeout still fires even when the timeout instant falls inside an ongoing + * aggregation (i.e. the aggregation lock is held). Before the fix, tryLock() in timeout() would silently return and the + * exchange would hang forever. + */ +@Isolated("Uses short timeouts and latches") +public class MulticastParallelTimeoutDuringAggregationTest extends ContextTestSupport { + + private final CountDownLatch aggregationProceed = new CountDownLatch(1); + private final CountDownLatch done = new CountDownLatch(1); + + @Test + public void testTimeoutNotLostDuringAggregation() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:result"); + mock.expectedMessageCount(1); + mock.setResultWaitTime(10000); + + template.sendBody("direct:start", "Hello"); + + mock.assertIsSatisfied(); + + Exchange out = mock.getReceivedExchanges().get(0); + assertNotNull(out, "Should have received an exchange from multicast timeout"); + + done.countDown(); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("direct:start") + .multicast(new SlowAggregationStrategy()) + .parallelProcessing() + .timeout(500) + .to("direct:fast", "direct:stuck") + .end() + .to("mock:result"); + + from("direct:fast").setBody(constant("A")); + + // block until test signals done — the timeout must handle this + from("direct:stuck").process(e -> done.await(30, TimeUnit.SECONDS)); + } + }; + } + + class SlowAggregationStrategy implements AggregationStrategy { + @Override + public Exchange aggregate(Exchange oldExchange, Exchange newExchange) { + if (oldExchange == null) { + try { + // hold the lock for 2s so the 500ms timeout fires in this window + aggregationProceed.await(2, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return newExchange; + } + String body = oldExchange.getIn().getBody(String.class); + oldExchange.getIn().setBody(body + newExchange.getIn().getBody(String.class)); + return oldExchange; + } + } +}
