This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 136ffd929263 CAMEL-26802: camel-zookeeper-master - MasterRoutePolicy
gives up the mastership when its route is removed, and keeps a slave stopped
(#29583)
136ffd929263 is described below
commit 136ffd929263047c90bc2e6ce87d91e3a670ae06
Author: allthingssecurity <[email protected]>
AuthorDate: Fri Nov 13 03:15:16 2026 +0530
CAMEL-26802: camel-zookeeper-master - MasterRoutePolicy gives up the
mastership when its route is removed, and keeps a slave stopped (#29583)
MasterRoutePolicy joined the ZooKeeper group in doStart and left it only
when
the CamelContext stopped, so a removed route kept the mastership (the route
ran
nowhere), a route added again failed with a NullPointerException or never
started its consumer, and a slave started its consumer on startRoute or
resumeRoute.
- onRemove leaves the group, so another node becomes the master
- doStart joins only when the policy has a route; a route added to a running
CamelContext joins on its RouteAddedEvent, after its consumer was created
- onStart and onResume keep the consumer of a slave stopped
- election and disconnect callbacks ignore a removed route; JMX attributes
read
the group listener once
- with a user CuratorFramework, the group listener is removed from it when
the
policy leaves the group; a failed join undoes itself
- a policy stopped on its own leaves the group as when its route is removed
- 4.24 upgrade-guide entry; the javadoc says to use one policy per route
Follows the same fixes in HazelcastRoutePolicy (CAMEL-26784).
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
components/camel-zookeeper-master/pom.xml | 15 +
.../zookeepermaster/policy/MasterRoutePolicy.java | 334 +++++++++++++---
.../MasterRoutePolicyRouteLifecycleTest.java | 441 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 14 +
4 files changed, 753 insertions(+), 51 deletions(-)
diff --git a/components/camel-zookeeper-master/pom.xml
b/components/camel-zookeeper-master/pom.xml
index 21711e5e3b88..6ef9eb82e874 100644
--- a/components/camel-zookeeper-master/pom.xml
+++ b/components/camel-zookeeper-master/pom.xml
@@ -110,6 +110,21 @@
<scope>test</scope>
</dependency>
+ <!-- the ZooKeeper server that MasterRoutePolicyRouteLifecycleTest
runs in the test JVM needs these two
+ (both are "provided" dependencies of zookeeper) -->
+ <dependency>
+ <groupId>io.dropwizard.metrics</groupId>
+ <artifactId>metrics-core</artifactId>
+ <version>${metrics-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.xerial.snappy</groupId>
+ <artifactId>snappy-java</artifactId>
+ <version>${snappy-java-version}</version>
+ <scope>test</scope>
+ </dependency>
+
<!-- test infra -->
<dependency>
<groupId>org.apache.camel</groupId>
diff --git
a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/policy/MasterRoutePolicy.java
b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/policy/MasterRoutePolicy.java
index f4e633892008..978ae584567d 100644
---
a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/policy/MasterRoutePolicy.java
+++
b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/policy/MasterRoutePolicy.java
@@ -26,6 +26,7 @@ import java.util.concurrent.locks.ReentrantLock;
import org.apache.camel.CamelContext;
import org.apache.camel.CamelContextAware;
+import org.apache.camel.NonManagedService;
import org.apache.camel.Route;
import org.apache.camel.api.management.ManagedAttribute;
import org.apache.camel.api.management.ManagedOperation;
@@ -34,7 +35,11 @@ import
org.apache.camel.component.zookeepermaster.CamelNodeState;
import org.apache.camel.component.zookeepermaster.ContainerIdFactory;
import org.apache.camel.component.zookeepermaster.DefaultContainerIdFactory;
import
org.apache.camel.component.zookeepermaster.ZookeeperGroupListenerSupport;
+import org.apache.camel.component.zookeepermaster.group.Group;
+import org.apache.camel.spi.CamelEvent;
+import org.apache.camel.spi.CamelEvent.RouteAddedEvent;
import org.apache.camel.support.RoutePolicySupport;
+import org.apache.camel.support.SimpleEventNotifierSupport;
import org.apache.camel.support.service.ServiceHelper;
import org.apache.camel.util.ObjectHelper;
import org.apache.camel.util.StringHelper;
@@ -49,6 +54,9 @@ import tools.jackson.databind.json.JsonMapper;
* <p/>
* <b>Important:</b> Make sure to set the route to autoStartup=false as the
route lifecycle is controlled by this route
* policy which will start/stop the route accordingly to being the master in
the zookeeper cluster group.
+ * <p/>
+ * Use one policy instance per route. When the route is removed (or reloaded)
the policy leaves the group, so another
+ * node can become the master.
*/
@ManagedResource(description = "Managed MasterRoutePolicy")
public class MasterRoutePolicy extends RoutePolicySupport implements
CamelContextAware {
@@ -77,10 +85,12 @@ public class MasterRoutePolicy extends RoutePolicySupport
implements CamelContex
private int failedStarts;
// read by the group, Curator connection and retry threads
- private volatile ZookeeperGroupListenerSupport groupListener;
+ private volatile ZookeeperGroupListenerSupport<CamelNodeState>
groupListener;
private volatile CamelNodeState thisNodeState;
private CamelContext camelContext;
- private Route route;
+ // the route of the policy: set when the route is initialized, cleared
when it is removed (under the service lock)
+ private volatile Route route;
+ private RouteAddedNotifier routeAddedNotifier;
@Override
public CamelContext getCamelContext() {
@@ -175,30 +185,27 @@ public class MasterRoutePolicy extends RoutePolicySupport
implements CamelContex
@ManagedAttribute(description = "Are we connected to ZooKeeper")
public boolean isConnected() {
- if (groupListener == null) {
- return false;
- }
- return groupListener.getGroup().isConnected();
+ Group<CamelNodeState> group = currentGroup();
+ return group != null && group.isConnected();
}
@ManagedAttribute(description = "Are we the master")
public boolean isMaster() {
- if (groupListener == null) {
- return false;
- }
- return groupListener.getGroup().isMaster();
+ Group<CamelNodeState> group = currentGroup();
+ return group != null && group.isMaster();
}
@ManagedOperation(description = "Information about all the slaves")
public String slaves() {
- if (groupListener == null) {
+ Group<CamelNodeState> group = currentGroup();
+ if (group == null) {
return null;
}
try {
return JsonMapper.builderWithJackson2Defaults()
.enable(SerializationFeature.INDENT_OUTPUT)
.build()
- .writeValueAsString(groupListener.getGroup().slaves());
+ .writeValueAsString(group.slaves());
} catch (Exception e) {
return null;
}
@@ -206,10 +213,11 @@ public class MasterRoutePolicy extends RoutePolicySupport
implements CamelContex
@ManagedOperation(description = "Information about the last event in the
cluster group")
public String lastEvent() {
- if (groupListener == null) {
+ Group<CamelNodeState> group = currentGroup();
+ if (group == null) {
return null;
}
- Object event = groupListener.getGroup().getLastState();
+ Object event = group.getLastState();
return event != null ? event.toString() : null;
}
@@ -221,7 +229,76 @@ public class MasterRoutePolicy extends RoutePolicySupport
implements CamelContex
@Override
public void onInit(Route route) {
super.onInit(route);
- this.route = route;
+
+ lock.lock();
+ try {
+ // a route that is added to a running CamelContext (a route reload
with the same or a new policy) joins the
+ // group when it is added (routeAdded), as its consumer is not
created yet
+ this.route = route;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ // the route was added to the running CamelContext, after its consumer was
created
+ private void routeAdded(Route added) {
+ lock.lock();
+ try {
+ if (added == this.route && isRunAllowed() && groupListener ==
null) {
+ joinGroup();
+ }
+ } catch (Exception e) {
+ getExceptionHandler().handleException("Failed to join the group
of: " + added.getEndpoint(), e);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public void onRemove(Route route) {
+ lock.lock();
+ try {
+ if (route != this.route) {
+ return;
+ }
+ // the route is removed: leave the group, so that another node
becomes the master and runs the route. The
+ // policy is a service of the CamelContext and keeps running, so
it joins again when the route is added again
+ leaveGroup();
+ this.route = null;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public void onStart(Route route) {
+ keepConsumerStoppedUnlessMaster(route);
+ }
+
+ @Override
+ public void onResume(Route route) {
+ keepConsumerStoppedUnlessMaster(route);
+ }
+
+ // the route was started or resumed (from JMX, the controlbus, a scheduled
route policy, or as it is not set to
+ // autoStartup=false), which started or resumed its consumer: a node that
is not the master keeps it stopped, and
+ // starts it when it is elected (a resumed consumer that is marked as
started would not be started again then)
+ private void keepConsumerStoppedUnlessMaster(Route route) {
+ leadershipLock.lock();
+ try {
+ // a master that has not started its consumer yet starts it with
its pending start (or retry)
+ if (route != this.route || masterConsumer.get() || isLeader()) {
+ return;
+ }
+ LOG.debug("Not the master. Keeping the consumer stopped: {}",
route.getEndpoint());
+ stopConsumer(route.getConsumer());
+ } catch (Exception e) {
+ getExceptionHandler().handleException("Failed to stop the consumer
of: " + route.getEndpoint()
+ + " which only the master
runs",
+ e);
+ } finally {
+ leadershipLock.unlock();
+ }
}
@Override
@@ -231,41 +308,155 @@ public class MasterRoutePolicy extends
RoutePolicySupport implements CamelContex
ObjectHelper.notNull(camelContext, "CamelContext");
StringHelper.notEmpty("groupName", groupName);
- String path = getCamelClusterPath(groupName);
this.retryExecutor = camelContext.getExecutorServiceManager()
.newSingleThreadScheduledExecutor(this,
"ZooKeeperMasterRoutePolicyRetry");
- this.groupListener = new ZookeeperGroupListenerSupport(path,
route.getEndpoint(), onLockOwned(), onDisconnected());
- this.groupListener.setCamelContext(camelContext);
- this.groupListener.setCurator(curator);
-
this.groupListener.setMaximumConnectionTimeout(maximumConnectionTimeout);
- this.groupListener.setZooKeeperUrl(zooKeeperUrl);
- this.groupListener.setZooKeeperPassword(zooKeeperPassword);
- ServiceHelper.startService(groupListener);
+ this.routeAddedNotifier = new RouteAddedNotifier();
+
camelContext.getManagementStrategy().addEventNotifier(routeAddedNotifier);
+ // a policy added to a running CamelContext is started before its
route is initialized: it joins the group
+ // when the route is added
+ if (route != null) {
+ joinGroup();
+ }
+ }
+
+ // under the service lock: joins the group of the nodes that run the route
+ private void joinGroup() throws Exception {
+ String path = getCamelClusterPath(groupName);
+ ZookeeperGroupListenerSupport<CamelNodeState> listener
+ = new ZookeeperGroupListenerSupport<>(path,
route.getEndpoint(), onLockOwned(), onDisconnected());
+ listener.setCamelContext(camelContext);
+ listener.setCurator(curator);
+ listener.setMaximumConnectionTimeout(maximumConnectionTimeout);
+ listener.setZooKeeperUrl(zooKeeperUrl);
+ listener.setZooKeeperPassword(zooKeeperPassword);
+ // set before it starts, as the group may tell it that this node is
the master while it starts
+ this.groupListener = listener;
+ try {
+ ServiceHelper.startService(listener);
+
+ LOG.info("Attempting to become master for endpoint: {} in {} with
singletonID: {}", route.getEndpoint(),
+ getCamelContext(), getGroupName());
+ thisNodeState = createNodeState(route);
+ listener.updateState(thisNodeState);
+ } catch (Exception e) {
+ // undo the half joined state: the JMX attributes and the
leadership checks must not read the group of a
+ // join that failed, and the policy joins again when the route is
added again or the policy is started again
+ try {
+ closeGroup(listener);
+ } catch (Exception e2) {
+ e.addSuppressed(e2);
+ }
+ throw e;
+ }
+ }
- LOG.info("Attempting to become master for endpoint: {} in {} with
singletonID: {}", route.getEndpoint(),
- getCamelContext(), getGroupName());
- thisNodeState = createNodeState();
- groupListener.updateState(thisNodeState);
+ // under the service lock: leaves the group, and this node is no longer
the master
+ private void leaveGroup() {
+ ZookeeperGroupListenerSupport<CamelNodeState> listener = groupListener;
+ if (listener == null) {
+ return;
+ }
+ try {
+ closeGroup(listener);
+ } catch (Exception e) {
+ LOG.warn("Failed to leave the group of: {}. This exception is
ignored.", route.getEndpoint(), e);
+ }
+ if (camelContext.isStopping()) {
+ LOG.debug("Left the group as the route was removed: {}",
route.getEndpoint());
+ } else {
+ LOG.info("Left the group as the route was removed: {}",
route.getEndpoint());
+ }
+ }
+
+ // under the service lock: closes the group the listener joined (or tried
to join), when the route is removed, the
+ // policy stops, or the join failed. Throws the first failure once all of
it is cleaned up
+ private void closeGroup(ZookeeperGroupListenerSupport<CamelNodeState>
listener) throws Exception {
+ // the client the listener registered itself on, which outlives it
+ CuratorFramework client = registeredCurator();
+ // a start (or retry) that checks the leadership from now on does not
start the consumer
+ groupListener = null;
+ Exception failure = null;
+ // the group thread is interrupted before the leadership lock is
taken: a consumer start in flight on it holds the
+ // lock. Closing the group tells the listeners that it is
disconnected, which stops the consumer
+ try {
+ ServiceHelper.stopAndShutdownServices(listener);
+ } catch (Exception e) {
+ failure = e;
+ }
+ if (client != null) {
+ client.getConnectionStateListenable().removeListener(listener);
+ }
+ leadershipLock.lock();
+ try {
+ thisNodeState = null;
+ failedStarts = 0;
+ cancelRetry();
+ Route current = this.route;
+ if (masterConsumer.getAndSet(false) && current != null) {
+ // a start that completed while the group was closed
+ stopConsumer(current.getConsumer());
+ }
+ } catch (Exception e) {
+ if (failure == null) {
+ failure = e;
+ } else {
+ failure.addSuppressed(e);
+ }
+ } finally {
+ leadershipLock.unlock();
+ }
+ if (failure != null) {
+ throw failure;
+ }
+ }
+
+ // the CuratorFramework the group listener registered itself on
(ZookeeperGroupSupport.setCurator): the curator of
+ // the policy, or else the one named curator in the registry. It does not
register on a client its group factory
+ // creates, which the factory closes
+ private CuratorFramework registeredCurator() {
+ if (curator != null) {
+ return curator;
+ }
+ try {
+ return camelContext.getRegistry().lookupByNameAndType("curator",
CuratorFramework.class);
+ } catch (RuntimeException e) {
+ return null;
+ }
}
@Override
protected void doStop() throws Exception {
super.doStop();
+ if (routeAddedNotifier != null) {
+
camelContext.getManagementStrategy().removeEventNotifier(routeAddedNotifier);
+ routeAddedNotifier = null;
+ }
+
// the retry thread and the group thread are interrupted first (as the
group did before): a consumer start in
// flight on them holds the leadership lock, and the stop must not
wait for it to complete
if (retryExecutor != null) {
camelContext.getExecutorServiceManager().shutdownNow(retryExecutor);
retryExecutor = null;
}
- ServiceHelper.stopAndShutdownServices(groupListener);
- leadershipLock.lock();
- try {
- failedStarts = 0;
- cancelRetry();
- masterConsumer.set(false);
- } finally {
- leadershipLock.unlock();
+ // the same clean up as when the route is removed (the policy may be
stopped on its own, from JMX), so that a
+ // shared CuratorFramework does not keep the listener and the JMX
attributes do not read a closed group
+ ZookeeperGroupListenerSupport<CamelNodeState> listener = groupListener;
+ if (listener != null) {
+ try {
+ closeGroup(listener);
+ } catch (Exception e) {
+ LOG.warn("Failed to leave the group of: {}. This exception is
ignored.", groupName, e);
+ }
+ } else {
+ leadershipLock.lock();
+ try {
+ failedStarts = 0;
+ cancelRetry();
+ masterConsumer.set(false);
+ } finally {
+ leadershipLock.unlock();
+ }
}
}
@@ -282,21 +473,23 @@ public class MasterRoutePolicy extends RoutePolicySupport
implements CamelContex
}
// the event may be stale: the group sets connected to false
before it tells the listeners that it is
// disconnected, and the disconnect may have been handled since
the event checked it
- if (!isRunAllowed() || masterConsumer.get() || !isLeader()) {
+ Route current = this.route;
+ if (!isRunAllowed() || masterConsumer.get() || !isLeader() ||
current == null
+ || current.getConsumer() == null) {
return;
}
try {
// ensure endpoint is also started
- LOG.info("Elected as master. Starting consumer: {}",
route.getEndpoint());
- startConsumer(route.getConsumer());
+ LOG.info("Elected as master. Starting consumer: {}",
current.getEndpoint());
+ startConsumer(current.getConsumer());
} catch (Exception e) {
if (!isRunAllowed()) {
- LOG.debug("Stopping while starting master consumer for:
{}", route.getEndpoint(), e);
+ LOG.debug("Stopping while starting master consumer for:
{}", current.getEndpoint(), e);
return;
}
// try again later, as long as this node is the master (no
later event may come)
failedStarts++;
- getExceptionHandler().handleException("Failed to start master
consumer for: " + route.getEndpoint()
+ getExceptionHandler().handleException("Failed to start master
consumer for: " + current.getEndpoint()
+ " (attempt #" +
failedStarts + "). Trying again in " + retryDelay
+ " millis.",
e);
@@ -310,22 +503,32 @@ public class MasterRoutePolicy extends RoutePolicySupport
implements CamelContex
try {
// Lets show we are starting the consumer.
- thisNodeState = createNodeState();
- thisNodeState.setStarted(true);
- groupListener.updateState(thisNodeState);
+ ZookeeperGroupListenerSupport<CamelNodeState> listener =
groupListener;
+ if (listener != null) {
+ thisNodeState = createNodeState(current);
+ thisNodeState.setStarted(true);
+ listener.updateState(thisNodeState);
+ }
} catch (Exception e) {
- getExceptionHandler().handleException("Failed to publish the
started state of: " + route.getEndpoint(), e);
+ getExceptionHandler().handleException("Failed to publish the
started state of: " + current.getEndpoint(), e);
}
- LOG.info("Elected as master. Consumer started: {}",
route.getEndpoint());
+ LOG.info("Elected as master. Consumer started: {}",
current.getEndpoint());
} finally {
leadershipLock.unlock();
}
}
private boolean isLeader() {
- return groupListener != null && groupListener.getGroup() != null &&
groupListener.getGroup().isConnected()
- && groupListener.getGroup().isMaster();
+ Group<CamelNodeState> group = currentGroup();
+ return group != null && group.isConnected() && group.isMaster();
+ }
+
+ // the group this node is in, or null: the listener is cleared when the
route is removed, and has no group before it
+ // is started
+ private Group<CamelNodeState> currentGroup() {
+ ZookeeperGroupListenerSupport<CamelNodeState> listener = groupListener;
+ return listener != null ? listener.getGroup() : null;
}
// under the lock
@@ -363,9 +566,16 @@ public class MasterRoutePolicy extends RoutePolicySupport
implements CamelContex
masterConsumer.set(false);
failedStarts = 0;
cancelRetry();
- stopConsumer(route.getConsumer());
- } catch (Exception e) {
- LOG.warn("Failed to stop master consumer: {}",
route.getEndpoint(), e);
+ Route current = this.route;
+ // the route was removed
+ if (current == null) {
+ return;
+ }
+ try {
+ stopConsumer(current.getConsumer());
+ } catch (Exception e) {
+ LOG.warn("Failed to stop master consumer: {}",
current.getEndpoint(), e);
+ }
} finally {
leadershipLock.unlock();
}
@@ -380,11 +590,33 @@ public class MasterRoutePolicy extends RoutePolicySupport
implements CamelContex
return path;
}
- private CamelNodeState createNodeState() {
+ private CamelNodeState createNodeState(Route route) {
String containerId = getContainerIdFactory().newContainerId();
CamelNodeState state = new CamelNodeState(getGroupName(), containerId);
state.setConsumer(route.getEndpoint().getEndpointUri());
return state;
}
+ // joins the group when the route of the policy is added to the running
CamelContext
+ private final class RouteAddedNotifier extends SimpleEventNotifierSupport
implements NonManagedService {
+
+ RouteAddedNotifier() {
+ // only route events, so the notifier adds no cost to the exchanges
+ setIgnoreCamelContextInitEvents(true);
+ setIgnoreCamelContextEvents(true);
+ setIgnoreServiceEvents(true);
+ setIgnoreExchangeEvents(true);
+ setIgnoreStepEvents(true);
+ }
+
+ @Override
+ public boolean isEnabled(CamelEvent event) {
+ return event instanceof RouteAddedEvent;
+ }
+
+ @Override
+ public void notify(CamelEvent event) {
+ routeAdded(((RouteAddedEvent) event).getRoute());
+ }
+ }
}
diff --git
a/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/policy/MasterRoutePolicyRouteLifecycleTest.java
b/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/policy/MasterRoutePolicyRouteLifecycleTest.java
new file mode 100644
index 000000000000..372743d8d645
--- /dev/null
+++
b/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();
+ }
+ }
+
+ private static MasterRoutePolicy policyOf(CamelContext context) {
+ return (MasterRoutePolicy)
context.getRoute(ROUTE).getRoutePolicyList().get(0);
+ }
+
+ private static boolean isConsumerStarted(CamelContext context) {
+ Consumer consumer = context.getRoute(ROUTE).getConsumer();
+ return ServiceHelper.isStarted(consumer);
+ }
+
+ private static void awaitConsuming(CamelContext context) {
+ await().atMost(30, TimeUnit.SECONDS).untilAsserted(
+ () -> assertTrue(policyOf(context).isMaster() &&
isConsumerStarted(context),
+ "The node must become the master and start its
consumer"));
+ }
+
+ private static void awaitSlave(CamelContext context) {
+ await().atMost(30, TimeUnit.SECONDS)
+ .until(() -> policyOf(context).isConnected() &&
policyOf(context).thisNode() != null);
+ assertFalse(policyOf(context).isMaster());
+ assertFalse(isConsumerStarted(context));
+ }
+
+ // a message sent to the node is routed, so the consumer runs and is not
only marked as started
+ private static void assertConsumes(CamelContext context) throws Exception {
+ MockEndpoint mock = context.getEndpoint("mock:out",
MockEndpoint.class);
+ mock.reset();
+ mock.expectedBodiesReceived("hello");
+ try (ProducerTemplate template = context.createProducerTemplate()) {
+ template.sendBody("seda:in", "hello");
+ }
+ MockEndpoint.assertIsSatisfied(context, 20, TimeUnit.SECONDS);
+ }
+
+ // an endpoint that runs a check when its consumer is created
+ private static final class WatchedEndpoint extends DefaultEndpoint {
+ private final Runnable onCreateConsumer;
+
+ WatchedEndpoint(Runnable onCreateConsumer) {
+ this.onCreateConsumer = onCreateConsumer;
+ }
+
+ @Override
+ protected String createEndpointUri() {
+ return "watched:in";
+ }
+
+ @Override
+ public Producer createProducer() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) {
+ onCreateConsumer.run();
+ return new DefaultConsumer(this, processor);
+ }
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 14a1ddf8a885..949c547b870c 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -3887,6 +3887,20 @@ a `NoNodeException`, as it did without `create=true`.
With `listChildren=true`, a failed write or delete of an InOut exchange now
fails the exchange, instead of being
replaced by the listing of the child _znodes_.
+=== camel-zookeeper-master - MasterRoutePolicy and route removal, and the
consumer of a slave
+
+A `MasterRoutePolicy` stayed in the ZooKeeper group until the CamelContext
stopped, also after its route was removed
+(with `removeRoute`, or by a route reload), so no other node became the master
and the route ran nowhere in the
+cluster. The policy now leaves the group when its route is removed, so another
node takes over, and joins it again when
+the route is added again. So a route reloaded on the master can now move to
another node.
+
+On a node that is not the master, the policy now stops the consumer of its
route again right after the route was
+started or resumed (from JMX, the controlbus or a route policy), and starts it
when the node becomes the master. Before,
+the consumer of such a route kept running next to the one of the master. This
includes a route left at
+`autoStartup=true`: it now consumes only once its node is the master, also
when the CamelContext starts, so it does not
+consume while the node cannot reach ZooKeeper. Setting the route to
`autoStartup=false` is still recommended, so the
+consumer does not run for a moment when the CamelContext starts.
+
=== camel-milo - each client connection has thread pools of its own
A `milo-client` or `milo-browse` connection now handles the outcome of its
write, method call, read and browse
