[server] Coordinator Server Supports High-Available

Author: zcoo

fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java

@@ -392,6 +392,12 @@ public class ConfigOptions {
                                     + " (“50100,50101”), ranges (“50100-50200”) or a combination of both."
                                     + "This option is deprecated. Please use bind.listeners instead, which provides a more flexible configuration for multiple ports");
 
+    public static final ConfigOption<Integer> COORDINATOR_ID =
+            key("coordinator.id")
+                    .intType()
+                    .noDefaultValue()
+                    .withDescription("The id for the coordinator server.");
+
     /**
      * @deprecated This option is deprecated. Please use {@link ConfigOptions#SERVER_IO_POOL_SIZE}
      *     instead.

fluss-common/src/main/java/org/apache/fluss/exception/CoordinatorEpochFencedException.java

@@ -0,0 +1,26 @@
+/*
+ * 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.fluss.exception;
+
+/** Exception thrown when the Coordinator leader epoch is fenced. */
+public class CoordinatorEpochFencedException extends RuntimeException {
+    public CoordinatorEpochFencedException(String message) {
+        super(message);
+    }
+}

fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorContext.java

@@ -55,6 +55,7 @@ public class CoordinatorContext {
     private static final Logger LOG = LoggerFactory.getLogger(CoordinatorContext.class);
 
     public static final int INITIAL_COORDINATOR_EPOCH = 0;
+    public static final int INITIAL_COORDINATOR_EPOCH_ZK_VERSION = 0;
 
     // for simplicity, we just use retry time, may consider make it a configurable value
     // and use combine retry times and retry delay
@@ -109,13 +110,23 @@ public class CoordinatorContext {
 
     private ServerInfo coordinatorServerInfo = null;
     private int coordinatorEpoch = INITIAL_COORDINATOR_EPOCH;
+    private int coordinatorEpochZkVersion = INITIAL_COORDINATOR_EPOCH_ZK_VERSION;
 
     public CoordinatorContext() {}
 
     public int getCoordinatorEpoch() {
         return coordinatorEpoch;
     }
 
+    public int getCoordinatorEpochZkVersion() {
+        return coordinatorEpochZkVersion;
+    }
+
+    public void setCoordinatorEpochAndZkVersion(int newEpoch, int newZkVersion) {
+        this.coordinatorEpoch = newEpoch;
+        this.coordinatorEpochZkVersion = newZkVersion;
+    }
+
     public Set<String> getLiveCoordinatorServers() {
         return liveCoordinatorServers;
     }
@@ -709,11 +720,12 @@ private void clearTablesState() {
     public void resetContext() {
         tablesToBeDeleted.clear();
         coordinatorEpoch = 0;
+        coordinatorEpochZkVersion = INITIAL_COORDINATOR_EPOCH_ZK_VERSION;
         clearTablesState();
         liveTabletServers.clear();
+        liveCoordinatorServers.clear();
         shuttingDownTabletServers.clear();
         serverTags.clear();
-        liveCoordinatorServers.clear();
     }
 
     public int getTotalPartitionCount() {

fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java

@@ -1602,7 +1602,10 @@ private void updateReplicaAssignmentForBucket(
             tableAssignment.forEach(
                     (bucket, replicas) ->
                             newTableAssignment.put(bucket, new BucketAssignment(replicas)));
-            zooKeeperClient.updateTableAssignment(tableId, new TableAssignment(newTableAssignment));
+            zooKeeperClient.updateTableAssignment(
+                    tableId,
+                    new TableAssignment(newTableAssignment),
+                    coordinatorContext.getCoordinatorEpochZkVersion());
         } else {
             Map<Integer, List<Integer>> partitionAssignment =
                     coordinatorContext.getPartitionAssignment(
@@ -1659,7 +1662,8 @@ private List<AdjustIsrResultForBucket> tryProcessAdjustIsr(
         }
 
         try {
-            zooKeeperClient.batchUpdateLeaderAndIsr(newLeaderAndIsrList);
+            zooKeeperClient.batchUpdateLeaderAndIsr(
+                    newLeaderAndIsrList, coordinatorContext.getCoordinatorEpochZkVersion());
             newLeaderAndIsrList.forEach(
                     (tableBucket, newLeaderAndIsr) ->
                             result.add(new AdjustIsrResultForBucket(tableBucket, newLeaderAndIsr)));
@@ -1670,7 +1674,10 @@ private List<AdjustIsrResultForBucket> tryProcessAdjustIsr(
                 TableBucket tableBucket = entry.getKey();
                 LeaderAndIsr newLeaderAndIsr = entry.getValue();
                 try {
-                    zooKeeperClient.updateLeaderAndIsr(tableBucket, newLeaderAndIsr);
+                    zooKeeperClient.updateLeaderAndIsr(
+                            tableBucket,
+                            newLeaderAndIsr,
+                            coordinatorContext.getCoordinatorEpochZkVersion());
                 } catch (Exception e) {
                     LOG.error("Error when register leader and isr.", e);
                     result.add(
@@ -2196,7 +2203,8 @@ private void updateBucketEpochAndSendRequest(TableBucket tableBucket, List<Integ
         LeaderAndIsr newLeaderAndIsr = leaderAndIsr.newLeaderAndIsr(leaderAndIsr.isr());
 
         coordinatorContext.putBucketLeaderAndIsr(tableBucket, newLeaderAndIsr);
-        zooKeeperClient.updateLeaderAndIsr(tableBucket, newLeaderAndIsr);
+        zooKeeperClient.updateLeaderAndIsr(
+                tableBucket, newLeaderAndIsr, coordinatorContext.getCoordinatorEpochZkVersion());
 
         coordinatorRequestBatch.newBatch();
         coordinatorRequestBatch.addNotifyLeaderRequestForTabletServers(

fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorLeaderElection.java

@@ -18,6 +18,8 @@
 
 package org.apache.fluss.server.coordinator;
 
+import org.apache.fluss.exception.CoordinatorEpochFencedException;
+import org.apache.fluss.server.zk.ZkEpoch;
 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.recipes.leader.LeaderLatch;
@@ -26,6 +28,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -51,6 +54,8 @@ public class CoordinatorLeaderElection implements AutoCloseable {
     private static final Logger LOG = LoggerFactory.getLogger(CoordinatorLeaderElection.class);
 
     private final String serverId;
+    private final ZooKeeperClient zkClient;
+    private final CoordinatorContext coordinatorContext;
     private final LeaderLatch leaderLatch;
     private final AtomicBoolean isLeader = new AtomicBoolean(false);
     // Cached thread pool to run leader init/cleanup callbacks outside Curator's EventThread.
@@ -63,8 +68,11 @@ public class CoordinatorLeaderElection implements AutoCloseable {
     private final AtomicReference<CompletableFuture<Void>> pendingCleanup =
             new AtomicReference<>(CompletableFuture.completedFuture(null));
 
-    public CoordinatorLeaderElection(ZooKeeperClient zkClient, String serverId) {
+    public CoordinatorLeaderElection(
+            ZooKeeperClient zkClient, String serverId, CoordinatorContext coordinatorContext) {
         this.serverId = serverId;
+        this.zkClient = zkClient;
+        this.coordinatorContext = coordinatorContext;
         this.leaderLatch =
                 new LeaderLatch(
                         zkClient.getCuratorClient(),
@@ -122,7 +130,25 @@ public void isLeader() {
                                                 e);
                                     }
                                     try {
+                                        // to avoid split-brain
+                                        Optional<ZkEpoch> optionalEpoch =
+                                                zkClient.fenceBecomeCoordinatorLeader(serverId);
+                                        optionalEpoch.ifPresent(
+                                                integer ->
+                                                        coordinatorContext
+                                                                .setCoordinatorEpochAndZkVersion(
+                                                                        optionalEpoch
+                                                                                .get()
+                                                                                .getCoordinatorEpoch(),
+                                                                        optionalEpoch
+                                                                                .get()
+                                                                                .getCoordinatorEpochZkVersion()));
                                         initLeaderServices.run();
+                                    } catch (CoordinatorEpochFencedException e) {
+                                        LOG.warn(
+                                                "Coordinator server {} has been fenced and not become leader successfully.",
+                                                serverId);
+                                        throw e;
                                     } catch (Exception e) {
                                         LOG.error(
                                                 "Failed to initialize leader services for server {}",

fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorRequestBatch.java

@@ -683,6 +683,7 @@ private UpdateMetadataRequest buildUpdateMetadataRequest() {
         // tablet servers.
         return makeUpdateMetadataRequest(
                 coordinatorContext.getCoordinatorServerInfo(),
+                coordinatorContext.getCoordinatorEpoch(),
                 new HashSet<>(coordinatorContext.getLiveTabletServers().values()),
                 tableMetadataList,
                 partitionMetadataList);

fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java

@@ -59,7 +59,6 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
-import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -220,9 +219,11 @@ protected void initCoordinatorStandby() throws Exception {
                             serverId);
 
             this.zkClient = ZooKeeperUtils.startZookeeperClient(conf, this);
+            this.coordinatorContext = new CoordinatorContext();
 
             // CoordinatorLeaderElection must be created after zkClient is initialized.
-            this.coordinatorLeaderElection = new CoordinatorLeaderElection(zkClient, serverId);
+            this.coordinatorLeaderElection =
+                    new CoordinatorLeaderElection(zkClient, serverId, coordinatorContext);
 
             this.lakeCatalogDynamicLoader = new LakeCatalogDynamicLoader(conf, pluginManager, true);
             this.dynamicConfigManager = new DynamicConfigManager(zkClient, conf, true);
@@ -232,7 +233,6 @@ protected void initCoordinatorStandby() throws Exception {
 
             dynamicConfigManager.startup();
 
-            this.coordinatorContext = new CoordinatorContext();
             this.metadataCache = new CoordinatorMetadataCache();
 
             this.authorizer = AuthorizerLoader.createAuthorizer(conf, zkClient, pluginManager);

fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java

@@ -54,6 +54,7 @@
 import org.apache.fluss.server.zk.data.PartitionRegistration;
 import org.apache.fluss.server.zk.data.TableAssignment;
 import org.apache.fluss.server.zk.data.TableRegistration;
+import org.apache.fluss.server.zk.data.ZkVersion;
 import org.apache.fluss.shaded.zookeeper3.org.apache.zookeeper.KeeperException;
 import org.apache.fluss.utils.function.RunnableWithException;
 import org.apache.fluss.utils.function.ThrowingRunnable;
@@ -337,7 +338,9 @@ public void completeDeleteTable(long tableId) {
         // delete bucket assignments node, which will also delete the bucket state node,
         // so that all the zk nodes related to this table are deleted.
         rethrowIfIsNotNoNodeException(
-                () -> zookeeperClient.deleteTableAssignment(tableId),
+                () ->
+                        zookeeperClient.deleteTableAssignment(
+                                tableId, ZkVersion.MATCH_ANY_VERSION.getVersion()),
                 String.format("Delete tablet assignment meta fail for table %s.", tableId));
     }
 
@@ -346,7 +349,9 @@ public void completeDeletePartition(long partitionId) {
         // delete partition assignments node, which will also delete the bucket state node,
         // so that all the zk nodes related to this partition are deleted.
         rethrowIfIsNotNoNodeException(
-                () -> zookeeperClient.deletePartitionAssignment(partitionId),
+                () ->
+                        zookeeperClient.deletePartitionAssignment(
+                                partitionId, ZkVersion.MATCH_ANY_VERSION.getVersion()),
                 String.format("Delete tablet assignment meta fail for partition %s.", partitionId));
     }
 
@@ -399,7 +404,8 @@ public long createTable(
                     long tableId = zookeeperClient.getTableIdAndIncrement();
                     if (tableAssignment != null) {
                         // register table assignment
-                        zookeeperClient.registerTableAssignment(tableId, tableAssignment);
+                        zookeeperClient.registerTableAssignment(
+                                tableId, tableAssignment, ZkVersion.MATCH_ANY_VERSION.getVersion());
                     }
                     // register the table
                     zookeeperClient.registerTable(

fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/ReplicaStateMachine.java

@@ -487,7 +487,8 @@ private Map<TableBucketReplica, LeaderAndIsr> doRemoveReplicaFromIsr(
             toUpdateLeaderAndIsrList.put(tableBucket, adjustLeaderAndIsr);
         }
         try {
-            zooKeeperClient.batchUpdateLeaderAndIsr(toUpdateLeaderAndIsrList);
+            zooKeeperClient.batchUpdateLeaderAndIsr(
+                    toUpdateLeaderAndIsrList, coordinatorContext.getCoordinatorEpochZkVersion());
             toUpdateLeaderAndIsrList.forEach(coordinatorContext::putBucketLeaderAndIsr);
             return adjustedLeaderAndIsr;
         } catch (Exception e) {

fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachine.java

@@ -311,7 +311,10 @@ private Optional<ElectionResult> initLeaderForTableBuckets(
             ElectionResult electionResult = optionalElectionResult.get();
             LeaderAndIsr leaderAndIsr = electionResult.leaderAndIsr;
             try {
-                zooKeeperClient.registerLeaderAndIsr(tableBucket, leaderAndIsr);
+                zooKeeperClient.registerLeaderAndIsr(
+                        tableBucket,
+                        leaderAndIsr,
+                        coordinatorContext.getCoordinatorEpochZkVersion());
             } catch (Exception e) {
                 LOG.error(
                         "Fail to create state node for table bucket {} in zookeeper.",
@@ -379,7 +382,8 @@ public void batchHandleOnlineChangeAndInitLeader(Set<TableBucket> tableBuckets)
         if (!tableBucketLeadAndIsrInfos.isEmpty()) {
             try {
                 zooKeeperClient.batchRegisterLeaderAndIsrForTablePartition(
-                        tableBucketLeadAndIsrInfos);
+                        tableBucketLeadAndIsrInfos,
+                        coordinatorContext.getCoordinatorEpochZkVersion());
                 registerSuccessList.addAll(tableBucketLeadAndIsrInfos);
             } catch (Exception e) {
                 LOG.error(
@@ -454,7 +458,10 @@ private List<RegisterTableBucketLeadAndIsrInfo> tryRegisterLeaderAndIsrOneByOne(
         List<RegisterTableBucketLeadAndIsrInfo> registerSuccessList = new ArrayList<>();
         for (RegisterTableBucketLeadAndIsrInfo info : registerList) {
             try {
-                zooKeeperClient.registerLeaderAndIsr(info.getTableBucket(), info.getLeaderAndIsr());
+                zooKeeperClient.registerLeaderAndIsr(
+                        info.getTableBucket(),
+                        info.getLeaderAndIsr(),
+                        coordinatorContext.getCoordinatorEpochZkVersion());
                 registerSuccessList.add(info);
             } catch (Exception e) {
                 LOG.error(
@@ -496,7 +503,10 @@ private Optional<ElectionResult> electNewLeaderForTableBuckets(
         }
         ElectionResult electionResult = optionalElectionResult.get();
         try {
-            zooKeeperClient.updateLeaderAndIsr(tableBucket, electionResult.leaderAndIsr);
+            zooKeeperClient.updateLeaderAndIsr(
+                    tableBucket,
+                    electionResult.leaderAndIsr,
+                    coordinatorContext.getCoordinatorEpochZkVersion());
         } catch (Exception e) {
             LOG.error(
                     "Fail to update bucket LeaderAndIsr for table bucket {}.",