You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2026/08/27 05:57:11 UTC
(camel) branch backport/25611-to-camel-4.22.x created (now 31082d5175b0)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a change to branch backport/25611-to-camel-4.22.x
in repository https://gitbox.apache.org/repos/asf/camel.git
at 31082d5175b0 CAMEL-24457: Fix ZooKeeper cluster split-brain on leader isolation
This branch includes the following new commits:
new 31082d5175b0 CAMEL-24457: Fix ZooKeeper cluster split-brain on leader isolation
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(camel) 01/01: CAMEL-24457: Fix ZooKeeper cluster split-brain on leader isolation
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch backport/25611-to-camel-4.22.x
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 31082d5175b0083d9a246e787e3fee75fa9a4d25
Author: Mustafa Kamal ALmonayer <89...@users.noreply.github.com>
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
(cherry picked from commit e227737de3325f997d6f99f9f4c4b8bc5d7af7cc)
---
.../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();
+ }
+}