CAMEL-17065: Fix cluster view release on all routes stop. Add autoStartup check before starting route.

core/camel-cluster/src/main/java/org/apache/camel/impl/cluster/ClusteredRoutePolicy.java

@@ -53,6 +53,7 @@ public final class ClusteredRoutePolicy extends RoutePolicySupport implements Ca
     private static final Logger LOG = LoggerFactory.getLogger(ClusteredRoutePolicy.class);
 
     private final AtomicBoolean leader;
+    private final Set<Route> autoStartupRoutes;
     private final Set<Route> startedRoutes;
     private final Set<Route> stoppedRoutes;
     private final ReferenceCount refCount;
@@ -84,6 +85,7 @@ private ClusteredRoutePolicy(CamelClusterService clusterService, CamelClusterSer
 
         this.stoppedRoutes = new HashSet<>();
         this.startedRoutes = new HashSet<>();
+        this.autoStartupRoutes = new HashSet<>();
         this.leader = new AtomicBoolean();
         this.contextStarted = new AtomicBoolean();
         this.initialDelay = Duration.ofMillis(0);
@@ -178,6 +180,10 @@ public void onInit(Route route) {
 
         this.refCount.retain();
 
+        if (route.isAutoStartup()) {
+            autoStartupRoutes.add(route);
+        }
+
         if (camelContext.isStarted() && isLeader()) {
             // when camel context is already started, and we add new routes
             // then let the route controller start the route as usual (no need to mark as auto startup false)
@@ -214,7 +220,12 @@ public void doStart() throws Exception {
     }
 
     @Override
-    public void doShutdown() throws Exception {
+    public void onRemove(Route route) {
+        autoStartupRoutes.remove(route);
+    }
+
+    @Override
+    public void onStop(Route route) {
         this.refCount.release();
     }
 
@@ -266,7 +277,9 @@ private void doStartManagedRoutes() {
         try {
             for (Route route : stoppedRoutes) {
                 ServiceStatus status = getStatus(route);
-                if (status != null && status.isStartable()) {
+                boolean autostart = autoStartupRoutes.contains(route);
+
+                if (status != null && status.isStartable() && autostart) {
                     LOG.debug("Starting route '{}'", route.getId());
                     camelContext.getRouteController().startRoute(route.getId());
 

core/camel-core/src/test/java/org/apache/camel/cluster/ClusteredRoutePolicyTest.java

@@ -31,6 +31,7 @@
 import org.junit.jupiter.api.Test;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 
 public class ClusteredRoutePolicyTest extends ContextTestSupport {
 
@@ -66,6 +67,23 @@ public void testClusteredRoutePolicy() throws Exception {
         assertEquals(ServiceStatus.Started, context.getRouteController().getRouteStatus("foo"));
     }
 
+    @Test
+    public void testClusteredRoutePolicyStopAllRoutes() throws Exception {
+        cs.getView().setLeader(true);
+
+        context.getRouteController().stopRoute("foo");
+        context.getRouteController().stopRoute("baz");
+
+        assertFalse(cs.getView().isRunning());
+    }
+
+    @Test
+    public void testClusteredRoutePolicyDontStartAutoStartFalseRoutes() throws Exception {
+        cs.getView().setLeader(true);
+
+        assertEquals(ServiceStatus.Stopped, context.getRouteController().getRouteStatus("baz"));
+    }
+
     @Test
     public void testClusteredRoutePolicyAddRoute() throws Exception {
         context.addRoutes(new RouteBuilder() {
@@ -129,6 +147,8 @@ protected RouteBuilder createRouteBuilder() throws Exception {
             public void configure() throws Exception {
                 from("seda:foo").routeId("foo").routePolicy(policy)
                         .to("mock:foo");
+                from("seda:baz").autoStartup(false).routeId("baz").routePolicy(policy)
+                        .to("mock:baz");
             }
         };
     }
@@ -139,6 +159,7 @@ public void configure() throws Exception {
 
     private static class TestClusterView extends AbstractCamelClusterView {
         private boolean leader;
+        private boolean running;
 
         public TestClusterView(CamelClusterService cluster, String namespace) {
             super(cluster, namespace);
@@ -176,10 +197,12 @@ public List<CamelClusterMember> getMembers() {
 
         @Override
         protected void doStart() throws Exception {
+            running = true;
         }
 
         @Override
         protected void doStop() throws Exception {
+            running = false;
         }
 
         public boolean isLeader() {
@@ -193,6 +216,10 @@ public void setLeader(boolean leader) {
                 fireLeadershipChangedEvent(getLeader());
             }
         }
+
+        public boolean isRunning() {
+            return running;
+        }
     }
 
     private static class TestClusterService extends AbstractCamelClusterService<TestClusterView> {

core/camel-core/src/test/java/org/apache/camel/cluster/ClusteredRoutePolicyUnmanagedClusterServiceTest.java

@@ -0,0 +1,161 @@
+/*
+ * 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.cluster;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.cluster.ClusteredRoutePolicy;
+import org.apache.camel.support.cluster.AbstractCamelClusterService;
+import org.apache.camel.support.cluster.AbstractCamelClusterView;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+public class ClusteredRoutePolicyUnmanagedClusterServiceTest extends ContextTestSupport {
+
+    private ClusteredRoutePolicy policy;
+    private TestClusterService cs;
+
+    @Override
+    protected CamelContext createCamelContext() throws Exception {
+        CamelContext context = super.createCamelContext();
+
+        cs = new TestClusterService("my-cluster-service");
+        cs.start();
+
+        policy = ClusteredRoutePolicy.forNamespace(cs, "my-ns");
+
+        return context;
+    }
+
+    @Test
+    public void testClusteredRoutePolicyReleaseViewOnCamelContextStop() {
+        cs.getView().setLeader(true);
+
+        context.stop();
+
+        assertFalse(cs.getView().isRunning());
+    }
+
+    @Override
+    protected RouteBuilder createRouteBuilder() throws Exception {
+        return new RouteBuilder() {
+            @Override
+            public void configure() throws Exception {
+                from("seda:foo").routeId("foo").routePolicy(policy)
+                        .to("mock:foo");
+                from("seda:bar").routeId("bar").routePolicy(policy)
+                        .to("mock:bar");
+            }
+        };
+    }
+
+    // *********************************
+    // Helpers
+    // *********************************
+
+    private static class TestClusterView extends AbstractCamelClusterView {
+        private boolean leader;
+        private boolean running;
+
+        public TestClusterView(CamelClusterService cluster, String namespace) {
+            super(cluster, namespace);
+        }
+
+        @Override
+        public Optional<CamelClusterMember> getLeader() {
+            return leader ? Optional.of(getLocalMember()) : Optional.empty();
+        }
+
+        @Override
+        public CamelClusterMember getLocalMember() {
+            return new CamelClusterMember() {
+                @Override
+                public boolean isLeader() {
+                    return leader;
+                }
+
+                @Override
+                public boolean isLocal() {
+                    return true;
+                }
+
+                @Override
+                public String getId() {
+                    return getClusterService().getId();
+                }
+            };
+        }
+
+        @Override
+        public List<CamelClusterMember> getMembers() {
+            return Collections.emptyList();
+        }
+
+        @Override
+        protected void doStart() throws Exception {
+            running = true;
+        }
+
+        @Override
+        protected void doStop() throws Exception {
+            running = false;
+        }
+
+        public boolean isLeader() {
+            return leader;
+        }
+
+        public void setLeader(boolean leader) {
+            this.leader = leader;
+
+            if (isRunAllowed()) {
+                fireLeadershipChangedEvent(getLeader());
+            }
+        }
+
+        public boolean isRunning() {
+            return running;
+        }
+    }
+
+    private static class TestClusterService extends AbstractCamelClusterService<TestClusterView> {
+
+        private TestClusterView view;
+
+        public TestClusterService(String id) {
+            super(id);
+        }
+
+        @Override
+        protected TestClusterView createView(String namespace) throws Exception {
+            if (view == null) {
+                view = new TestClusterView(this, namespace);
+            }
+            return view;
+        }
+
+        public TestClusterView getView() {
+            return view;
+        }
+    }
+}