This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 6469c548d6f7 CAMEL-25823: camel-core - Start Split and Multicast
output producers when the route starts (#28147)
6469c548d6f7 is described below
commit 6469c548d6f797fe441688e553273a5ae5c2e175
Author: Guillaume Nodet <[email protected]>
AuthorDate: Tue Nov 3 20:06:38 2026 +0100
CAMEL-25823: camel-core - Start Split and Multicast output producers when
the route starts (#28147)
Co-authored-by: Claude Sonnet 4.6 <[email protected]>
---
.../apache/camel/processor/MulticastProcessor.java | 50 +++-
.../errorhandler/RedeliveryErrorHandler.java | 23 ++
...pervisingRouteControllerMulticastSplitTest.java | 195 ++++++++++++++++
.../RouteStopCutOffCallerRouteTest.java | 169 ++++++++++++++
.../errorhandler/RouteStopCutOffLogNestedTest.java | 253 +++++++++++++++++++++
.../errorhandler/RouteStopCutOffLogTest.java | 116 +++++++++-
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 24 ++
7 files changed, 825 insertions(+), 5 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 7966de01cc42..71a8665f4595 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
@@ -55,6 +55,7 @@ import org.apache.camel.Traceable;
import org.apache.camel.processor.aggregate.ShareUnitOfWorkAggregationStrategy;
import org.apache.camel.processor.aggregate.UseOriginalAggregationStrategy;
import org.apache.camel.processor.errorhandler.ErrorHandlerSupport;
+import org.apache.camel.processor.errorhandler.RedeliveryErrorHandler;
import org.apache.camel.spi.AsyncProcessorAwaitManager;
import org.apache.camel.spi.ErrorHandlerAware;
import org.apache.camel.spi.IdAware;
@@ -705,6 +706,18 @@ public class MulticastProcessor extends
BaseProcessorSupport
return;
}
+ // Propagate the interrupted flag and the
cut-off-WARN-logged marker from the
+ // sub-exchange to the original exchange immediately,
before the outer error handler's
+ // callback fires. Only propagate when a stopped error
handler in a nested EIP
+ // (e.g. split inside multicast) already logged the
cut-off WARN -- identified by
+ // the CUTOFF_WARN_LOGGED property set by
RedeliveryErrorHandler.runNotAllowed().
+ // If interrupted was set by DirectProducer without
logging a WARN, we must NOT
+ // propagate: the outer error handler must still log
its WARN for this cut-off.
+ if
(exchange.getProperty(RedeliveryErrorHandler.CUTOFF_WARN_LOGGED, false,
Boolean.class)) {
+
original.getExchangeExtension().setInterrupted(true);
+
original.setProperty(RedeliveryErrorHandler.CUTOFF_WARN_LOGGED, Boolean.TRUE);
+ }
+
exchangeResult.accept(exchange);
// aggregate exchanges if any
@@ -834,6 +847,12 @@ public class MulticastProcessor extends
BaseProcessorSupport
// so we should set the failed exchange as the result and
be done
result.set(exchange);
}
+ // Propagate the interrupted flag and the cut-off-WARN-logged
marker from the
+ // sub-exchange to the original before doDone() fires.
+ if
(exchange.getProperty(RedeliveryErrorHandler.CUTOFF_WARN_LOGGED, false,
Boolean.class)) {
+ original.getExchangeExtension().setInterrupted(true);
+
original.setProperty(RedeliveryErrorHandler.CUTOFF_WARN_LOGGED, Boolean.TRUE);
+ }
// and do the done work
doDone(exchange, true);
return false;
@@ -998,6 +1017,15 @@ public class MulticastProcessor extends
BaseProcessorSupport
original.setProperty(ExchangePropertyKey.CORRELATION_ID,
correlationId);
}
}
+ // Propagate the interrupted flag and the cut-off-WARN-logged
marker from the sub-exchange to
+ // the original. Only propagate when a stopped error handler in a
nested EIP already logged the
+ // cut-off WARN (identified by CUTOFF_WARN_LOGGED set by
RedeliveryErrorHandler.runNotAllowed()).
+ // DirectProducer sets interrupted without logging, so its
interruption must NOT be forwarded here
+ // -- the outer handler must log the WARN for this cut-off.
+ if
(subExchange.getProperty(RedeliveryErrorHandler.CUTOFF_WARN_LOGGED, false,
Boolean.class)) {
+ original.getExchangeExtension().setInterrupted(true);
+
original.setProperty(RedeliveryErrorHandler.CUTOFF_WARN_LOGGED, Boolean.TRUE);
+ }
}
if (processorExchangeFactory != null && pairs instanceof Collection) {
@@ -1238,8 +1266,13 @@ public class MulticastProcessor extends
BaseProcessorSupport
// and wrap in unit of work processor so the copy exchange
also can run under UoW
answer = createUnitOfWorkProcessor(route, processor, exchange);
- // must start the error handler
- ServiceHelper.startService(answer);
+ // Start before publishing to the cache: a second exchange to
the same new recipient
+ // can take the cache-hit path and use the wrapper before it
is started.
+ // isRunAllowed() covers STARTING/STARTED/SUSPENDING/SUSPENDED
phases; during doInit()
+ // (INITIALIZING) it is false, so wrappers built there are
deferred to doStart().
+ if (isRunAllowed()) {
+ ServiceHelper.startService(answer);
+ }
if (errorHandlers != null) {
errorHandlers.putIfAbsent(cacheKey, answer);
@@ -1323,6 +1356,11 @@ public class MulticastProcessor extends
BaseProcessorSupport
}
CamelContextAware.trySetCamelContext(aggregationStrategy,
camelContext);
ServiceHelper.startService(aggregationStrategy, processors,
processorExchangeFactory);
+ // start any error handlers that were pre-built during doInit() but
not yet started
+ // (they were deferred to avoid starting producers before CamelContext
is fully started)
+ if (errorHandlers != null) {
+ ServiceHelper.startService(errorHandlers.values());
+ }
}
/**
@@ -1344,12 +1382,16 @@ public class MulticastProcessor extends
BaseProcessorSupport
@Override
protected void doStop() throws Exception {
- ServiceHelper.stopService(processors, errorHandlers,
aggregationStrategy, processorExchangeFactory);
+ ServiceHelper.stopService(
+ processors, errorHandlers != null ? errorHandlers.values() :
null,
+ aggregationStrategy, processorExchangeFactory);
}
@Override
protected void doShutdown() throws Exception {
- ServiceHelper.stopAndShutdownServices(processors, errorHandlers,
aggregationStrategy, processorExchangeFactory);
+ ServiceHelper.stopAndShutdownServices(
+ processors, errorHandlers != null ? errorHandlers.values() :
null,
+ aggregationStrategy, processorExchangeFactory);
// only clear error handlers when shutting down
if (errorHandlers != null) {
errorHandlers.clear();
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/RedeliveryErrorHandler.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/RedeliveryErrorHandler.java
index 5c05406e6a86..10fbcc955210 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/RedeliveryErrorHandler.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/RedeliveryErrorHandler.java
@@ -80,6 +80,17 @@ public abstract class RedeliveryErrorHandler extends
ErrorHandlerSupport
private static final Logger LOG =
LoggerFactory.getLogger(RedeliveryErrorHandler.class);
+ /**
+ * Exchange property key (internal): set to {@code Boolean.TRUE} by {@link
SimpleTask#runNotAllowed()} immediately
+ * after it emits the cut-off WARN. Propagated across sub-exchange
boundaries by
+ * {@link org.apache.camel.processor.MulticastProcessor} so that outer
error handlers know a WARN was already logged
+ * for this cut-off and must not log a second one.
+ */
+ // Internal on purpose: a plain property, not an ExchangePropertyKey or a
constant on Exchange, so it is not public
+ // API. As any plain property, it shows in Exchange.getProperties() and in
tracer or debugger listings, and
+ // removeProperties("*") removes it.
+ public static final String CUTOFF_WARN_LOGGED =
"CamelErrorHandlerCutOffLogged";
+
// the exception of a failure that was continued (onException continued),
which is not a previous exception of a
// later failure. A key of its own: Exchange.EXCEPTION_HANDLED is the
doCatch's marker that a previous doCatch of
// the doTry handled the exception, and a direct call shares the exchange
with the caller's doTry.
@@ -990,6 +1001,18 @@ public abstract class RedeliveryErrorHandler extends
ErrorHandlerSupport
LOG.trace("Run not allowed, will reject executing exchange: {}",
exchange);
if (exchange.getException() == null) {
exchange.setException(new
RouteStoppingException(notAllowedReason()));
+ } else if (logger != null &&
!Boolean.TRUE.equals(exchange.getProperty(CUTOFF_WARN_LOGGED))) {
+ // log the cut-off WARN when the exchange already has an
exception and a WARN has not yet
+ // been logged for this cut-off: logCutOffByRouteStop guards
for instanceof RouteStoppingException
+ // and is a no-op for other exceptions. The exchange is
passing through a stopped error handler
+ // (e.g. in MulticastProcessor.doStop()) while inflight. Mark
CUTOFF_WARN_LOGGED after logging
+ // so that nested error handlers (e.g. a split inside a
filter) do not each log their own WARN
+ // for the same exchange. Note: isInterrupted() is NOT used as
the gate here — it may already
+ // be true from DirectProducer (CAMEL-25502) even when no WARN
has been emitted yet.
+ if (logCutOffByRouteStop(exchange, exchange.getException(),
logger.getLevel())) {
+ exchange.getExchangeExtension().setInterrupted(true);
+ exchange.setProperty(CUTOFF_WARN_LOGGED, Boolean.TRUE);
+ }
}
AsyncCallback cb = callback;
taskFactory.release(this);
diff --git
a/core/camel-core/src/test/java/org/apache/camel/impl/engine/DefaultSupervisingRouteControllerMulticastSplitTest.java
b/core/camel-core/src/test/java/org/apache/camel/impl/engine/DefaultSupervisingRouteControllerMulticastSplitTest.java
new file mode 100644
index 000000000000..d080aab40d70
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/impl/engine/DefaultSupervisingRouteControllerMulticastSplitTest.java
@@ -0,0 +1,195 @@
+/*
+ * 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.impl.engine;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Component;
+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.ServiceStatus;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.spi.SupervisingRouteController;
+import org.apache.camel.support.DefaultComponent;
+import org.apache.camel.support.DefaultConsumer;
+import org.apache.camel.support.DefaultEndpoint;
+import org.apache.camel.support.DefaultProducer;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Tests that Split and Multicast EIPs do not start child producers during
doInit(), which would bypass the
+ * SupervisingRouteController retry logic and fail the entire context when a
producer cannot connect yet.
+ * <p>
+ * CAMEL-25823: Split and Multicast start their child producers before
CamelContext starts, so with the supervising
+ * route controller a producer that cannot start yet fails the whole context.
+ */
+class DefaultSupervisingRouteControllerMulticastSplitTest extends
ContextTestSupport {
+
+ @Override
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ SupervisingRouteController src =
context.getRouteController().supervising();
+ src.setBackOffDelay(25);
+ src.setBackOffMaxAttempts(10);
+ src.setInitialDelay(10);
+ src.setThreadPoolSize(2);
+ return context;
+ }
+
+ @Test
+ void testSplitWithFlakyProducer() throws Exception {
+ FlakyComponent component = new FlakyComponent(2);
+ context.addComponent("flaky", component);
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:split")
+ .routeId("split-route")
+ .split(body())
+ .to("flaky:target");
+ }
+ });
+
+ // Must not throw - supervising route controller should retry
+ context.start();
+
+ // Route should eventually reach Started state after retries
+ await().atMost(Duration.ofSeconds(5))
+ .untilAsserted(() -> assertEquals(ServiceStatus.Started,
+
context.getRouteController().getRouteStatus("split-route")));
+
+ // Verify the producer was retried (failCount=2 means 3 start attempts
total)
+ FlakyEndpoint endpoint = (FlakyEndpoint)
context.getEndpoint("flaky:target");
+ assertEquals(3, endpoint.startAttempts.get(), "Expected 3 start
attempts (2 failures + 1 success)");
+
+ // Verify that a message can actually be processed once the route is
started
+ template.sendBody("direct:split", "a,b,c");
+ assertEquals(3, endpoint.processCount.get(), "Expected 3 process calls
for 3 split tokens");
+ }
+
+ @Test
+ void testMulticastWithFlakyProducer() throws Exception {
+ FlakyComponent component = new FlakyComponent(2);
+ context.addComponent("flaky", component);
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:multicast")
+ .routeId("multicast-route")
+ .multicast()
+ .to("flaky:target");
+ }
+ });
+
+ // Must not throw - supervising route controller should retry
+ context.start();
+
+ // Route should eventually reach Started state after retries
+ await().atMost(Duration.ofSeconds(5))
+ .untilAsserted(() -> assertEquals(ServiceStatus.Started,
+
context.getRouteController().getRouteStatus("multicast-route")));
+
+ // Verify the producer was retried (failCount=2 means 3 start attempts
total)
+ FlakyEndpoint endpoint = (FlakyEndpoint)
context.getEndpoint("flaky:target");
+ assertEquals(3, endpoint.startAttempts.get(), "Expected 3 start
attempts (2 failures + 1 success)");
+
+ // Verify that a message can actually be processed once the route is
started
+ template.sendBody("direct:multicast", "hello");
+ assertEquals(1, endpoint.processCount.get(), "Expected 1 process call
for multicast to single target");
+ }
+
+ /**
+ * A component whose producer fails its first {@code failCount} doStart()
calls with an exception, simulating a
+ * broker that is temporarily unavailable. The failure count is tracked on
the endpoint (shared across producer
+ * instances) so retries eventually succeed even when a new producer is
created on each restart attempt.
+ */
+ static class FlakyComponent extends DefaultComponent {
+
+ private final int failCount;
+
+ FlakyComponent(int failCount) {
+ this.failCount = failCount;
+ }
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining,
Map<String, Object> parameters) {
+ return new FlakyEndpoint(uri, this, failCount);
+ }
+ }
+
+ static class FlakyEndpoint extends DefaultEndpoint {
+
+ final AtomicInteger startAttempts = new AtomicInteger();
+ final AtomicInteger processCount = new AtomicInteger();
+ private final int failCount;
+
+ FlakyEndpoint(String uri, Component component, int failCount) {
+ super(uri, component);
+ this.failCount = failCount;
+ }
+
+ @Override
+ public Producer createProducer() {
+ return new FlakyProducer(this);
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) {
+ return new DefaultConsumer(this, processor);
+ }
+
+ boolean shouldFail() {
+ return startAttempts.incrementAndGet() <= failCount;
+ }
+ }
+
+ static class FlakyProducer extends DefaultProducer {
+
+ FlakyProducer(FlakyEndpoint endpoint) {
+ super(endpoint);
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ FlakyEndpoint ep = (FlakyEndpoint) getEndpoint();
+ if (ep.shouldFail()) {
+ throw new IllegalStateException("Cannot connect");
+ }
+ }
+
+ @Override
+ public void process(Exchange exchange) {
+ FlakyEndpoint ep = (FlakyEndpoint) getEndpoint();
+ ep.processCount.incrementAndGet();
+ }
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/errorhandler/RouteStopCutOffCallerRouteTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/errorhandler/RouteStopCutOffCallerRouteTest.java
new file mode 100644
index 000000000000..1e9dfe3287bf
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/errorhandler/RouteStopCutOffCallerRouteTest.java
@@ -0,0 +1,169 @@
+/*
+ * 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.errorhandler;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.RouteStoppingException;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+/**
+ * CAMEL-25823: when route B is force-stopped while an inflight exchange is
inside a nested EIP (split inside
+ * multicast), the {@code .process()} step's own stopped channel handler
re-enters {@code run()}, which sets a
+ * {@code RouteStoppingException} on the exchange (exchange had no exception
yet). The next outer stopped handler finds
+ * the exception already set, logs the cut-off WARN, and marks the exchange as
interrupted
+ * ({@code exchange.getExchangeExtension().setInterrupted(true)}). A calling
route A that is still running and passed
+ * the same exchange into route B over {@code direct} then also skips its
{@code onException}, redelivery, and dead
+ * letter channel — the same behaviour as when {@code DirectProducer} is cut
off while waiting for a consumer
+ * (CAMEL-25502). This test pins down that behaviour: the DLC mock is NOT
called when the nested-EIP cut-off propagates
+ * back to the caller.
+ *
+ * <p>
+ * Design: a {@code .process()} inside route B's nested EIP counts down an
{@code entered} latch and then waits on a
+ * {@code release} latch (with a timeout). The test waits for {@code entered},
force-stops route B (setting its
+ * error-handler wrappers to stopped state), then releases the latch. The
processor exits normally; the
+ * {@code .process()} step's own stopped channel handler re-enters {@code
run()}, which sets a
+ * {@code RouteStoppingException}. The next outer stopped handler finds the
exception already set, logs the WARN, and
+ * marks {@code interrupted=true}. Route A is kept running throughout; it has
a DLC so it runs a {@code RedeliveryTask}
+ * whose {@code isDone()} path sees the interrupted flag and skips redelivery
and the dead letter channel.
+ */
+public class RouteStopCutOffCallerRouteTest extends ContextTestSupport {
+
+ private final CountDownLatch entered = new CountDownLatch(1);
+ private final CountDownLatch release = new CountDownLatch(1);
+
+ /**
+ * Route A has a dead letter channel and sends to route B over {@code
direct}. Route B is stopped inside a split
+ * nested in a multicast while a {@code .process()} is blocked on a latch.
After the stop, the latch is released:
+ * the processor exits normally and the {@code .process()} step's own
stopped channel handler re-enters
+ * {@code run()}, which sets a {@code RouteStoppingException}. The next
outer stopped handler finds the exception
+ * already set, logs the WARN, and marks {@code interrupted=true}. Route A
is kept running; it has a DLC so it runs
+ * a {@code RedeliveryTask} whose {@code isDone()} path sees the
interrupted flag and skips redelivery,
+ * {@code onException}, and the dead letter channel.
+ */
+ @Test
+ public void callerRouteDlcNotCalledOnNestedEipCutOff() throws Exception {
+ MockEndpoint dlc = getMockEndpoint("mock:dlc");
+ dlc.expectedMessageCount(0);
+
+ MockEndpoint afterBlock = getMockEndpoint("mock:afterBlock");
+ afterBlock.expectedMessageCount(0);
+
+ // Route A sends synchronously to route B via direct (same exchange
object).
+ // The Future lets us wait for the exchange to fully propagate back.
+ // Set a short shutdown timeout so stopRoute("nested-b") force-stops
after 1 s instead of
+ // the default 300 s. route B is a direct: consumer with no executor
of its own; the inflight
+ // thread belongs to seda:caller (route A) and is not interrupted by
the force-stop.
+ context.getShutdownStrategy().setTimeout(1);
+ context.getShutdownStrategy().setTimeUnit(TimeUnit.SECONDS);
+
+ Future<Object> future = template.asyncRequestBody("seda:caller", "x",
Object.class);
+
+ // Wait until the processor inside route B's nested EIP has entered
and is blocking.
+ assertTrue(entered.await(5, TimeUnit.SECONDS), "Processor did not
enter within 5 seconds");
+
+ // 1. Stop route B: waits up to 1 s for the inflight exchange, then
force-stops.
+ // Force-stop calls MulticastProcessor.doStop(), which stops the
nested error-handler wrappers
+ // (isStoppingOrStopped() returns true). The seda:caller thread is
NOT interrupted.
+ // Route A is NOT stopped — it stays running throughout the test.
+ context.getRouteController().stopRoute("nested-b");
+
+ // 2. Release the latch: the processor exits normally (no interrupt,
no manually thrown exception).
+ // The .process() step's own stopped channel handler re-enters
run(), which sets a
+ // RouteStoppingException. The next outer stopped handler finds the
exception already set, logs
+ // the WARN, and marks interrupted=true. Route A has a DLC so it
runs a RedeliveryTask
+ // whose isDone() path sees the interrupted flag and skips
redelivery and the dead letter channel.
+ release.countDown();
+
+ // Wait for the exchange to finish. It should complete with a
RouteStoppingException
+ // (route A has a DLC so it takes the RedeliveryTask.isDone() path,
which keeps the exception
+ // and fires the callback without invoking the DLC).
+ ExecutionException ex = null;
+ try {
+ future.get(10, TimeUnit.SECONDS);
+ } catch (ExecutionException e) {
+ ex = e;
+ } catch (TimeoutException e) {
+ fail("Exchange did not complete within 10 seconds — likely a hang
in the error handler chain", e);
+ }
+ assertNotNull(ex, "Expected the exchange to fail with a
RouteStoppingException");
+ assertInstanceOf(CamelExecutionException.class, ex.getCause());
+ assertInstanceOf(RouteStoppingException.class,
ex.getCause().getCause());
+
+ // mock:afterBlock must NOT have been called: the stopped
error-handler wrapper intercepted
+ // the exchange before it reached the endpoint.
+ afterBlock.assertIsSatisfied();
+
+ // The DLC must NOT have been called: route A's error handler skipped
it because the exchange
+ // was marked as interrupted by route B's nested stopped error handler.
+ dlc.assertIsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // Route A: dead letter channel configured; calls route B via
direct (same exchange object).
+ // Route A is NOT stopped during the test — it stays running
throughout.
+ from("seda:caller").routeId("caller-a")
+
.errorHandler(deadLetterChannel("mock:dlc").maximumRedeliveries(0))
+ .to("direct:b");
+
+ // Route B: split nested inside multicast. The processor
counts down the entered latch
+ // and then waits on the release latch (with a timeout so the
test cannot hang).
+ // After the processor exits, the .process() step's own
stopped channel handler
+ // re-enters run(), which sets a RouteStoppingException. The
next outer stopped handler
+ // finds the exception already set, logs the WARN, and marks
interrupted=true.
+ // MulticastProcessor propagates interrupted=true to the
original exchange,
+ // so route A's running error handler
(RedeliveryTask.isDone()) skips the DLC.
+ from("direct:b").routeId("nested-b")
+ .multicast()
+ .split(body())
+ .process(new Processor() {
+ @Override
+ public void process(Exchange exchange) throws
Exception {
+ entered.countDown();
+ // Block until the test releases, or until 30
s to avoid
+ // hanging the test suite if something goes
wrong.
+ if (!release.await(30, TimeUnit.SECONDS)) {
+ fail("Release latch timed out after 30
seconds");
+ }
+ }
+ })
+ .to("mock:afterBlock").id("afterBlock")
+ .end()
+ .end();
+ }
+ };
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/errorhandler/RouteStopCutOffLogNestedTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/errorhandler/RouteStopCutOffLogNestedTest.java
new file mode 100644
index 000000000000..07782de8bbde
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/errorhandler/RouteStopCutOffLogNestedTest.java
@@ -0,0 +1,253 @@
+/*
+ * 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.errorhandler;
+
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.log.ConsumingAppender;
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.core.LoggerContext;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+/**
+ * CAMEL-25823: verifies that a route-stop cut-off produces exactly one WARN
line even when the exchange passes through
+ * multiple nested stopped error handlers. Two scenarios are tested:
+ *
+ * <ol>
+ * <li>{@code filter} — a filter wrapping a {@code .process()} step gives the
exchange two stopped error-handler
+ * channels in route B (the {@code .process()} step's channel and the filter's
channel) plus route A's running channel.
+ * The {@code setInterrupted(true)} marker set by the filter's stopped channel
prevents route A's running handler from
+ * logging the same cut-off again.</li>
+ * <li>{@code multicast+split} — a {@code split} nested inside a {@code
multicast} creates a sub-exchange copy.
+ * {@code MulticastProcessor} propagates the {@code interrupted} flag from the
sub-exchange to the original so that
+ * neither the original exchange's channel nor the caller's error handler logs
a second WARN.</li>
+ * </ol>
+ *
+ * <p>
+ * Design (both tests): route A ({@code seda:start} or {@code seda:start-ms})
forwards to route B via {@code direct}
+ * (same exchange object, no thread switch). A {@code .process()} inside B's
nested EIP counts down an {@code entered}
+ * latch and waits on a {@code release} latch. The test waits for {@code
entered}, stops route B (whose error-handler
+ * wrappers are now in stopped state), then releases the latch. The processor
exits normally; the {@code .process()}
+ * step's own stopped channel handler re-enters {@code run()}, which sets the
{@code RouteStoppingException} on the
+ * (sub-)exchange. The next stopped handler logs the WARN and marks {@code
interrupted=true}. Subsequent stopped
+ * handlers and the caller's running handler see the flag and skip. The WARN
count is asserted after the exchange
+ * completes.
+ */
+public class RouteStopCutOffLogNestedTest extends ContextTestSupport {
+
+ private static final String LOGGER = DefaultErrorHandler.class.getName();
+
+ private final Queue<String> messages = new ConcurrentLinkedQueue<>();
+ private final CountDownLatch entered = new CountDownLatch(1);
+ private final CountDownLatch release = new CountDownLatch(1);
+
+ @AfterEach
+ public void removeAppender() {
+ LoggerContext ctx = (LoggerContext) LogManager.getContext(false);
+ ctx.getConfiguration().removeLogger(LOGGER);
+ ctx.updateLoggers();
+ }
+
+ /**
+ * A filter wrapping the blocked processor gives the exchange two stopped
error-handler channels in route B (the
+ * {@code .process()} step's channel and the filter's channel). The
single-WARN guarantee should hold: only one WARN
+ * line per exchange, not one per stopped channel or per route.
+ */
+ @Test
+ public void nestedEipCutOffIsOneWarnLine() throws Exception {
+ // The tests in this module run in parallel, and log through the same
error handler logger:
+ // only keep the events from the threads of this CamelContext.
+ String threadMarker = "(" + context.getName() + ")";
+ ConsumingAppender.newAppender(LOGGER, "RouteStopCutOffLogNestedTest",
Level.DEBUG, event -> {
+ if (event.getThreadName() != null &&
event.getThreadName().contains(threadMarker)) {
+ messages.add(event.getLevel() + " " +
event.getMessage().getFormattedMessage()
+ + (event.getThrown() != null ? " THROWN" : ""));
+ }
+ });
+ // A short shutdown timeout ensures stopRoute force-stops route B
after 1 s.
+ // Route B is a direct: consumer with no executor thread of its own;
the seda:start
+ // consumer thread is NOT interrupted when only route B is stopped.
+ context.getShutdownStrategy().setTimeout(1);
+ context.getShutdownStrategy().setTimeUnit(TimeUnit.SECONDS);
+
+ // Send asynchronously so the test thread is free to call stopRoute.
+ Future<Object> future = template.asyncRequestBody("seda:start", "x",
Object.class);
+
+ // Wait until the processor inside route B's nested EIP has entered
and is blocking.
+ assertTrue(entered.await(5, TimeUnit.SECONDS), "Processor did not
enter within 5 seconds");
+
+ // Stop route B: stopRouteService stops B's error-handler wrappers
after the 1 s force-stop.
+ // The seda:start consumer thread is NOT interrupted (it owns no
thread in route B).
+ context.getRouteController().stopRoute("nested");
+
+ // Release the latch: the processor exits normally (no interrupt, no
manually thrown exception).
+ // The .process() step's own stopped channel handler re-enters run(),
which sets a
+ // RouteStoppingException. The filter's stopped channel handler finds
the exception already set,
+ // logs the WARN, and marks interrupted=true. Route A's running
channel sees the flag and skips.
+ release.countDown();
+
+ // Wait for the exchange to complete fully.
+ try {
+ future.get(10, TimeUnit.SECONDS);
+ } catch (ExecutionException e) {
+ // expected: RouteStoppingException propagated back
+ } catch (TimeoutException e) {
+ fail("Exchange did not complete within 10 seconds — likely a hang
in the error handler chain", e);
+ }
+
+ // Exactly one WARN should be emitted, even though the exchange passed
through two nested
+ // stopped error-handler channels in route B (process step, filter)
plus route A's running channel.
+ assertEquals(1, messages.size(), messages.toString());
+ String message = messages.peek();
+ assertTrue(message.startsWith("WARN Exchange cut off (MessageId: "),
message);
+ assertTrue(message.contains(": its route is being stopped or reloaded,
so it is not continued."), message);
+ assertFalse(message.contains("Exhausted"), message);
+ assertFalse(message.contains("Message History"), message);
+ assertFalse(message.endsWith("THROWN"), message);
+ }
+
+ /**
+ * A {@code split} nested inside a {@code multicast} creates a
sub-exchange copy for each split item.
+ * {@code MulticastProcessor} propagates the {@code interrupted} flag from
the sub-exchange to the original
+ * exchange, so the original exchange's channel and route A's running
error handler also skip logging. The
+ * single-WARN guarantee holds across the sub-exchange copy boundary.
+ */
+ @Test
+ public void multicastSplitCutOffIsOneWarnLine() throws Exception {
+ String threadMarker = "(" + context.getName() + ")";
+ ConsumingAppender.newAppender(LOGGER, "RouteStopCutOffLogNestedTest",
Level.DEBUG, event -> {
+ if (event.getThreadName() != null &&
event.getThreadName().contains(threadMarker)) {
+ messages.add(event.getLevel() + " " +
event.getMessage().getFormattedMessage()
+ + (event.getThrown() != null ? " THROWN" : ""));
+ }
+ });
+ context.getShutdownStrategy().setTimeout(1);
+ context.getShutdownStrategy().setTimeUnit(TimeUnit.SECONDS);
+
+ // Send a single-item body so the split produces exactly one
sub-exchange.
+ Future<Object> future = template.asyncRequestBody("seda:start-ms",
"x", Object.class);
+
+ assertTrue(entered.await(5, TimeUnit.SECONDS), "Processor did not
enter within 5 seconds");
+
+ // Stop route B: MulticastProcessor.doStop() stops the nested
error-handler wrappers.
+ // The seda:start-ms consumer thread is NOT interrupted.
+ context.getRouteController().stopRoute("nested-ms");
+
+ // Release the latch: the processor exits normally. The .process()
step's stopped channel
+ // handler sets the RouteStoppingException on the sub-exchange. The
split's stopped channel
+ // handler logs the WARN and marks the sub-exchange interrupted=true.
MulticastProcessor
+ // propagates interrupted=true to the original exchange. The
multicast's channel and route A's
+ // running channel see the flag and skip logging.
+ release.countDown();
+
+ try {
+ future.get(10, TimeUnit.SECONDS);
+ } catch (ExecutionException e) {
+ // expected: RouteStoppingException propagated back
+ } catch (TimeoutException e) {
+ fail("Exchange did not complete within 10 seconds — likely a hang
in the error handler chain", e);
+ }
+
+ // Exactly one WARN even though the cut-off crosses a sub-exchange
copy boundary.
+ assertEquals(1, messages.size(), messages.toString());
+ String message = messages.peek();
+ assertTrue(message.startsWith("WARN Exchange cut off (MessageId: "),
message);
+ assertTrue(message.contains(": its route is being stopped or reloaded,
so it is not continued."), message);
+ assertFalse(message.contains("Exhausted"), message);
+ assertFalse(message.contains("Message History"), message);
+ assertFalse(message.endsWith("THROWN"), message);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // Route A (filter test): forwards to route B via direct (same
exchange object, no thread switch).
+ // Route A is NOT stopped during the test — it stays running
throughout.
+ from("seda:start").routeId("caller-nested")
+ .to("direct:nested");
+
+ // Route B (filter test): a filter wrapping the processor
gives the exchange two stopped
+ // error-handler channels in route B. The processor counts
down the entered latch and
+ // waits on the release latch. After the processor exits, the
.process() step's stopped
+ // channel sets the RouteStoppingException; the filter's
stopped channel logs the WARN
+ // and marks interrupted=true; route A's running channel sees
the flag and skips.
+ from("direct:nested").routeId("nested")
+ .filter(constant(true))
+ .process(new Processor() {
+ @Override
+ public void process(Exchange exchange) throws
Exception {
+ entered.countDown();
+ // Block until the test releases, or 30 s to
avoid hanging
+ // the test suite if something goes wrong.
+ if (!release.await(30, TimeUnit.SECONDS)) {
+ fail("Release latch timed out after 30
seconds");
+ }
+ }
+ })
+ .to("mock:afterBlock")
+ .end()
+ .to("mock:result");
+
+ // Route A (multicast+split test): forwards to route B via
direct (same exchange object).
+ // Route A is NOT stopped — it stays running throughout.
+ from("seda:start-ms").routeId("caller-nested-ms")
+ .to("direct:nested-ms");
+
+ // Route B (multicast+split test): a split nested inside a
multicast.
+ // The split creates a sub-exchange copy. After the processor
exits, the .process() step's
+ // stopped channel sets the RouteStoppingException on the
sub-exchange; the split's stopped
+ // channel logs the WARN and marks the sub-exchange
interrupted=true.
+ // MulticastProcessor propagates interrupted=true to the
original exchange,
+ // so the multicast's channel and route A's running channel
both skip logging.
+ from("direct:nested-ms").routeId("nested-ms")
+ .multicast()
+ .split(body())
+ .process(new Processor() {
+ @Override
+ public void process(Exchange exchange) throws
Exception {
+ entered.countDown();
+ if (!release.await(30, TimeUnit.SECONDS)) {
+ fail("Release latch timed out after 30
seconds");
+ }
+ }
+ })
+ .to("mock:afterBlock-ms")
+ .end()
+ .end();
+ }
+ };
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/errorhandler/RouteStopCutOffLogTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/errorhandler/RouteStopCutOffLogTest.java
index 4d33c34bee03..d8075554f11f 100644
---
a/core/camel-core/src/test/java/org/apache/camel/processor/errorhandler/RouteStopCutOffLogTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/errorhandler/RouteStopCutOffLogTest.java
@@ -18,10 +18,14 @@ package org.apache.camel.processor.errorhandler;
import java.util.List;
import java.util.Queue;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
import org.apache.camel.RouteStoppingException;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.log.ConsumingAppender;
@@ -34,7 +38,10 @@ import org.junit.jupiter.api.Test;
import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
/**
* CAMEL-25484: an exchange that a route stop or a dev mode reload cuts off is
logged as one WARN line, not as an
@@ -45,6 +52,8 @@ public class RouteStopCutOffLogTest extends
ContextTestSupport {
private static final String LOGGER = DefaultErrorHandler.class.getName();
private final Queue<String> messages = new ConcurrentLinkedQueue<>();
+ private final CountDownLatch entered = new CountDownLatch(1);
+ private final CountDownLatch release = new CountDownLatch(1);
@AfterEach
public void removeAppender() {
@@ -53,6 +62,10 @@ public class RouteStopCutOffLogTest extends
ContextTestSupport {
ctx.updateLoggers();
}
+ // ---- Original DirectProducer cut-off tests (CAMEL-25484 / CAMEL-25502)
----
+ // These use seda:start → split → direct:shipment (no consumer). The seda
thread is interrupted
+ // by stopRoute, which causes DirectProducer to set the
RouteStoppingException and log one WARN.
+
@Test
public void theCutOffIsOneWarnLine() throws Exception {
// the tests in this module run in parallel, and log through the same
error handler logger:
@@ -66,7 +79,6 @@ public class RouteStopCutOffLogTest extends
ContextTestSupport {
});
context.getShutdownStrategy().setTimeout(1);
context.getShutdownStrategy().setTimeUnit(TimeUnit.SECONDS);
-
context.getInflightRepository().setInflightBrowseEnabled(true);
template.sendBody("seda:start", List.of("A", "B"));
@@ -121,17 +133,119 @@ public class RouteStopCutOffLogTest extends
ContextTestSupport {
() -> context.getErrorRegistry().browse().toString());
}
+ // ---- Latch-based cut-off tests (this PR) ----
+ // These use direct:start → split with a processor that blocks. stopRoute
force-stops after the
+ // timeout, and the stopped error-handler chain emits exactly one cut-off
WARN.
+
+ @Test
+ public void processedCutOffIsOneWarnLine() throws Exception {
+ String threadMarker = "(" + context.getName() + ")";
+ ConsumingAppender.newAppender(LOGGER, "RouteStopCutOffLogTest",
Level.DEBUG, event -> {
+ if (event.getThreadName() != null &&
event.getThreadName().contains(threadMarker)) {
+ messages.add(event.getLevel() + " " +
event.getMessage().getFormattedMessage()
+ + (event.getThrown() != null ? " THROWN" : ""));
+ }
+ });
+ // Use a bounded shutdown timeout so that stopRoute force-stops the
route (stopping its
+ // error handlers) after the timeout, since the inflight exchange will
not complete on its own.
+ // Using direct: (not seda:) ensures the route stop does not interrupt
the processing thread.
+ context.getShutdownStrategy().setTimeout(5);
+ context.getShutdownStrategy().setTimeUnit(TimeUnit.SECONDS);
+
+ CompletableFuture<Exchange> future =
template.asyncSend("direct:start", exchange -> {
+ exchange.getIn().setBody(List.of("A"));
+ });
+
+ assertTrue(entered.await(5, TimeUnit.SECONDS), "Processor did not
enter within 5 seconds");
+ context.getRouteController().stopRoute("slow-latch");
+ release.countDown();
+
+ Exchange result = null;
+ try {
+ result = future.get(10, TimeUnit.SECONDS);
+ } catch (TimeoutException e) {
+ fail("Exchange did not complete within 10 seconds — likely a hang
in the error handler chain", e);
+ }
+ assertNotNull(result);
+ assertInstanceOf(RouteStoppingException.class, result.getException());
+
+ assertEquals(1, messages.size(), messages.toString());
+ String message = messages.peek();
+ assertTrue(message.startsWith("WARN Exchange cut off (MessageId: "),
message);
+ assertTrue(message.contains(": its route is being stopped or reloaded,
so it is not continued."), message);
+ assertFalse(message.contains("Exhausted"), message);
+ assertFalse(message.contains("Message History"), message);
+ assertFalse(message.endsWith("THROWN"), message);
+ }
+
+ @Test
+ public void processedCutOffIsNotInTheErrorRegistry() throws Exception {
+ context.getErrorRegistry().setEnabled(true);
+ context.getShutdownStrategy().setTimeout(5);
+ context.getShutdownStrategy().setTimeUnit(TimeUnit.SECONDS);
+
+ template.asyncSend("direct:start", exchange ->
exchange.getIn().setBody(List.of("A")));
+
+ assertTrue(entered.await(5, TimeUnit.SECONDS), "Processor did not
enter within 5 seconds");
+ context.getRouteController().stopRoute("slow-latch");
+ release.countDown();
+
+ await().atMost(10, TimeUnit.SECONDS)
+ .until(() ->
context.getRouteController().getRouteStatus("slow-latch").isStopped());
+ assertEquals(0, context.getErrorRegistry().size(), () ->
context.getErrorRegistry().browse().toString());
+ }
+
+ @Test
+ public void processedCutOffIsInTheErrorRegistryWhenIncluded() throws
Exception {
+ context.getErrorRegistry().setEnabled(true);
+ context.getErrorRegistry().setIncludeRouteStopping(true);
+ context.getShutdownStrategy().setTimeout(5);
+ context.getShutdownStrategy().setTimeUnit(TimeUnit.SECONDS);
+
+ template.asyncSend("direct:start", exchange ->
exchange.getIn().setBody(List.of("A")));
+
+ assertTrue(entered.await(5, TimeUnit.SECONDS), "Processor did not
enter within 5 seconds");
+ context.getRouteController().stopRoute("slow-latch");
+ release.countDown();
+
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
context.getErrorRegistry().size() > 0);
+ assertTrue(context.getErrorRegistry().browse().stream()
+ .allMatch(e ->
RouteStoppingException.class.getName().equals(e.getExceptionType())),
+ () -> context.getErrorRegistry().browse().toString());
+ }
+
@Override
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
@Override
public void configure() {
+ // Original scenario: DirectProducer cut-off (CAMEL-25484 /
CAMEL-25502)
from("seda:start").routeId("slow")
.split(body())
.log("Picked ${body}")
.to("direct:shipment").id("shipment")
.end()
.to("mock:result");
+
+ // Latch-based scenario: runNotAllowed() cut-off through
stopped error handler
+ from("direct:start").routeId("slow-latch")
+ .split(body())
+ .log("Picked ${body}")
+ // Block here so the test can stop the route while an
exchange is inflight
+ // inside the split. When the route is stopped, the
stopped error-handler
+ // chain emits exactly one cut-off WARN.
+ .process(exchange -> {
+ entered.countDown();
+ try {
+ if (!release.await(30, TimeUnit.SECONDS)) {
+ throw new IllegalStateException("release
latch timed out");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ })
+ .end()
+ .to("mock:result2");
}
};
}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 0054539bf3db..84ba80b8ff78 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -1266,6 +1266,30 @@ The generated Endpoint DSL header accessors on the Swift
`HeaderNameBuilder` are
`openstackSwiftVersionsLocation()`, `xContainerRead()` ->
`openstackSwiftContainerRead()`, `xContainerWrite()` ->
`openstackSwiftContainerWrite()`).
+=== camel-core - Split and Multicast child producers start with their route
+
+`Split` and `Multicast` EIPs now start their child output producers during the
route's own start, instead of during
+`CamelContext` initialization. The visible changes are:
+
+* The producers now start when the route starts, not during `CamelContext`
initialization. A `CamelContext` that is
+ only built or initialized (without starting routes) no longer starts them.
+* Each route's producers now start in `startupOrder` with their route, instead
of ahead of all routes.
+* The producers of an `autoStartup=false` route are no longer started when the
context starts; they start when the
+ route itself is started (regardless of which route controller is used).
+
+=== camel-core - Forced route-stop cut-off marks exchange as interrupted in
any nested EIP
+
+An exchange that a forced route stop cuts off inside a nested EIP (`split`,
`multicast`,
+`filter`, `choice`, ...) of the stopped route is now marked as interrupted
+(`exchange.getExchangeExtension().setInterrupted(true)`) after the cut-off
WARN is logged. This prevents
+nested stopped error handlers from each emitting their own WARN for the same
exchange. The interrupted flag
+also causes the error handler to skip redelivery, `onException` routes, and
the dead letter channel — the
+same behavior that `DirectProducer` applies (CAMEL-25502) when a route is
stopped while an exchange is
+waiting for a consumer. A calling route that is still running (for example
through `direct`) then also no
+longer runs its `onException`, redelivery or dead letter channel for that
exchange. If your application
+relied on `onException` or the dead letter channel handling a
`RouteStoppingException` from a nested EIP
+of the stopped route, it will no longer trigger on a forced route stop.
+
=== camel-opentelemetry-metrics - a counter increment or decrement that does
not give a number fails the exchange
A `counter` with the `increment` or `decrement` option (such as
`+increment=${body}+` or `+increment=${header.amount}+`)
