support coordinator epoch2

Author: zcoo

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

@@ -237,8 +237,6 @@ protected void startCoordinatorLeaderService() throws Exception {
                                     serverMetricGroup));
             rpcServer.start();
 
-            registerCoordinatorLeader();
-
             this.clientMetricGroup = new ClientMetricGroup(metricRegistry, SERVER_NAME);
             this.rpcClient = RpcClient.create(conf, clientMetricGroup, true);
 
@@ -253,6 +251,7 @@ protected void startCoordinatorLeaderService() throws Exception {
                     Executors.newFixedThreadPool(
                             ioExecutorPoolSize, new ExecutorThreadFactory("coordinator-io"));
 
+            registerCoordinatorLeader();
             // start coordinator event processor after we register coordinator leader to zk
             // so that the event processor can get the coordinator leader node from zk during start
             // up.

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

@@ -271,6 +271,7 @@ public void startCoordinatorServer() throws Exception {
         } else {
             // start the existing coordinator server
             coordinatorServer.start();
+            waitUntilCoordinatorServerElected();
             coordinatorServerInfo =
                     new ServerInfo(
                             0,