CAMEL-25823: camel-core - Start Split and Multicast output producers when the route starts

core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java

@@ -1210,8 +1210,13 @@ protected Processor wrapInErrorHandler(Route route, Exchange exchange, Processor
                 // 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(key, answer);
@@ -1278,6 +1283,11 @@ protected void doStart() throws Exception {
         }
         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());
+        }
     }
 
     /**
@@ -1299,12 +1309,16 @@ protected ExecutorService createAggregateExecutorService(String name) {
 
     @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();

core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/RedeliveryErrorHandler.java

@@ -860,6 +860,12 @@ private void runNotAllowed() {
             LOG.trace("Run not allowed, will reject executing exchange: {}", exchange);
             if (exchange.getException() == null) {
                 exchange.setException(new RouteStoppingException(notAllowedReason()));
+            } else if (!exchange.getExchangeExtension().isInterrupted()) {
+                // log the cut-off WARN when the exchange already has a RouteStoppingException and
+                // is not marked as interrupted: this is the route-level exchange whose error handler
+                // was stopped (e.g. in MulticastProcessor.doStop()) while a sub-exchange was inflight;
+                // interrupted sub-exchanges are logged only once through their own path
+                logCutOffByRouteStop(exchange, exchange.getException(), logger.getLevel());
             }
             AsyncCallback cb = callback;
             taskFactory.release(this);

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();
+        }
+    }
+}

docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc

@@ -328,6 +328,17 @@ 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-pqc - an unknown hybridKdfAlgorithm is rejected
 
 The hybrid KEM operations now fail with an `IllegalArgumentException` when `hybridKdfAlgorithm` is not one of