From 5afc545aa2f1cfce750d9375766018e73eaa74de Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Thu, 24 Sep 2026 21:48:04 +0530 Subject: [PATCH] CAMEL-25010: camel-core - Supervising route controller: do not start a route stopped during its restart attempt The restart attempt of the supervising route controller only checked CamelContext.isRunAllowed() and then called doStartRoute, which takes the controller lock and starts the route. stopRoute takes the same lock and cancels the route's restart task, but cancelling only prevents future attempts. An attempt that was already running, for example waiting for the controller lock while another route was being stopped, started the route after stopRoute had returned. The route was then running without supervision, and nothing reported it. The attempt now checks, while holding the controller lock, that its task is still the route's restart task and still active, and that Camel is not stopping. Otherwise it gives up without starting the route. Co-Authored-By: Claude Opus 5.5 --- .../DefaultSupervisingRouteController.java | 15 +- ...outeControllerStopWhileRestartingTest.java | 128 ++++++++++++++++++ 2 files changed, 142 insertions(+), 1 deletion(-) create mode 100644 core/camel-core/src/test/java/org/apache/camel/impl/engine/DefaultSupervisingRouteControllerStopWhileRestartingTest.java diff --git a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultSupervisingRouteController.java b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultSupervisingRouteController.java index 0e21c2a190791..a04b5200540cb 100644 --- a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultSupervisingRouteController.java +++ b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultSupervisingRouteController.java @@ -688,7 +688,20 @@ void start(RouteHolder route) { try { logger.info("Restarting route: {} attempt: {}", r.getId(), attempt); EventHelper.notifyRouteRestarting(getCamelContext(), r.get(), attempt); - doStartRoute(r, false, rx -> DefaultSupervisingRouteController.super.startRoute(rx.getId())); + lock.lock(); + try { + // the route may have been stopped or started manually (which cancels this task), + // or Camel may be stopping, while this attempt waited for the lock + if (routes.get(r) != context || context.getStatus() != BackOffTimer.Task.Status.Active + || !getCamelContext().isRunAllowed()) { + logger.info("Restarting route: {} attempt: {} is cancelled", r.getId(), attempt); + return false; + } + doStartRoute(r, false, + rx -> DefaultSupervisingRouteController.super.startRoute(rx.getId())); + } finally { + lock.unlock(); + } logger.info("Route: {} started after {} attempts", r.getId(), attempt); return false; } catch (Exception e) { diff --git a/core/camel-core/src/test/java/org/apache/camel/impl/engine/DefaultSupervisingRouteControllerStopWhileRestartingTest.java b/core/camel-core/src/test/java/org/apache/camel/impl/engine/DefaultSupervisingRouteControllerStopWhileRestartingTest.java new file mode 100644 index 0000000000000..d04d6c2d865e1 --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/impl/engine/DefaultSupervisingRouteControllerStopWhileRestartingTest.java @@ -0,0 +1,128 @@ +/* + * 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.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.camel.Consumer; +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Endpoint; +import org.apache.camel.Processor; +import org.apache.camel.Producer; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.spi.CamelEvent; +import org.apache.camel.spi.CamelEvent.RouteRestartingEvent; +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.SimpleEventNotifierSupport; +import org.apache.camel.util.backoff.BackOffTimer; +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.assertTrue; + +/** + * A route that is stopped manually while its restart attempt is running must not be started by that attempt. + */ +class DefaultSupervisingRouteControllerStopWhileRestartingTest extends ContextTestSupport { + + private final AtomicBoolean failing = new AtomicBoolean(true); + + @Override + public boolean isUseRouteBuilder() { + return false; + } + + @Test + void testStopRouteWhileRestartAttemptIsRunning() throws Exception { + CountDownLatch attemptStarted = new CountDownLatch(1); + CountDownLatch routeStopped = new CountDownLatch(1); + AtomicReference restartTask = new AtomicReference<>(); + + context.addComponent("flaky", new FlakyComponent()); + context.addRoutes(new RouteBuilder() { + @Override + public void configure() { + from("flaky:start").routeId("flaky").to("mock:result"); + } + }); + + SupervisingRouteController src = context.getRouteController().supervising(); + src.setBackOffDelay(10); + src.setBackOffMaxAttempts(3); + src.setInitialDelay(10); + + context.getManagementStrategy().addEventNotifier(new SimpleEventNotifierSupport() { + @Override + public void notify(CamelEvent event) throws Exception { + if (event instanceof RouteRestartingEvent rre && "flaky".equals(rre.getRoute().getRouteId()) + && restartTask.compareAndSet(null, src.getRestartingRouteState("flaky"))) { + // the restart attempt is about to start the route: let the route be stopped first + attemptStarted.countDown(); + assertTrue(routeStopped.await(10, TimeUnit.SECONDS)); + } + } + }); + + context.start(); + assertTrue(attemptStarted.await(10, TimeUnit.SECONDS)); + + // the cause is fixed, but the route is stopped manually while the restart attempt is running + failing.set(false); + context.getRouteController().stopRoute("flaky"); + assertEquals("Stopped", context.getRouteController().getRouteStatus("flaky").toString()); + routeStopped.countDown(); + + // the restart attempt finishes, and the route stays stopped + await().atMost(10, TimeUnit.SECONDS) + .until(() -> restartTask.get().getStatus() == BackOffTimer.Task.Status.Completed); + assertEquals("Stopped", context.getRouteController().getRouteStatus("flaky").toString()); + } + + private final class FlakyComponent extends DefaultComponent { + + @Override + protected Endpoint createEndpoint(String uri, String remaining, Map parameters) { + return new DefaultEndpoint(uri, this) { + @Override + public Producer createProducer() { + throw new UnsupportedOperationException(); + } + + @Override + public Consumer createConsumer(Processor processor) { + return new DefaultConsumer(this, processor) { + @Override + protected void doStart() throws Exception { + if (failing.get()) { + throw new IllegalStateException("Cannot connect"); + } + super.doStart(); + } + }; + } + }; + } + } +}