fix cr

Author: zcoo

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

@@ -112,6 +112,11 @@ public void isLeader() {
                         CompletableFuture<Void> cleanup = pendingCleanup.get();
                         // Run init on a separate thread to avoid deadlock with
                         // Curator's EventThread when performing ZK operations.
+
+                        // Set leader flag before init completes, so when zk found this leader, the
+                        // leader can accept requests
+                        isLeader.set(true);
+
                         leaderCallbackExecutor.execute(
                                 () -> {
                                     // Wait for any pending cleanup to finish first.
@@ -133,17 +138,15 @@ public void isLeader() {
                                         LOG.warn(
                                                 "Coordinator server {} has been fenced and not become leader successfully.",
                                                 serverId);
-                                        throw e;
+                                        notLeader();
                                     } catch (Exception e) {
                                         LOG.error(
                                                 "Failed to initialize leader services for server {}",
                                                 serverId,
                                                 e);
+                                        notLeader();
                                     }
                                 });
-                        // Set leader flag before init completes, so when zk found this leader, the
-                        // leader can accept requests
-                        isLeader.set(true);
                     }
 
                     @Override

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

@@ -348,6 +348,9 @@ protected void cleanupCoordinatorLeader() {
 
             try {
                 // make sure the current coordinator leader node is unregistered.
+                // Different from ZK disconnection,
+                // when we actively release the Leader's election,
+                // we need to manually delete the node
                 unregisterCoordinatorLeader();
             } catch (Throwable t) {
                 LOG.warn("Failed to unregister coordinator leader from Zookeeper", t);

fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java

@@ -200,10 +200,6 @@ public void afterEach(ExtensionContext extensionContext) throws Exception {
             }
         }
         CompletableFuture.allOf(dropFutures.toArray(new CompletableFuture[0])).join();
-
-        for (TabletServer tabletServer : tabletServers.values()) {
-            tabletServer.getReplicaManager().resetCoordinatorEpoch();
-        }
     }
 
     public void start() throws Exception {

fluss-server/src/test/java/org/apache/fluss/server/zk/ZooKeeperClientTest.java

@@ -55,6 +55,7 @@
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -88,15 +89,19 @@ class ZooKeeperClientTest {
     private static ZkEpoch zkEpoch;
 
     @BeforeAll
-    static void beforeAll() throws Exception {
+    static void beforeAll() {
         zookeeperClient =
                 ZOO_KEEPER_EXTENSION_WRAPPER
                         .getCustomExtension()
                         .getZooKeeperClient(NOPErrorHandler.INSTANCE);
-        zkEpoch = zookeeperClient.fenceBecomeCoordinatorLeader("1");
         remoteDataDir = zookeeperClient.getDefaultRemoteDataDir();
     }
 
+    @BeforeEach
+    void beforeEach() throws Exception {
+        zkEpoch = zookeeperClient.fenceBecomeCoordinatorLeader("tmp");
+    }
+
     @AfterEach
     void afterEach() {
         ZOO_KEEPER_EXTENSION_WRAPPER.getCustomExtension().cleanupRoot();