This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch camel-4.18.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.18.x by this push:
new 9b5301f28dfe [backport camel-4.18.x] CAMEL-24457: Fix ZooKeeper
cluster split-brain on leader isolation (#25791)
9b5301f28dfe is described below
commit 9b5301f28dfeb4ef271033d9dce7399975eaa209
Author: Claus Ibsen <[email protected]>
AuthorDate: Thu Aug 27 08:46:24 2026 +0200
[backport camel-4.18.x] CAMEL-24457: Fix ZooKeeper cluster split-brain on
leader isolation (#25791)
Straight cherry-pick of #25611 onto camel-4.18.x.
Fixes a split-brain where a node that lost its ZooKeeper session kept
running a clustered route because the leadership-lost event fired
while Curator's hasLeadership() still returned true. Leadership state
is now tracked in a view-owned volatile flag, and the LeaderSelector
is closed (not just interrupted) on doStop() so a deliberate stop
can't race the ZK-reconnect auto-requeue.
See #25611 for full details, review discussion, and test plan.
Closes #25791
---
.../zookeeper/cluster/ZooKeeperClusterView.java | 30 +++---
.../ZooKeeperClusterViewLeadershipLostIT.java | 117 +++++++++++++++++++++
2 files changed, 132 insertions(+), 15 deletions(-)
diff --git
a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/cluster/ZooKeeperClusterView.java
b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/cluster/ZooKeeperClusterView.java
index c276b152d26a..594134bb4207 100644
---
a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/cluster/ZooKeeperClusterView.java
+++
b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/cluster/ZooKeeperClusterView.java
@@ -47,6 +47,7 @@ final class ZooKeeperClusterView extends
AbstractCamelClusterView {
private final CuratorFramework client;
private final CuratorLocalMember localMember;
private volatile LeaderSelector leaderSelector;
+ private volatile boolean leader;
public ZooKeeperClusterView(CamelClusterService cluster,
ZooKeeperCuratorConfiguration configuration,
CuratorFramework client, String namespace) {
@@ -110,6 +111,7 @@ final class ZooKeeperClusterView extends
AbstractCamelClusterView {
if (leaderSelector == null) {
leaderSelector = new LeaderSelector(client, getFullPath(), new
CamelLeaderElectionListener());
leaderSelector.setId(getClusterService().getId());
+ leaderSelector.autoRequeue();
leaderSelector.start();
} else {
leaderSelector.requeue();
@@ -118,16 +120,11 @@ final class ZooKeeperClusterView extends
AbstractCamelClusterView {
@Override
protected void doStop() throws Exception {
- if (leaderSelector != null) {
- leaderSelector.interruptLeadership();
- fireLeadershipChangedEvent(getLeader().orElse(null));
- }
- }
-
- @Override
- protected void doShutdown() throws Exception {
- if (leaderSelector != null) {
- leaderSelector.close();
+ LeaderSelector selector = leaderSelector;
+ leaderSelector = null;
+ if (selector != null) {
+ leader = false;
+ selector.close();
}
}
@@ -142,6 +139,7 @@ final class ZooKeeperClusterView extends
AbstractCamelClusterView {
private final class CamelLeaderElectionListener extends
LeaderSelectorListenerAdapter {
@Override
public void takeLeadership(CuratorFramework curatorFramework) throws
Exception {
+ leader = true;
fireLeadershipChangedEvent(localMember);
BlockingTask task =
Tasks.foregroundTask().withBudget(Budgets.iterationBudget()
@@ -149,17 +147,19 @@ final class ZooKeeperClusterView extends
AbstractCamelClusterView {
.withInterval(Duration.ofSeconds(5))
.build())
.build();
-
- task.run(getCamelContext(), () -> !isRunAllowed());
-
- fireLeadershipChangedEvent(getLeader().orElse(null));
+ try {
+ task.run(getCamelContext(), () -> !isRunAllowed());
+ } finally {
+ leader = false;
+ fireLeadershipChangedEvent((CamelClusterMember) null);
+ }
}
}
private final class CuratorLocalMember implements CamelClusterMember {
@Override
public boolean isLeader() {
- return leaderSelector != null && leaderSelector.hasLeadership();
+ return leader;
}
@Override
diff --git
a/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/cluster/integration/ZooKeeperClusterViewLeadershipLostIT.java
b/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/cluster/integration/ZooKeeperClusterViewLeadershipLostIT.java
new file mode 100644
index 000000000000..f23b136da743
--- /dev/null
+++
b/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/cluster/integration/ZooKeeperClusterViewLeadershipLostIT.java
@@ -0,0 +1,117 @@
+/*
+ * 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.zookeeper.cluster.integration;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.cluster.CamelClusterView;
+import org.apache.camel.component.zookeeper.cluster.ZooKeeperClusterService;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.impl.cluster.ClusteredRoutePolicy;
+import org.apache.camel.test.infra.common.services.ContainerService;
+import org.apache.camel.test.infra.zookeeper.services.ZooKeeperService;
+import org.apache.camel.test.infra.zookeeper.services.ZooKeeperServiceFactory;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.testcontainers.containers.GenericContainer;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+class ZooKeeperClusterViewLeadershipLostIT {
+
+ private static final String NAMESPACE = "my-ns";
+ private static final String ROUTE_ID = "clustered-route";
+
+ @RegisterExtension
+ static ZooKeeperService service = ZooKeeperServiceFactory.createService();
+
+ @Test
+ void leadershipIsReleasedAndReacquiredAroundAZooKeeperOutage() throws
Exception {
+ GenericContainer<?> zooKeeper = zooKeeperContainer();
+
+ try (DefaultCamelContext context = new DefaultCamelContext()) {
+ ZooKeeperClusterService clusterService = new
ZooKeeperClusterService();
+ clusterService.setId("node-1");
+ clusterService.setNodes(service.serverUrls());
+ clusterService.setBasePath("/camel");
+
+ clusterService.setSessionTimeout(5000);
+
+ context.disableJMX();
+ context.addService(clusterService);
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("timer:zookeeper?period=1000")
+ .routeId(ROUTE_ID)
+
.routePolicy(ClusteredRoutePolicy.forNamespace(NAMESPACE))
+ .to("log:zookeeper?level=DEBUG");
+ }
+ });
+
+ context.start();
+
+ CamelClusterView view = clusterService.getView(NAMESPACE);
+
+ await().atMost(1, TimeUnit.MINUTES)
+ .untilAsserted(() -> {
+ assertEquals(true,
+ view.getLocalMember().isLeader(),
+ "the only node of the cluster must be the
leader");
+ assertEquals(true,
+
context.getRouteController().getRouteStatus(ROUTE_ID).isStarted(),
+ "the leader must have started the clustered
route");
+ });
+
+
zooKeeper.getDockerClient().pauseContainerCmd(zooKeeper.getContainerId()).exec();
+
+ try {
+ await().atMost(1, TimeUnit.MINUTES)
+ .untilAsserted(() -> {
+ assertEquals(false,
+ view.getLocalMember().isLeader(),
+ "the leadership must be given up once
ZooKeeper is no longer reachable");
+ assertEquals(false,
+
context.getRouteController().getRouteStatus(ROUTE_ID).isStarted(),
+ "the clustered route must be stopped once
the leadership is lost");
+ });
+ } finally {
+
zooKeeper.getDockerClient().unpauseContainerCmd(zooKeeper.getContainerId()).exec();
+ }
+
+ await().atMost(1, TimeUnit.MINUTES)
+ .untilAsserted(() -> {
+ assertEquals(true,
+ view.getLocalMember().isLeader(),
+ "the node must re-enter the election once
ZooKeeper is reachable again");
+ assertEquals(true,
+
context.getRouteController().getRouteStatus(ROUTE_ID).isStarted(),
+ "the clustered route must be restarted once
the leadership is taken back");
+ });
+ }
+ }
+
+ private static GenericContainer<?> zooKeeperContainer() {
+ assumeTrue(service instanceof ContainerService<?>,
+ "This test requires the local ZooKeeper container infra
service");
+
+ return ((ContainerService<?>) service).getContainer();
+ }
+}
