gnodet-bot commented on code in PR #29583: URL: https://github.com/apache/camel/pull/29583#discussion_r4537710352
########## components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/policy/MasterRoutePolicyRouteLifecycleTest.java: ########## @@ -0,0 +1,441 @@ +/* + * 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.component.zookeepermaster.policy; + +import java.io.File; +import java.net.InetSocketAddress; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.camel.CamelContext; +import org.apache.camel.Consumer; +import org.apache.camel.Processor; +import org.apache.camel.Producer; +import org.apache.camel.ProducerTemplate; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.component.zookeepermaster.ZookeeperGroupSupport; +import org.apache.camel.impl.DefaultCamelContext; +import org.apache.camel.support.DefaultConsumer; +import org.apache.camel.support.DefaultEndpoint; +import org.apache.camel.support.service.ServiceHelper; +import org.apache.camel.test.AvailablePortFinder; +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.framework.listen.ListenerManager; +import org.apache.curator.retry.RetryOneTime; +import org.apache.zookeeper.server.NIOServerCnxnFactory; +import org.apache.zookeeper.server.ZooKeeperServer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +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.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * {@link MasterRoutePolicy} when its route is removed, added again, started or resumed, with two CamelContexts (two + * nodes) and a ZooKeeper server in this JVM. + */ +class MasterRoutePolicyRouteLifecycleTest { + + private static final String ROUTE = "master"; + + @TempDir + Path dataDir; + + private final int port = AvailablePortFinder.getNextAvailable(); + private final List<CamelContext> contexts = new ArrayList<>(); + private final List<CuratorFramework> clients = new ArrayList<>(); + + private ZooKeeperServer server; + private NIOServerCnxnFactory factory; + + @BeforeEach + void startServer() throws Exception { + File dir = dataDir.toFile(); + server = new ZooKeeperServer(dir, dir, 2000); + factory = new NIOServerCnxnFactory(); + factory.configure(new InetSocketAddress("127.0.0.1", port), 60); + factory.startup(server); + } + + @AfterEach + void stopServer() { + // the CamelContexts first, so their policies leave the group while the server runs + contexts.forEach(CamelContext::stop); + clients.forEach(CuratorFramework::close); + if (factory != null) { + factory.shutdown(); + } + if (server != null) { + server.shutdown(); + } + } + + @Test + void removedRouteGivesUpTheMastership() throws Exception { + CamelContext a = startNode(newPolicy()); + awaitConsuming(a); + CamelContext b = startNode(newPolicy()); + awaitSlave(b); + + // the route is removed on the master: the other node must take over, otherwise the route runs nowhere + a.getRouteController().stopRoute(ROUTE); + assertTrue(a.removeRoute(ROUTE)); + + awaitConsuming(b); + assertConsumes(b); + } + + @Test + void removedRouteOfASlaveLeavesTheGroup() throws Exception { + CamelContext a = startNode(newPolicy()); + awaitConsuming(a); + CamelContext b = startNode(newPolicy()); + awaitSlave(b); + await().atMost(30, TimeUnit.SECONDS).until(() -> groupMembers() == 2); + + // the route is removed on the slave: its policy must not be elected later, as it would start the consumer of + // the removed route + b.getRouteController().stopRoute(ROUTE); + assertTrue(b.removeRoute(ROUTE)); + await().atMost(30, TimeUnit.SECONDS).untilAsserted( + () -> assertEquals(1, groupMembers(), "The policy of a removed route must leave the group")); + } + + @Test + void routeAddedToARunningContextJoinsTheGroupWithItsConsumer() throws Exception { + CamelContext a = startNode(newPolicy()); + awaitConsuming(a); + + // a route with a new policy is added to the running CamelContext: the policy must join the group only once the + // consumer of the route is created, as a master that is elected before cannot start it + MasterRoutePolicy policy = newPolicy(); + policy.setGroupName("added"); + AtomicBoolean joinedBeforeConsumer = new AtomicBoolean(); + a.addEndpoint("watched:in", new WatchedEndpoint(() -> joinedBeforeConsumer.set(policy.thisNode() != null))); + a.addRoutes(new RouteBuilder() { + @Override + public void configure() { + from("watched:in").routeId("added").autoStartup(false).routePolicy(policy).to("mock:added"); + } + }); + + assertFalse(joinedBeforeConsumer.get(), "The policy joined the group before the consumer was created"); + await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> assertTrue( + policy.isMaster() && ServiceHelper.isStarted(a.getRoute("added").getConsumer()), + "The node must become the master and start its consumer")); + } + + @Test + void restartedContextRunsTheRouteAgain() throws Exception { + // a control (passes on main too): the policy leaves the group when the CamelContext stops, and joins it again + CamelContext a = startNode(newPolicy()); + awaitConsuming(a); + + a.stop(); + a.start(); + + awaitConsuming(a); + assertConsumes(a); + } + + @Test + void routeAddedAgainWithANewPolicyRunsAgain() throws Exception { + CamelContext a = startNode(newPolicy()); + awaitConsuming(a); + + // a route reload: the route is removed and added again, with a new policy instance + a.getRouteController().stopRoute(ROUTE); + assertTrue(a.removeRoute(ROUTE)); + a.addRoutes(route(newPolicy())); + + awaitConsuming(a); + assertConsumes(a); + } + + @Test + void routeAddedAgainWithTheSamePolicyRunsAgain() throws Exception { + MasterRoutePolicy policy = newPolicy(); + CamelContext a = startNode(policy); + awaitConsuming(a); + + a.getRouteController().stopRoute(ROUTE); + assertTrue(a.removeRoute(ROUTE)); + a.addRoutes(route(policy)); + + awaitConsuming(a); + assertConsumes(a); + } + + @Test + void routeOfASlaveIsKeptStoppedWhenStarted() throws Exception { + CamelContext a = startNode(newPolicy()); + awaitConsuming(a); + CamelContext b = startNode(newPolicy()); + awaitSlave(b); + + // the route of the slave is started (from JMX or the controlbus): its consumer must stay stopped + b.getRouteController().startRoute(ROUTE); + assertFalse(isConsumerStarted(b), "The slave must not consume"); + + // the master goes away: the slave takes over and consumes + a.stop(); + awaitConsuming(b); + assertConsumes(b); + } + + @Test + void routeOfASlaveIsKeptStoppedWhenResumed() throws Exception { + CamelContext a = startNode(newPolicy()); + awaitConsuming(a); + CamelContext b = startNode(newPolicy()); + awaitSlave(b); + + // the route of the slave is resumed, which starts the consumer of a route that is not started, and then it is + // started, suspended and resumed: the consumer must stay stopped + b.getRouteController().resumeRoute(ROUTE); + assertFalse(isConsumerStarted(b), "The slave must not consume after a resume"); + b.getRouteController().startRoute(ROUTE); + b.getRouteController().suspendRoute(ROUTE); + b.getRouteController().resumeRoute(ROUTE); + assertFalse(isConsumerStarted(b), "The slave must not consume after a suspend and resume"); + + a.stop(); + awaitConsuming(b); + assertConsumes(b); + } + + @Test + void routeWithAutoStartupRunsOnlyOnTheMaster() throws Exception { + CamelContext a = startNode(newPolicy(), true); + awaitConsuming(a); + CamelContext b = startNode(newPolicy(), true); + awaitSlave(b); + + a.stop(); + awaitConsuming(b); + assertConsumes(b); + } + + @Test + void failedJoinLeavesNoHalfJoinedState() throws Exception { + CuratorFramework client = newClient(); + MasterRoutePolicy policy = newPolicy(); + policy.setCurator(client); + // the first join fails after the group listener started (the node state cannot be created) + AtomicInteger failures = new AtomicInteger(1); + policy.setContainerIdFactory(() -> { + if (failures.getAndDecrement() > 0) { + throw new IllegalStateException("Forced failure to create the container id"); + } + return "node-a"; + }); + CamelContext a = new DefaultCamelContext(); + contexts.add(a); + a.start(); + + // the route is added to the running CamelContext, and the join fails: the policy must not keep the listener + // (the JMX attributes would read its group) nor leave it on the shared CuratorFramework + a.addRoutes(route(policy)); + assertTrue(failures.get() <= 0, "The policy must have tried to join"); + assertFalse(policy.isConnected(), "The policy must not keep the group of a failed join"); + assertEquals(0, groupListenersOn(client), "The listener of a failed join must be removed from the curator"); + assertFalse(isConsumerStarted(a)); + + // the route is added again: the policy joins the group and the node runs the route + a.getRouteController().stopRoute(ROUTE); + assertTrue(a.removeRoute(ROUTE)); + a.addRoutes(route(policy)); + awaitConsuming(a); + assertConsumes(a); + assertEquals(1, groupListenersOn(client)); + } + + @Test + void failedJoinAtStartLeavesNoListenerOnTheCurator() throws Exception { + CuratorFramework client = newClient(); + MasterRoutePolicy policy = newPolicy(); + policy.setCurator(client); + policy.setContainerIdFactory(() -> { + throw new IllegalStateException("Forced failure to create the container id"); + }); + CamelContext a = new DefaultCamelContext(); + contexts.add(a); + a.addRoutes(route(policy)); + + // the policy joins when the CamelContext starts, which fails with the join. A control: the CamelContext that + // fails to start also removes the route, whose onRemove cleaned up before the join undid itself + assertThrows(Exception.class, a::start); + assertFalse(policy.isConnected()); + assertEquals(0, groupListenersOn(client), "The listener of a failed join must be removed from the curator"); + } + + @Test + void policyStoppedOnItsOwnLeavesTheGroupAndJoinsAgainWhenStarted() throws Exception { + CuratorFramework client = newClient(); + MasterRoutePolicy policy = newPolicy(); + policy.setCurator(client); + CamelContext a = startNode(policy); + awaitConsuming(a); + assertEquals(1, groupListenersOn(client)); + + // the policy is stopped on its own (from JMX): it leaves the group like a removed route, and the shared + // CuratorFramework must not keep its listener + ServiceHelper.stopService(policy); + assertEquals(0, groupListenersOn(client), "The stopped policy must remove its listener from the curator"); + assertFalse(policy.isConnected()); + assertFalse(policy.isMaster()); + assertFalse(isConsumerStarted(a), "A node that left the group must not consume"); + + // started again, it joins the group again and runs the route + ServiceHelper.startService(policy); + awaitConsuming(a); + assertConsumes(a); + assertEquals(1, groupListenersOn(client)); + } + + private MasterRoutePolicy newPolicy() { + MasterRoutePolicy policy = new MasterRoutePolicy(); + policy.setZooKeeperUrl("127.0.0.1:" + port); + policy.setZkRoot("/camel/lifecycle"); + policy.setGroupName("lifecycle"); + return policy; + } + + private static RouteBuilder route(MasterRoutePolicy policy) { + return route(policy, false); + } + + private static RouteBuilder route(MasterRoutePolicy policy, boolean autoStartup) { + return new RouteBuilder() { + @Override + public void configure() { + from("seda:in").routeId(ROUTE).autoStartup(autoStartup).routePolicy(policy).to("mock:out"); + } + }; + } + + private CamelContext startNode(MasterRoutePolicy policy) throws Exception { + return startNode(policy, false); + } + + private CamelContext startNode(MasterRoutePolicy policy, boolean autoStartup) throws Exception { + CamelContext context = new DefaultCamelContext(); + contexts.add(context); + context.addRoutes(route(policy, autoStartup)); + context.start(); + return context; + } + + // a CuratorFramework of the user, shared by the policies, which outlives them + private CuratorFramework newClient() throws Exception { + CuratorFramework client = CuratorFrameworkFactory.newClient("127.0.0.1:" + port, new RetryOneTime(100)); + clients.add(client); + client.start(); + assertTrue(client.blockUntilConnected(20, TimeUnit.SECONDS)); + return client; + } + + // the group listeners of the policies that are registered on the client + private static int groupListenersOn(CuratorFramework client) { + AtomicInteger count = new AtomicInteger(); + ((ListenerManager<?, ?>) client.getConnectionStateListenable()).forEach(listener -> { + if (listener instanceof ZookeeperGroupSupport) { + count.incrementAndGet(); + } + }); + return count.get(); + } + + // the members of the group in ZooKeeper + private int groupMembers() throws Exception { + try (CuratorFramework client + = CuratorFrameworkFactory.newClient("127.0.0.1:" + port, new RetryOneTime(100))) { + client.start(); + client.blockUntilConnected(20, TimeUnit.SECONDS); + return client.getChildren().forPath("/camel/lifecycle/lifecycle").size(); + } Review Comment: ⚠️ **New ZK client per polling call** — `groupMembers()` creates, connects, and closes a `CuratorFramework` on every invocation. This method is called inside `await().until()` polling loops (lines 119 and 126 of this file), which poll it repeatedly until the ZK group membership changes. Each call adds a full TCP connect + handshake to a local ZK server. Under CI load, repeated rapid connects can cause connection-refused or session-creation throttling on the embedded `NIOServerCnxnFactory`, which caps concurrent connections at 60 (the value passed to `factory.configure(..., 60)`). Consider reusing the test's shared client infrastructure: ```suggestion // the members of the group in ZooKeeper — uses an ad-hoc client; called in polling loops, // so each invocation connects to the embedded ZK server. At most a few polls in a 30s timeout // but be aware that NIOServerCnxnFactory limits concurrent connections to 60. private int groupMembers() throws Exception { try (CuratorFramework client = CuratorFrameworkFactory.newClient("127.0.0.1:" + port, new RetryOneTime(100))) { client.start(); client.blockUntilConnected(20, TimeUnit.SECONDS); return client.getChildren().forPath("/camel/lifecycle/lifecycle").size(); } } ``` ########## components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/policy/MasterRoutePolicyRouteLifecycleTest.java: ########## @@ -0,0 +1,441 @@ +/* + * 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.component.zookeepermaster.policy; + +import java.io.File; +import java.net.InetSocketAddress; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.camel.CamelContext; +import org.apache.camel.Consumer; +import org.apache.camel.Processor; +import org.apache.camel.Producer; +import org.apache.camel.ProducerTemplate; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.component.zookeepermaster.ZookeeperGroupSupport; +import org.apache.camel.impl.DefaultCamelContext; +import org.apache.camel.support.DefaultConsumer; +import org.apache.camel.support.DefaultEndpoint; +import org.apache.camel.support.service.ServiceHelper; +import org.apache.camel.test.AvailablePortFinder; +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.framework.listen.ListenerManager; +import org.apache.curator.retry.RetryOneTime; +import org.apache.zookeeper.server.NIOServerCnxnFactory; +import org.apache.zookeeper.server.ZooKeeperServer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +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.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * {@link MasterRoutePolicy} when its route is removed, added again, started or resumed, with two CamelContexts (two + * nodes) and a ZooKeeper server in this JVM. + */ +class MasterRoutePolicyRouteLifecycleTest { + + private static final String ROUTE = "master"; + + @TempDir + Path dataDir; + + private final int port = AvailablePortFinder.getNextAvailable(); + private final List<CamelContext> contexts = new ArrayList<>(); + private final List<CuratorFramework> clients = new ArrayList<>(); + + private ZooKeeperServer server; + private NIOServerCnxnFactory factory; + + @BeforeEach + void startServer() throws Exception { + File dir = dataDir.toFile(); + server = new ZooKeeperServer(dir, dir, 2000); + factory = new NIOServerCnxnFactory(); + factory.configure(new InetSocketAddress("127.0.0.1", port), 60); + factory.startup(server); + } + + @AfterEach + void stopServer() { + // the CamelContexts first, so their policies leave the group while the server runs + contexts.forEach(CamelContext::stop); + clients.forEach(CuratorFramework::close); + if (factory != null) { + factory.shutdown(); + } + if (server != null) { + server.shutdown(); + } + } + + @Test + void removedRouteGivesUpTheMastership() throws Exception { + CamelContext a = startNode(newPolicy()); + awaitConsuming(a); + CamelContext b = startNode(newPolicy()); + awaitSlave(b); + + // the route is removed on the master: the other node must take over, otherwise the route runs nowhere + a.getRouteController().stopRoute(ROUTE); + assertTrue(a.removeRoute(ROUTE)); + + awaitConsuming(b); + assertConsumes(b); + } + + @Test + void removedRouteOfASlaveLeavesTheGroup() throws Exception { + CamelContext a = startNode(newPolicy()); + awaitConsuming(a); + CamelContext b = startNode(newPolicy()); + awaitSlave(b); + await().atMost(30, TimeUnit.SECONDS).until(() -> groupMembers() == 2); + + // the route is removed on the slave: its policy must not be elected later, as it would start the consumer of + // the removed route + b.getRouteController().stopRoute(ROUTE); + assertTrue(b.removeRoute(ROUTE)); + await().atMost(30, TimeUnit.SECONDS).untilAsserted( + () -> assertEquals(1, groupMembers(), "The policy of a removed route must leave the group")); + } + + @Test + void routeAddedToARunningContextJoinsTheGroupWithItsConsumer() throws Exception { + CamelContext a = startNode(newPolicy()); + awaitConsuming(a); + + // a route with a new policy is added to the running CamelContext: the policy must join the group only once the + // consumer of the route is created, as a master that is elected before cannot start it + MasterRoutePolicy policy = newPolicy(); + policy.setGroupName("added"); + AtomicBoolean joinedBeforeConsumer = new AtomicBoolean(); + a.addEndpoint("watched:in", new WatchedEndpoint(() -> joinedBeforeConsumer.set(policy.thisNode() != null))); + a.addRoutes(new RouteBuilder() { + @Override + public void configure() { + from("watched:in").routeId("added").autoStartup(false).routePolicy(policy).to("mock:added"); + } + }); + + assertFalse(joinedBeforeConsumer.get(), "The policy joined the group before the consumer was created"); + await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> assertTrue( + policy.isMaster() && ServiceHelper.isStarted(a.getRoute("added").getConsumer()), + "The node must become the master and start its consumer")); + } + + @Test + void restartedContextRunsTheRouteAgain() throws Exception { + // a control (passes on main too): the policy leaves the group when the CamelContext stops, and joins it again + CamelContext a = startNode(newPolicy()); + awaitConsuming(a); + + a.stop(); + a.start(); + + awaitConsuming(a); + assertConsumes(a); + } + + @Test + void routeAddedAgainWithANewPolicyRunsAgain() throws Exception { + CamelContext a = startNode(newPolicy()); + awaitConsuming(a); + + // a route reload: the route is removed and added again, with a new policy instance + a.getRouteController().stopRoute(ROUTE); + assertTrue(a.removeRoute(ROUTE)); + a.addRoutes(route(newPolicy())); + + awaitConsuming(a); + assertConsumes(a); + } + + @Test + void routeAddedAgainWithTheSamePolicyRunsAgain() throws Exception { + MasterRoutePolicy policy = newPolicy(); + CamelContext a = startNode(policy); + awaitConsuming(a); + + a.getRouteController().stopRoute(ROUTE); + assertTrue(a.removeRoute(ROUTE)); + a.addRoutes(route(policy)); + + awaitConsuming(a); + assertConsumes(a); + } + + @Test + void routeOfASlaveIsKeptStoppedWhenStarted() throws Exception { + CamelContext a = startNode(newPolicy()); + awaitConsuming(a); + CamelContext b = startNode(newPolicy()); + awaitSlave(b); + + // the route of the slave is started (from JMX or the controlbus): its consumer must stay stopped + b.getRouteController().startRoute(ROUTE); + assertFalse(isConsumerStarted(b), "The slave must not consume"); + + // the master goes away: the slave takes over and consumes + a.stop(); + awaitConsuming(b); + assertConsumes(b); + } + + @Test + void routeOfASlaveIsKeptStoppedWhenResumed() throws Exception { + CamelContext a = startNode(newPolicy()); + awaitConsuming(a); + CamelContext b = startNode(newPolicy()); + awaitSlave(b); + + // the route of the slave is resumed, which starts the consumer of a route that is not started, and then it is + // started, suspended and resumed: the consumer must stay stopped + b.getRouteController().resumeRoute(ROUTE); + assertFalse(isConsumerStarted(b), "The slave must not consume after a resume"); + b.getRouteController().startRoute(ROUTE); + b.getRouteController().suspendRoute(ROUTE); + b.getRouteController().resumeRoute(ROUTE); + assertFalse(isConsumerStarted(b), "The slave must not consume after a suspend and resume"); + + a.stop(); + awaitConsuming(b); + assertConsumes(b); + } + + @Test + void routeWithAutoStartupRunsOnlyOnTheMaster() throws Exception { + CamelContext a = startNode(newPolicy(), true); + awaitConsuming(a); + CamelContext b = startNode(newPolicy(), true); + awaitSlave(b); + + a.stop(); + awaitConsuming(b); + assertConsumes(b); + } + + @Test + void failedJoinLeavesNoHalfJoinedState() throws Exception { + CuratorFramework client = newClient(); + MasterRoutePolicy policy = newPolicy(); + policy.setCurator(client); + // the first join fails after the group listener started (the node state cannot be created) + AtomicInteger failures = new AtomicInteger(1); + policy.setContainerIdFactory(() -> { + if (failures.getAndDecrement() > 0) { + throw new IllegalStateException("Forced failure to create the container id"); + } + return "node-a"; + }); + CamelContext a = new DefaultCamelContext(); + contexts.add(a); + a.start(); + + // the route is added to the running CamelContext, and the join fails: the policy must not keep the listener + // (the JMX attributes would read its group) nor leave it on the shared CuratorFramework + a.addRoutes(route(policy)); + assertTrue(failures.get() <= 0, "The policy must have tried to join"); + assertFalse(policy.isConnected(), "The policy must not keep the group of a failed join"); + assertEquals(0, groupListenersOn(client), "The listener of a failed join must be removed from the curator"); + assertFalse(isConsumerStarted(a)); + + // the route is added again: the policy joins the group and the node runs the route + a.getRouteController().stopRoute(ROUTE); + assertTrue(a.removeRoute(ROUTE)); + a.addRoutes(route(policy)); + awaitConsuming(a); + assertConsumes(a); + assertEquals(1, groupListenersOn(client)); + } + + @Test + void failedJoinAtStartLeavesNoListenerOnTheCurator() throws Exception { + CuratorFramework client = newClient(); + MasterRoutePolicy policy = newPolicy(); + policy.setCurator(client); + policy.setContainerIdFactory(() -> { + throw new IllegalStateException("Forced failure to create the container id"); + }); + CamelContext a = new DefaultCamelContext(); + contexts.add(a); + a.addRoutes(route(policy)); + + // the policy joins when the CamelContext starts, which fails with the join. A control: the CamelContext that + // fails to start also removes the route, whose onRemove cleaned up before the join undid itself + assertThrows(Exception.class, a::start); + assertFalse(policy.isConnected()); + assertEquals(0, groupListenersOn(client), "The listener of a failed join must be removed from the curator"); + } + + @Test + void policyStoppedOnItsOwnLeavesTheGroupAndJoinsAgainWhenStarted() throws Exception { + CuratorFramework client = newClient(); + MasterRoutePolicy policy = newPolicy(); + policy.setCurator(client); + CamelContext a = startNode(policy); + awaitConsuming(a); + assertEquals(1, groupListenersOn(client)); + + // the policy is stopped on its own (from JMX): it leaves the group like a removed route, and the shared + // CuratorFramework must not keep its listener + ServiceHelper.stopService(policy); + assertEquals(0, groupListenersOn(client), "The stopped policy must remove its listener from the curator"); + assertFalse(policy.isConnected()); + assertFalse(policy.isMaster()); + assertFalse(isConsumerStarted(a), "A node that left the group must not consume"); + + // started again, it joins the group again and runs the route + ServiceHelper.startService(policy); + awaitConsuming(a); + assertConsumes(a); + assertEquals(1, groupListenersOn(client)); + } + + private MasterRoutePolicy newPolicy() { + MasterRoutePolicy policy = new MasterRoutePolicy(); + policy.setZooKeeperUrl("127.0.0.1:" + port); + policy.setZkRoot("/camel/lifecycle"); + policy.setGroupName("lifecycle"); + return policy; + } + + private static RouteBuilder route(MasterRoutePolicy policy) { + return route(policy, false); + } + + private static RouteBuilder route(MasterRoutePolicy policy, boolean autoStartup) { + return new RouteBuilder() { + @Override + public void configure() { + from("seda:in").routeId(ROUTE).autoStartup(autoStartup).routePolicy(policy).to("mock:out"); + } + }; + } + + private CamelContext startNode(MasterRoutePolicy policy) throws Exception { + return startNode(policy, false); + } + + private CamelContext startNode(MasterRoutePolicy policy, boolean autoStartup) throws Exception { + CamelContext context = new DefaultCamelContext(); + contexts.add(context); + context.addRoutes(route(policy, autoStartup)); + context.start(); + return context; + } + + // a CuratorFramework of the user, shared by the policies, which outlives them + private CuratorFramework newClient() throws Exception { + CuratorFramework client = CuratorFrameworkFactory.newClient("127.0.0.1:" + port, new RetryOneTime(100)); + clients.add(client); + client.start(); + assertTrue(client.blockUntilConnected(20, TimeUnit.SECONDS)); + return client; + } + + // the group listeners of the policies that are registered on the client + private static int groupListenersOn(CuratorFramework client) { + AtomicInteger count = new AtomicInteger(); + ((ListenerManager<?, ?>) client.getConnectionStateListenable()).forEach(listener -> { Review Comment: ⚠️ **Internal API cast** — `getConnectionStateListenable()` returns `Listenable<ConnectionStateListener>`, which does not declare `forEach`. The cast to `ListenerManager<?, ?>` works because Curator 5.x backs it with `StandardListenerManager`, but this is an implementation detail not guaranteed by the public API. A Curator major-version bump could change the backing class and break this cast at runtime. Add a comment pinning the assumption: ```suggestion // the group listeners of the policies that are registered on the client. // Curator 5.x: getConnectionStateListenable() is backed by StandardListenerManager (implements ListenerManager, which has forEach). // If this cast breaks on a Curator upgrade, replace with client.getConnectionStateListenable().getListeners() or similar. @SuppressWarnings("unchecked") private static int groupListenersOn(CuratorFramework client) { ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
