From 9196faed730fa97716d2ef9f30a3df26a14eb765 Mon Sep 17 00:00:00 2001 From: duankaixuan <1417048384@qq.com> Date: Wed, 16 Sep 2026 17:53:00 +0800 Subject: [PATCH 1/3] [server] Recover coordinator leader election when ZK election parent is garbage-collected When the ZooKeeper session of the coordinator expires, ZooKeeper deletes the ephemeral LeaderLatch node and may then garbage-collect the empty container election parent /coordinators/election. On reconnection, the Curator 5.4.0 LeaderLatch silently ignores the NONODE result of listing the election path and stays permanently idle, so a single-coordinator cluster remains without a leader until it is restarted (upstream CURATOR-724, fixed in Curator 5.8.0 which Fluss does not shade yet). Recreate the election parent as a persistent znode on RECONNECTED before the latch's own reconnection handling, so the latch can re-create its election node and the coordinator regains leadership without a restart. The recovery piggybacks on a LeaderLatch subclass since Curator notifies connection-state listeners in ConcurrentHashMap iteration order, which does not follow registration order. Add a regression test that kills the ZK session and deletes the empty election parent, verifying that leadership is regained without restart; it fails with a timeout before the fix and passes after. Co-Authored-By: Qoder AI-Model: Qoder Auto AI-Contributed/Feature: 73/73 AI-Contributed/UT: 93/93 --- .../CoordinatorLeaderElection.java | 71 ++++++++++++++++++- .../CoordinatorHighAvailabilityITCase.java | 60 ++++++++++++++++ 2 files changed, 129 insertions(+), 2 deletions(-) diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorLeaderElection.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorLeaderElection.java index 0ba51dafe09..4fd8aba48d8 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorLeaderElection.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorLeaderElection.java @@ -22,8 +22,11 @@ import org.apache.fluss.exception.CoordinatorEpochFencedException; import org.apache.fluss.server.zk.ZooKeeperClient; import org.apache.fluss.server.zk.data.ZkData; +import org.apache.fluss.shaded.curator5.org.apache.curator.framework.CuratorFramework; import org.apache.fluss.shaded.curator5.org.apache.curator.framework.recipes.leader.LeaderLatch; import org.apache.fluss.shaded.curator5.org.apache.curator.framework.recipes.leader.LeaderLatchListener; +import org.apache.fluss.shaded.curator5.org.apache.curator.framework.state.ConnectionState; +import org.apache.fluss.shaded.zookeeper3.org.apache.zookeeper.CreateMode; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -77,6 +80,7 @@ public class CoordinatorLeaderElection implements AutoCloseable { private static final long DEFAULT_CLOSE_TIMEOUT_MS = 10000L; private final String serverId; + private final CuratorFramework curatorClient; private final LeaderLatch leaderLatch; // Single-threaded executor to run leader init/cleanup callbacks outside Curator's EventThread. // Curator's LeaderLatchListener callbacks run on its internal EventThread; performing @@ -101,9 +105,10 @@ public CoordinatorLeaderElection(ZooKeeperClient zkClient, String serverId) { checkArgument(closeTimeoutMs > 0, "Close timeout must be positive."); this.serverId = serverId; this.closeTimeoutMs = closeTimeoutMs; + this.curatorClient = zkClient.getCuratorClient(); this.leaderLatch = - new LeaderLatch( - zkClient.getCuratorClient(), + new RecoveringLeaderLatch( + curatorClient, ZkData.CoordinatorElectionZNode.path(), String.valueOf(serverId)); this.leaderCallbackExecutor = @@ -313,4 +318,66 @@ private enum State { STANDBY, CLOSED } + + /** + * A {@link LeaderLatch} that recreates the election parent znode if it has disappeared. + * + *

The election parent is created as a container znode and is garbage-collected by ZooKeeper + * once the session expires and the ephemeral latch node is deleted. On reconnect, the + * LeaderLatch ignores the NONODE result of listing the election path and stays permanently + * idle. Recreating the parent before the latch's own reconnection handling lets the latch + * re-create its election node and regain leadership without a restart. + * + *

The recovery piggybacks on {@code handleStateChange} so that it strictly precedes the + * latch's reconnection handling: registering a separate connection-state listener would not + * guarantee this order, as listeners are notified in {@code ConcurrentHashMap} iteration order, + * which does not follow registration order. + */ + private class RecoveringLeaderLatch extends LeaderLatch { + + RecoveringLeaderLatch(CuratorFramework client, String latchPath, String id) { + super(client, latchPath, id); + } + + @Override + protected void handleStateChange(ConnectionState newState) { + if (newState == ConnectionState.RECONNECTED) { + recoverElectionParentIfNeeded(); + } + super.handleStateChange(newState); + } + + /** + * Recreates the election parent znode if it no longer exists. + * + *

The znode is recreated as PERSISTENT: unlike a container znode, a persistent znode is + * never garbage-collected when empty, so a future session expiration cannot remove the + * election parent again. + */ + private void recoverElectionParentIfNeeded() { + String electionPath = ZkData.CoordinatorElectionZNode.path(); + try { + if (curatorClient.checkExists().forPath(electionPath) == null) { + curatorClient + .create() + .creatingParentsIfNeeded() + .withMode(CreateMode.PERSISTENT) + .forPath(electionPath); + LOG.warn( + "Coordinator server {}: election parent {} was missing, recreate it " + + "to recover the leader election.", + serverId, + electionPath); + } + } catch (Exception e) { + // Ignore the failure and keep the latch's own reconnection handling unchanged: a + // later reconnection event retries the recovery. + LOG.error( + "Coordinator server {}: failed to recover the election parent {}.", + serverId, + electionPath, + e); + } + } + } } diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorHighAvailabilityITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorHighAvailabilityITCase.java index 20f34e3ad24..eef65854614 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorHighAvailabilityITCase.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorHighAvailabilityITCase.java @@ -43,6 +43,7 @@ import org.apache.fluss.server.zk.ZooKeeperExtension; import org.apache.fluss.server.zk.data.CoordinatorAddress; import org.apache.fluss.server.zk.data.LeaderAndIsr; +import org.apache.fluss.server.zk.data.ZkData; import org.apache.fluss.shaded.curator5.org.apache.curator.framework.CuratorFramework; import org.apache.fluss.shaded.zookeeper3.org.apache.zookeeper.KeeperException; import org.apache.fluss.shaded.zookeeper3.org.apache.zookeeper.Watcher; @@ -225,6 +226,65 @@ void testLeaderLosesLeadershipAndReElected() throws Exception { createGatewayForServer(leader).metadata(new MetadataRequest()).get(); } + /** + * Regression test for #4362: a single-coordinator cluster must regain leadership without a + * restart after a ZooKeeper session expiration, even when the now-empty election parent node + * has been deleted. + * + *

The LeaderLatch creates the election parent {@code /coordinators/election} as a container + * znode. When the session expires, ZooKeeper deletes the ephemeral latch node, and the empty + * container parent can then be garbage-collected by ZooKeeper while the coordinator is still + * recovering. This is simulated here by deleting the parent right after the session expiration. + * The coordinator must reconnect, re-create its election node, and become leader again without + * a restart. + */ + @Test + void testRegainsLeadershipAfterSessionExpirationWithElectionParentDeleted() throws Exception { + // single coordinator: no standby can take over, matching the #4362 scenario + coordinatorServer1 = new CoordinatorServer(createConfiguration()); + coordinatorServer1.start(); + waitUntilCoordinatorServerElected(); + + String electionPath = ZkData.CoordinatorElectionZNode.path(); + assertThat(zookeeperClient.getStat(electionPath)) + .as("election parent should exist while the latch node exists") + .isPresent(); + + // kill the coordinator's ZK session: ZooKeeper deletes the ephemeral latch node + killZkSession(coordinatorServer1); + + // wait until the latch node is gone, then simulate ZooKeeper's container GC by + // deleting the now-empty election parent + waitUntil( + () -> { + try { + return zookeeperClient.getChildren(electionPath).isEmpty(); + } catch (Exception e) { + return false; + } + }, + Duration.ofSeconds(30), + "latch node was not deleted after session expiration"); + zookeeperClient.deletePath(electionPath); + + // the coordinator reconnects with a new session and registers itself again + waitUntilServerRegistered(coordinatorServer1); + + // regression assertion for #4362: leadership must be regained without a restart + waitUntil( + () -> coordinatorServer1.getCoordinatorService().isLeader(), + Duration.ofSeconds(30), + "Coordinator did not regain leadership after session expiration and election " + + "parent deletion (#4362)"); + createGatewayForServer(coordinatorServer1).metadata(new MetadataRequest()).get(); + + // the recovered leader must actually participate in the election again: its latch + // node must exist under the election parent + assertThat(zookeeperClient.getChildren(electionPath)) + .as("recovered leader should have its election latch node") + .isNotEmpty(); + } + /** * Regression test for #3625: a standby coordinator must keep applying dynamic config changes so * that, after failover, the promoted leader uses the latest config rather than a stale snapshot From 30946f04d4f5b9b1bc8e5c9173879af0d042295a Mon Sep 17 00:00:00 2001 From: duankaixuan <1417048384@qq.com> Date: Thu, 17 Sep 2026 11:42:33 +0800 Subject: [PATCH 2/3] [server] Restart coordinator leader election after ZK session loss MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When the ZooKeeper session of the coordinator expires, ZooKeeper deletes the ephemeral election node, so the latch's election participation becomes stale: the shaded Curator 5.4.0 LeaderLatch then fails to recover if the empty election parent is also garbage-collected, silently ignoring the NONODE result of listing the election path, and a single-coordinator cluster stays without a leader until restart. Instead of repairing the election parent, restart the election at the lifecycle level: when the session is lost, abandon the old participation and, once the connection returns, start a fresh LeaderLatch. The fresh latch registers a new election node for the current session with parents created as needed — the Curator create path recovers a missing election parent — and determines leadership through the normal election, including epoch fencing. A short suspension that keeps the session does not restart the election; the existing latch revalidates leadership on reconnection. Add a regression test that kills the ZK session and deletes the empty election parent, verifying that leadership is regained without restart. --- .../CoordinatorLeaderElection.java | 159 ++++++++++-------- .../CoordinatorHighAvailabilityITCase.java | 75 +++++++-- 2 files changed, 145 insertions(+), 89 deletions(-) diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorLeaderElection.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorLeaderElection.java index 4fd8aba48d8..9c7c9c81808 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorLeaderElection.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorLeaderElection.java @@ -26,7 +26,7 @@ import org.apache.fluss.shaded.curator5.org.apache.curator.framework.recipes.leader.LeaderLatch; import org.apache.fluss.shaded.curator5.org.apache.curator.framework.recipes.leader.LeaderLatchListener; import org.apache.fluss.shaded.curator5.org.apache.curator.framework.state.ConnectionState; -import org.apache.fluss.shaded.zookeeper3.org.apache.zookeeper.CreateMode; +import org.apache.fluss.shaded.curator5.org.apache.curator.framework.state.ConnectionStateListener; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -55,6 +55,13 @@ *

  • Can be re-elected as leader multiple times * * + *

    When the ZooKeeper session is lost, ZooKeeper deletes the ephemeral election node of this + * coordinator, so the previous election participation becomes stale. After the connection returns, + * the election is restarted with a fresh latch: the fresh latch registers a new election node + * belonging to the current session and determines leadership through the normal election. A short + * connection suspension that keeps the session does not restart the election; the existing latch + * revalidates leadership on reconnection instead. + * *

    Leadership callbacks and state transitions are serialized by {@code leaderCallbackExecutor}. * The state machine is: * @@ -81,12 +88,18 @@ public class CoordinatorLeaderElection implements AutoCloseable { private final String serverId; private final CuratorFramework curatorClient; - private final LeaderLatch leaderLatch; + + /** + * The latch currently participating in the election. Replaced with a fresh latch when the + * ZooKeeper session was lost and the connection returns, see {@link #restartElection()}. + */ + private volatile LeaderLatch leaderLatch; + // Single-threaded executor to run leader init/cleanup callbacks outside Curator's EventThread. // Curator's LeaderLatchListener callbacks run on its internal EventThread; performing // synchronous ZK operations there causes deadlock because ZK response dispatch also // needs that same thread. Serial execution also guarantees that leader initialization, - // cleanup, and state transitions never overlap. + // cleanup, state transitions, and election restarts never overlap. private final ExecutorService leaderCallbackExecutor; private final CompletableFuture closeFuture = new CompletableFuture<>(); private final long closeTimeoutMs; @@ -94,8 +107,29 @@ public class CoordinatorLeaderElection implements AutoCloseable { private final AtomicBoolean closing = new AtomicBoolean(false); private volatile State state = State.INITIAL; + private volatile Runnable initLeaderServices; private volatile Consumer cleanupLeaderServices; + /** + * Marks that the ZooKeeper session was lost; the election restarts on the next reconnection. + */ + private final AtomicBoolean sessionLost = new AtomicBoolean(false); + + /** + * Watches for a lost ZooKeeper session and schedules the election restart for the first + * reconnection after it. A reconnection without a preceding session loss keeps the current + * latch, whose election node still belongs to the alive session. + */ + private final ConnectionStateListener sessionLossListener = + (client, newState) -> { + if (newState == ConnectionState.LOST) { + sessionLost.set(true); + } else if (newState == ConnectionState.RECONNECTED + && sessionLost.compareAndSet(true, false)) { + submitLeadershipEvent(this::restartElection); + } + }; + public CoordinatorLeaderElection(ZooKeeperClient zkClient, String serverId) { this(zkClient, serverId, DEFAULT_CLOSE_TIMEOUT_MS); } @@ -106,11 +140,6 @@ public CoordinatorLeaderElection(ZooKeeperClient zkClient, String serverId) { this.serverId = serverId; this.closeTimeoutMs = closeTimeoutMs; this.curatorClient = zkClient.getCuratorClient(); - this.leaderLatch = - new RecoveringLeaderLatch( - curatorClient, - ZkData.CoordinatorElectionZNode.path(), - String.valueOf(serverId)); this.leaderCallbackExecutor = Executors.newSingleThreadExecutor( r -> { @@ -126,15 +155,33 @@ public CoordinatorLeaderElection(ZooKeeperClient zkClient, String serverId) { * Starts the leader election process asynchronously. * *

    After the first election, the server will continue to participate in future elections. - * When re-elected as leader, the initLeaderServices callback will be invoked again. + * When re-elected as leader, the initLeaderServices callback will be invoked again. If the + * ZooKeeper session is lost, the election is restarted with a fresh latch once the connection + * returns. * * @param initLeaderServices the callback to initialize leader services once elected * @param cleanupLeaderServices the callback to clean up leader services when losing leadership */ public void startElectLeaderAsync( Runnable initLeaderServices, Consumer cleanupLeaderServices) { + this.initLeaderServices = initLeaderServices; this.cleanupLeaderServices = cleanupLeaderServices; - leaderLatch.addListener( + curatorClient.getConnectionStateListenable().addListener(sessionLossListener); + startLatch(); + } + + /** Creates a latch, subscribes to its leadership notifications, and starts it. */ + private void startLatch() { + if (closing.get()) { + return; + } + + LeaderLatch latch = + new LeaderLatch( + curatorClient, + ZkData.CoordinatorElectionZNode.path(), + String.valueOf(serverId)); + latch.addListener( new LeaderLatchListener() { @Override public void isLeader() { @@ -146,20 +193,46 @@ public void notLeader() { submitLeadershipEvent(CoordinatorLeaderElection.this::becomeStandby); } }); - + this.leaderLatch = latch; try { - leaderLatch.start(); + latch.start(); LOG.info("Coordinator server {} started leader election.", serverId); } catch (Exception e) { LOG.error("Failed to start LeaderLatch for server {}", serverId, e); } } + /** + * Restarts the election after the ZooKeeper session was lost and the connection has returned. + * + *

    The session loss deletes the ephemeral election node of this coordinator, so the old latch + * no longer represents a valid participation: its election node is gone and any leadership it + * held was already revoked when the session was lost. The old latch is abandoned and a fresh + * latch is started instead, which registers a new election node for the current session and + * determines leadership through the normal election, including the epoch fencing in the leader + * initialization. Creating the new election node also recreates the election parent if it was + * garbage-collected while empty, so the restart recovers even when the election parent is + * missing. + */ + private void restartElection() { + LOG.info( + "Coordinator server {}: ZooKeeper session was lost, restarting the leader " + + "election.", + serverId); + try { + leaderLatch.close(); + } catch (Exception e) { + LOG.warn("Failed to close the stale LeaderLatch for server {}.", serverId, e); + } + startLatch(); + } + @Override public void close() { LOG.info("Closing LeaderLatch for server {}.", serverId); if (closing.compareAndSet(false, true)) { + curatorClient.getConnectionStateListenable().removeListener(sessionLossListener); try { leaderLatch.close(); } catch (Exception e) { @@ -318,66 +391,4 @@ private enum State { STANDBY, CLOSED } - - /** - * A {@link LeaderLatch} that recreates the election parent znode if it has disappeared. - * - *

    The election parent is created as a container znode and is garbage-collected by ZooKeeper - * once the session expires and the ephemeral latch node is deleted. On reconnect, the - * LeaderLatch ignores the NONODE result of listing the election path and stays permanently - * idle. Recreating the parent before the latch's own reconnection handling lets the latch - * re-create its election node and regain leadership without a restart. - * - *

    The recovery piggybacks on {@code handleStateChange} so that it strictly precedes the - * latch's reconnection handling: registering a separate connection-state listener would not - * guarantee this order, as listeners are notified in {@code ConcurrentHashMap} iteration order, - * which does not follow registration order. - */ - private class RecoveringLeaderLatch extends LeaderLatch { - - RecoveringLeaderLatch(CuratorFramework client, String latchPath, String id) { - super(client, latchPath, id); - } - - @Override - protected void handleStateChange(ConnectionState newState) { - if (newState == ConnectionState.RECONNECTED) { - recoverElectionParentIfNeeded(); - } - super.handleStateChange(newState); - } - - /** - * Recreates the election parent znode if it no longer exists. - * - *

    The znode is recreated as PERSISTENT: unlike a container znode, a persistent znode is - * never garbage-collected when empty, so a future session expiration cannot remove the - * election parent again. - */ - private void recoverElectionParentIfNeeded() { - String electionPath = ZkData.CoordinatorElectionZNode.path(); - try { - if (curatorClient.checkExists().forPath(electionPath) == null) { - curatorClient - .create() - .creatingParentsIfNeeded() - .withMode(CreateMode.PERSISTENT) - .forPath(electionPath); - LOG.warn( - "Coordinator server {}: election parent {} was missing, recreate it " - + "to recover the leader election.", - serverId, - electionPath); - } - } catch (Exception e) { - // Ignore the failure and keep the latch's own reconnection handling unchanged: a - // later reconnection event retries the recovery. - LOG.error( - "Coordinator server {}: failed to recover the election parent {}.", - serverId, - electionPath, - e); - } - } - } } diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorHighAvailabilityITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorHighAvailabilityITCase.java index eef65854614..c1be11fcccc 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorHighAvailabilityITCase.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorHighAvailabilityITCase.java @@ -227,28 +227,27 @@ void testLeaderLosesLeadershipAndReElected() throws Exception { } /** - * Regression test for #4362: a single-coordinator cluster must regain leadership without a - * restart after a ZooKeeper session expiration, even when the now-empty election parent node - * has been deleted. + * Regression test: a single-coordinator cluster must regain leadership without a restart after + * a ZooKeeper session expiration, even when the now-empty election parent node has been + * deleted. * *

    The LeaderLatch creates the election parent {@code /coordinators/election} as a container * znode. When the session expires, ZooKeeper deletes the ephemeral latch node, and the empty * container parent can then be garbage-collected by ZooKeeper while the coordinator is still * recovering. This is simulated here by deleting the parent right after the session expiration. - * The coordinator must reconnect, re-create its election node, and become leader again without - * a restart. + * The coordinator must restart the election and become leader again without a restart of the + * process, registering a fresh election node under the recreated parent. */ @Test void testRegainsLeadershipAfterSessionExpirationWithElectionParentDeleted() throws Exception { - // single coordinator: no standby can take over, matching the #4362 scenario + // single coordinator: no standby can take over coordinatorServer1 = new CoordinatorServer(createConfiguration()); coordinatorServer1.start(); waitUntilCoordinatorServerElected(); String electionPath = ZkData.CoordinatorElectionZNode.path(); - assertThat(zookeeperClient.getStat(electionPath)) - .as("election parent should exist while the latch node exists") - .isPresent(); + List electionNodesBefore = zookeeperClient.getChildren(electionPath); + assertThat(electionNodesBefore).hasSize(1); // kill the coordinator's ZK session: ZooKeeper deletes the ephemeral latch node killZkSession(coordinatorServer1); @@ -270,19 +269,65 @@ void testRegainsLeadershipAfterSessionExpirationWithElectionParentDeleted() thro // the coordinator reconnects with a new session and registers itself again waitUntilServerRegistered(coordinatorServer1); - // regression assertion for #4362: leadership must be regained without a restart + // leadership must be regained without a restart waitUntil( () -> coordinatorServer1.getCoordinatorService().isLeader(), Duration.ofSeconds(30), "Coordinator did not regain leadership after session expiration and election " - + "parent deletion (#4362)"); + + "parent deletion"); createGatewayForServer(coordinatorServer1).metadata(new MetadataRequest()).get(); - // the recovered leader must actually participate in the election again: its latch - // node must exist under the election parent + // the recovered leader must actually participate in the election again with a fresh + // election node: the old node is gone with the lost session, and since the election + // parent was deleted, the new node can only be created by the restarted election assertThat(zookeeperClient.getChildren(electionPath)) - .as("recovered leader should have its election latch node") - .isNotEmpty(); + .hasSize(1) + .isNotEqualTo(electionNodesBefore); + } + + /** + * A short ZooKeeper outage that suspends the connection but keeps the session must not restart + * the election: the existing latch revalidates leadership on reconnection with the same + * election node, which still belongs to the alive session. + */ + @Test + void testSuspensionKeepsElectionNodeWithoutRestart() throws Exception { + // a session timeout well above the outage, so the outage suspends the connection but + // keeps the session alive + Configuration conf = createConfiguration(); + conf.set(ConfigOptions.ZOOKEEPER_SESSION_TIMEOUT, Duration.ofSeconds(30)); + coordinatorServer1 = new CoordinatorServer(conf); + coordinatorServer1.start(); + waitUntilCoordinatorServerElected(); + + String electionPath = ZkData.CoordinatorElectionZNode.path(); + List childrenBefore = zookeeperClient.getChildren(electionPath); + assertThat(childrenBefore).hasSize(1); + + // stop the ZK server long enough to suspend the connection, but shorter than the session + // timeout, then restart it + ZOO_KEEPER_EXTENSION_WRAPPER.getCustomExtension().stop(); + try { + Thread.sleep(8000); + } finally { + ZOO_KEEPER_EXTENSION_WRAPPER.getCustomExtension().restart(); + } + + // the existing latch revalidates leadership on reconnection, with the same election node + waitUntil( + () -> coordinatorServer1.getCoordinatorService().isLeader(), + Duration.ofSeconds(30), + "Coordinator did not revalidate leadership after a suspension"); + waitUntil( + () -> { + try { + return zookeeperClient.getChildren(electionPath).equals(childrenBefore); + } catch (Exception e) { + return false; + } + }, + Duration.ofSeconds(30), + "Election node changed during a suspension"); } /** From b7df583121825317e59768cae954d65c168a95ac Mon Sep 17 00:00:00 2001 From: duankaixuan <1417048384@qq.com> Date: Thu, 17 Sep 2026 22:09:27 +0800 Subject: [PATCH 3/3] address feedback --- .../CoordinatorLeaderElection.java | 115 ++++++++---------- .../CoordinatorHighAvailabilityITCase.java | 52 ++++---- 2 files changed, 76 insertions(+), 91 deletions(-) diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorLeaderElection.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorLeaderElection.java index 9c7c9c81808..336cdae4d74 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorLeaderElection.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorLeaderElection.java @@ -55,12 +55,10 @@ *

  • Can be re-elected as leader multiple times * * - *

    When the ZooKeeper session is lost, ZooKeeper deletes the ephemeral election node of this - * coordinator, so the previous election participation becomes stale. After the connection returns, - * the election is restarted with a fresh latch: the fresh latch registers a new election node - * belonging to the current session and determines leadership through the normal election. A short - * connection suspension that keeps the session does not restart the election; the existing latch - * revalidates leadership on reconnection instead. + *

    A lost ZooKeeper session deletes the election node of the latch, so its participation is + * stale. The election is then restarted with a fresh latch that registers a new election node for + * the next session. A connection suspension that keeps the session does not restart the election, + * because the election node of the latch is still valid. * *

    Leadership callbacks and state transitions are serialized by {@code leaderCallbackExecutor}. * The state machine is: @@ -89,17 +87,13 @@ public class CoordinatorLeaderElection implements AutoCloseable { private final String serverId; private final CuratorFramework curatorClient; - /** - * The latch currently participating in the election. Replaced with a fresh latch when the - * ZooKeeper session was lost and the connection returns, see {@link #restartElection()}. - */ + /** The latch participating in the election, replaced when the ZooKeeper session was lost. */ private volatile LeaderLatch leaderLatch; - // Single-threaded executor to run leader init/cleanup callbacks outside Curator's EventThread. // Curator's LeaderLatchListener callbacks run on its internal EventThread; performing // synchronous ZK operations there causes deadlock because ZK response dispatch also // needs that same thread. Serial execution also guarantees that leader initialization, - // cleanup, state transitions, and election restarts never overlap. + // cleanup, and state transitions never overlap. private final ExecutorService leaderCallbackExecutor; private final CompletableFuture closeFuture = new CompletableFuture<>(); private final long closeTimeoutMs; @@ -107,25 +101,17 @@ public class CoordinatorLeaderElection implements AutoCloseable { private final AtomicBoolean closing = new AtomicBoolean(false); private volatile State state = State.INITIAL; - private volatile Runnable initLeaderServices; + private volatile LeaderLatchListener latchListener; private volatile Consumer cleanupLeaderServices; /** - * Marks that the ZooKeeper session was lost; the election restarts on the next reconnection. - */ - private final AtomicBoolean sessionLost = new AtomicBoolean(false); - - /** - * Watches for a lost ZooKeeper session and schedules the election restart for the first - * reconnection after it. A reconnection without a preceding session loss keeps the current - * latch, whose election node still belongs to the alive session. + * Restarts the election when the ZooKeeper session is lost. A suspended connection that keeps + * the session is not handled here: the latch keeps its election node and revalidates leadership + * on reconnection. */ private final ConnectionStateListener sessionLossListener = (client, newState) -> { if (newState == ConnectionState.LOST) { - sessionLost.set(true); - } else if (newState == ConnectionState.RECONNECTED - && sessionLost.compareAndSet(true, false)) { submitLeadershipEvent(this::restartElection); } }; @@ -155,22 +141,32 @@ public CoordinatorLeaderElection(ZooKeeperClient zkClient, String serverId) { * Starts the leader election process asynchronously. * *

    After the first election, the server will continue to participate in future elections. - * When re-elected as leader, the initLeaderServices callback will be invoked again. If the - * ZooKeeper session is lost, the election is restarted with a fresh latch once the connection - * returns. + * When re-elected as leader, the initLeaderServices callback will be invoked again. When the + * ZooKeeper session is lost, the election is restarted with a fresh latch. * * @param initLeaderServices the callback to initialize leader services once elected * @param cleanupLeaderServices the callback to clean up leader services when losing leadership */ public void startElectLeaderAsync( Runnable initLeaderServices, Consumer cleanupLeaderServices) { - this.initLeaderServices = initLeaderServices; this.cleanupLeaderServices = cleanupLeaderServices; + this.latchListener = + new LeaderLatchListener() { + @Override + public void isLeader() { + submitLeadershipEvent(() -> becomeLeader(initLeaderServices)); + } + + @Override + public void notLeader() { + submitLeadershipEvent(CoordinatorLeaderElection.this::becomeStandby); + } + }; curatorClient.getConnectionStateListenable().addListener(sessionLossListener); startLatch(); } - /** Creates a latch, subscribes to its leadership notifications, and starts it. */ + /** Starts a latch that participates in the election on behalf of the current session. */ private void startLatch() { if (closing.get()) { return; @@ -181,50 +177,51 @@ private void startLatch() { curatorClient, ZkData.CoordinatorElectionZNode.path(), String.valueOf(serverId)); - latch.addListener( - new LeaderLatchListener() { - @Override - public void isLeader() { - submitLeadershipEvent(() -> becomeLeader(initLeaderServices)); - } - - @Override - public void notLeader() { - submitLeadershipEvent(CoordinatorLeaderElection.this::becomeStandby); - } - }); - this.leaderLatch = latch; + latch.addListener(latchListener); + leaderLatch = latch; try { + // The latch waits for the connection before it registers its election node, so + // starting it while disconnected does not need a later reconnection event. latch.start(); LOG.info("Coordinator server {} started leader election.", serverId); } catch (Exception e) { LOG.error("Failed to start LeaderLatch for server {}", serverId, e); } + + // A close() running concurrently may not have seen this latch yet. + if (closing.get()) { + closeLatch(latch); + } } /** - * Restarts the election after the ZooKeeper session was lost and the connection has returned. + * Restarts the election with a fresh latch after the ZooKeeper session was lost. * - *

    The session loss deletes the ephemeral election node of this coordinator, so the old latch - * no longer represents a valid participation: its election node is gone and any leadership it - * held was already revoked when the session was lost. The old latch is abandoned and a fresh - * latch is started instead, which registers a new election node for the current session and - * determines leadership through the normal election, including the epoch fencing in the leader - * initialization. Creating the new election node also recreates the election parent if it was - * garbage-collected while empty, so the restart recovers even when the election parent is - * missing. + *

    The lost session deletes the election node of the latch, so its participation is stale. + * Local leadership is revoked before the stale latch is abandoned, because the leadership of a + * lost session must never survive into the election of the next session. The fresh latch + * registers a new election node with parents created as needed, which also recreates an + * election parent that was garbage-collected while empty. */ private void restartElection() { LOG.info( - "Coordinator server {}: ZooKeeper session was lost, restarting the leader " - + "election.", + "Coordinator server {}: ZooKeeper session was lost, restarting leader election.", serverId); + becomeStandby(); + closeLatch(leaderLatch); + startLatch(); + } + + /** Closes a latch that is still participating in the election. */ + private void closeLatch(LeaderLatch latch) { + if (latch == null || latch.getState() != LeaderLatch.State.STARTED) { + return; + } try { - leaderLatch.close(); + latch.close(); } catch (Exception e) { - LOG.warn("Failed to close the stale LeaderLatch for server {}.", serverId, e); + LOG.error("Failed to close LeaderLatch for server {}.", serverId, e); } - startLatch(); } @Override @@ -233,11 +230,7 @@ public void close() { if (closing.compareAndSet(false, true)) { curatorClient.getConnectionStateListenable().removeListener(sessionLossListener); - try { - leaderLatch.close(); - } catch (Exception e) { - LOG.error("Failed to close LeaderLatch for server {}.", serverId, e); - } + closeLatch(leaderLatch); // Events submitted after closing starts are ignored by their executor-side check. // Since the executor is single-threaded, this task runs after all leadership work diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorHighAvailabilityITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorHighAvailabilityITCase.java index c1be11fcccc..09aeb66e0f2 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorHighAvailabilityITCase.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorHighAvailabilityITCase.java @@ -227,16 +227,13 @@ void testLeaderLosesLeadershipAndReElected() throws Exception { } /** - * Regression test: a single-coordinator cluster must regain leadership without a restart after - * a ZooKeeper session expiration, even when the now-empty election parent node has been - * deleted. + * A single-coordinator cluster must regain leadership without a restart of the process after a + * ZooKeeper session expiration, even when the now-empty election parent node has been deleted. * - *

    The LeaderLatch creates the election parent {@code /coordinators/election} as a container - * znode. When the session expires, ZooKeeper deletes the ephemeral latch node, and the empty - * container parent can then be garbage-collected by ZooKeeper while the coordinator is still - * recovering. This is simulated here by deleting the parent right after the session expiration. - * The coordinator must restart the election and become leader again without a restart of the - * process, registering a fresh election node under the recreated parent. + *

    The election parent {@code /coordinators/election} is created as a container znode. When + * the session expires, ZooKeeper deletes the ephemeral election node, and the empty container + * parent can then be garbage-collected by ZooKeeper while the coordinator is still recovering. + * This is simulated here by deleting the parent right after the session expiration. */ @Test void testRegainsLeadershipAfterSessionExpirationWithElectionParentDeleted() throws Exception { @@ -249,10 +246,10 @@ void testRegainsLeadershipAfterSessionExpirationWithElectionParentDeleted() thro List electionNodesBefore = zookeeperClient.getChildren(electionPath); assertThat(electionNodesBefore).hasSize(1); - // kill the coordinator's ZK session: ZooKeeper deletes the ephemeral latch node + // kill the coordinator's ZK session: ZooKeeper deletes the ephemeral election node killZkSession(coordinatorServer1); - // wait until the latch node is gone, then simulate ZooKeeper's container GC by + // wait until the election node is gone, then simulate ZooKeeper's container GC by // deleting the now-empty election parent waitUntil( () -> { @@ -263,13 +260,9 @@ void testRegainsLeadershipAfterSessionExpirationWithElectionParentDeleted() thro } }, Duration.ofSeconds(30), - "latch node was not deleted after session expiration"); + "election node was not deleted after session expiration"); zookeeperClient.deletePath(electionPath); - // the coordinator reconnects with a new session and registers itself again - waitUntilServerRegistered(coordinatorServer1); - - // leadership must be regained without a restart waitUntil( () -> coordinatorServer1.getCoordinatorService().isLeader(), Duration.ofSeconds(30), @@ -277,23 +270,23 @@ void testRegainsLeadershipAfterSessionExpirationWithElectionParentDeleted() thro + "parent deletion"); createGatewayForServer(coordinatorServer1).metadata(new MetadataRequest()).get(); - // the recovered leader must actually participate in the election again with a fresh - // election node: the old node is gone with the lost session, and since the election - // parent was deleted, the new node can only be created by the restarted election + // the restarted election registers a fresh election node: the old node is gone with the + // lost session, and since the parent was deleted, the new node can only come from the + // restarted election assertThat(zookeeperClient.getChildren(electionPath)) .hasSize(1) .isNotEqualTo(electionNodesBefore); } /** - * A short ZooKeeper outage that suspends the connection but keeps the session must not restart - * the election: the existing latch revalidates leadership on reconnection with the same - * election node, which still belongs to the alive session. + * A ZooKeeper outage that suspends the connection but keeps the session must not restart the + * election: the latch keeps its election node, which still belongs to the alive session, and + * revalidates leadership on reconnection. */ @Test void testSuspensionKeepsElectionNodeWithoutRestart() throws Exception { - // a session timeout well above the outage, so the outage suspends the connection but - // keeps the session alive + // a session timeout well above the outage keeps the session alive while the connection + // is suspended Configuration conf = createConfiguration(); conf.set(ConfigOptions.ZOOKEEPER_SESSION_TIMEOUT, Duration.ofSeconds(30)); coordinatorServer1 = new CoordinatorServer(conf); @@ -301,11 +294,9 @@ void testSuspensionKeepsElectionNodeWithoutRestart() throws Exception { waitUntilCoordinatorServerElected(); String electionPath = ZkData.CoordinatorElectionZNode.path(); - List childrenBefore = zookeeperClient.getChildren(electionPath); - assertThat(childrenBefore).hasSize(1); + List electionNodesBefore = zookeeperClient.getChildren(electionPath); + assertThat(electionNodesBefore).hasSize(1); - // stop the ZK server long enough to suspend the connection, but shorter than the session - // timeout, then restart it ZOO_KEEPER_EXTENSION_WRAPPER.getCustomExtension().stop(); try { Thread.sleep(8000); @@ -313,7 +304,6 @@ void testSuspensionKeepsElectionNodeWithoutRestart() throws Exception { ZOO_KEEPER_EXTENSION_WRAPPER.getCustomExtension().restart(); } - // the existing latch revalidates leadership on reconnection, with the same election node waitUntil( () -> coordinatorServer1.getCoordinatorService().isLeader(), Duration.ofSeconds(30), @@ -321,7 +311,9 @@ void testSuspensionKeepsElectionNodeWithoutRestart() throws Exception { waitUntil( () -> { try { - return zookeeperClient.getChildren(electionPath).equals(childrenBefore); + return zookeeperClient + .getChildren(electionPath) + .equals(electionNodesBefore); } catch (Exception e) { return false; }