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 e227737de332 CAMEL-24457: Fix ZooKeeper cluster split-brain on leader
isolation
e227737de332 is described below
commit e227737de3325f997d6f99f9f4c4b8bc5d7af7cc
Author: Mustafa Kamal ALmonayer
<[email protected]>
AuthorDate: Thu Aug 27 08:52:30 2026 +0300
CAMEL-24457: Fix ZooKeeper cluster split-brain on leader isolation
ZooKeeperClusterView.CamelLeaderElectionListener.takeLeadership() fired
its leadership-lost event while Curator's hasLeadership() still
returned true, since the event fired from inside takeLeadership()
before Curator's internals had flipped leadership state. This caused
ClusteredRoutePolicy to see no leadership change and keep the route
running on a node that had lost its ZooKeeper session, while a healthy
node elected a new leader and started the same route — a split-brain
with duplicate message processing.
Track leadership state in a view-owned volatile flag instead of
delegating to leaderSelector.hasLeadership(), set it before firing the
leadership-gained/lost events in takeLeadership()'s try/finally, and
close (not just interrupt) the LeaderSelector in doStop() so a
deliberate stop cannot race an auto-requeue. autoRequeue() is enabled
in doStart() so the node re-enters the election after ZooKeeper
reconnects.
Adds ZooKeeperClusterViewLeadershipLostIT to exercise the split-brain
scenario end-to-end: acquire leadership, lose it on ZK session loss,
and reclaim it after reconnect.
Closes #25611
---
.../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();
+ }
+}
