This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch backport/25611-to-camel-4.18.x in repository https://gitbox.apache.org/repos/asf/camel.git
commit 4655b10384b83d6176981e36eab59da23dbf67da 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 (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(); + } +}
