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

@@ -55,6 +55,7 @@
 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 void run() {
                             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 @@ boolean doRun() throws Exception {
                     // 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 @@ protected void doDone(
                     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 @@ 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(cacheKey, answer);
@@ -1323,6 +1356,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());
+        }
     }
 
     /**
@@ -1344,12 +1382,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

@@ -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 @@ private void runNotAllowed() {
             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);

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