support coordinator epoch2

Author: zcoo

fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java

@@ -1675,20 +1675,21 @@ public void createRecursiveWithEpochCheck(
                 throw new IllegalArgumentException("Invalid path: " + path);
             } else if (indexOfLastSlash == 0) {
                 // root path can be directly create without fence
-                zkClient.create()
-                        .creatingParentsIfNeeded()
-                        .withMode(CreateMode.PERSISTENT)
-                        .forPath(path);
-            }
-            // If indexOfLastSlash is 0, it means the parent is root "/" which should already exist
-            // We should not try to create it, just retry creating the current path
-            if (indexOfLastSlash > 0) {
+                try {
+                    zkClient.create()
+                            .creatingParentsIfNeeded()
+                            .withMode(CreateMode.PERSISTENT)
+                            .forPath(path);
+                } catch (KeeperException.NodeExistsException ignored) {
+                }
+            } else {
+                // indexOfLastSlash > 0
                 String parentPath = path.substring(0, indexOfLastSlash);
                 createRecursiveWithEpochCheck(
                         parentPath, null, expectedZkVersion, throwIfPathExists);
+                // After creating parent (or if parent is root), retry creating the original path
+                zkClient.transaction().forOperations(ops);
             }
-            // After creating parent (or if parent is root), retry creating the original path
-            zkClient.transaction().forOperations(ops);
         }
     }
 

fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorServerElectionTest.java

@@ -83,18 +83,18 @@ void testCoordinatorServerElection() throws Exception {
         CoordinatorAddress firstLeaderAddress = zookeeperClient.getCoordinatorLeaderAddress().get();
 
         // Find the leader and try to restart it.
-        CoordinatorServer elected = null;
+        CoordinatorServer firstLeader = null;
         for (CoordinatorServer coordinatorServer : coordinatorServerList) {
             if (coordinatorServer.getServerId() == firstLeaderAddress.getId()) {
-                elected = coordinatorServer;
+                firstLeader = coordinatorServer;
                 break;
             }
         }
-        assertThat(elected).isNotNull();
+        assertThat(firstLeader).isNotNull();
         assertThat(zookeeperClient.getCurrentEpoch().getCoordinatorEpoch())
                 .isEqualTo(CoordinatorContext.INITIAL_COORDINATOR_EPOCH);
-        elected.close();
-        elected.start();
+        firstLeader.close();
+        firstLeader.start();
 
         // Then we should get another Coordinator server leader elected
         waitUntilCoordinatorServerReelected(firstLeaderAddress);
@@ -104,12 +104,25 @@ void testCoordinatorServerElection() throws Exception {
         assertThat(zookeeperClient.getCurrentEpoch().getCoordinatorEpoch())
                 .isEqualTo(CoordinatorContext.INITIAL_COORDINATOR_EPOCH + 1);
 
-        // kill other 2 coordinator servers except the first one
+        CoordinatorServer secondLeader = null;
+        for (CoordinatorServer coordinatorServer : coordinatorServerList) {
+            if (coordinatorServer.getServerId() == secondLeaderAddress.getId()) {
+                secondLeader = coordinatorServer;
+                break;
+            }
+        }
+        CoordinatorServer nonLeader = null;
         for (CoordinatorServer coordinatorServer : coordinatorServerList) {
-            if (coordinatorServer.getServerId() != firstLeaderAddress.getId()) {
-                coordinatorServer.close();
+            if (coordinatorServer.getServerId() != firstLeaderAddress.getId()
+                    && coordinatorServer.getServerId() != secondLeaderAddress.getId()) {
+                nonLeader = coordinatorServer;
+                break;
             }
         }
+        // kill other 2 coordinator servers except the first one
+        nonLeader.close();
+        secondLeader.close();
+
         // the origin coordinator server should become leader again
         waitUntilCoordinatorServerReelected(secondLeaderAddress);
         CoordinatorAddress thirdLeaderAddress = zookeeperClient.getCoordinatorLeaderAddress().get();