[server] Use async mode for delete partition assignment in zk

Author: LiebingYu

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

@@ -168,7 +168,7 @@ public CoordinatorEventProcessor(
                                 coordinatorEventManager,
                                 coordinatorContext),
                         zooKeeperClient);
-        this.metadataManager = new MetadataManager(zooKeeperClient, conf);
+        this.metadataManager = new MetadataManager(zooKeeperClient, conf, ioExecutor);
 
         this.tableManager =
                 new TableManager(

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

@@ -183,7 +183,7 @@ protected void startServices() throws Exception {
 
             this.lakeTableTieringManager = new LakeTableTieringManager();
 
-            MetadataManager metadataManager = new MetadataManager(zkClient, conf);
+            MetadataManager metadataManager = new MetadataManager(zkClient, conf, ioExecutor);
             this.coordinatorService =
                     new CoordinatorService(
                             conf,
@@ -321,7 +321,7 @@ private void registerCoordinatorLeader() throws Exception {
     }
 
     private void createDefaultDatabase() {
-        MetadataManager metadataManager = new MetadataManager(zkClient, conf);
+        MetadataManager metadataManager = new MetadataManager(zkClient, conf, ioExecutor);
         List<String> databases = metadataManager.listDatabases();
         if (databases.isEmpty()) {
             metadataManager.createDatabase(DEFAULT_DATABASE, DatabaseDescriptor.EMPTY, true);

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

@@ -47,6 +47,7 @@
 import org.apache.fluss.server.zk.data.TableAssignment;
 import org.apache.fluss.server.zk.data.TableRegistration;
 import org.apache.fluss.shaded.zookeeper3.org.apache.zookeeper.KeeperException;
+import org.apache.fluss.utils.concurrent.Executors;
 import org.apache.fluss.utils.function.RunnableWithException;
 import org.apache.fluss.utils.function.ThrowingRunnable;
 
@@ -60,6 +61,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.Callable;
+import java.util.concurrent.Executor;
 
 import static org.apache.fluss.server.utils.TableDescriptorValidation.validateTableDescriptor;
 
@@ -73,17 +75,26 @@ public class MetadataManager {
     private final int maxPartitionNum;
     private final int maxBucketNum;
 
+    private final Executor ioExecutor;
+
+    public MetadataManager(ZooKeeperClient zookeeperClient, Configuration conf) {
+        this(zookeeperClient, conf, Executors.directExecutor());
+    }
+
     /**
      * Creates a new metadata manager.
      *
      * @param zookeeperClient the zookeeper client
      * @param conf the cluster configuration
+     * @param ioExecutor the io executor
      */
-    public MetadataManager(ZooKeeperClient zookeeperClient, Configuration conf) {
+    public MetadataManager(
+            ZooKeeperClient zookeeperClient, Configuration conf, Executor ioExecutor) {
         this.zookeeperClient = zookeeperClient;
         this.defaultTableLakeOptions = LakeStorageUtils.generateDefaultTableLakeOptions(conf);
         this.maxPartitionNum = conf.get(ConfigOptions.MAX_PARTITION_NUM);
         this.maxBucketNum = conf.get(ConfigOptions.MAX_BUCKET_NUM);
+        this.ioExecutor = ioExecutor;
     }
 
     public void createDatabase(
@@ -232,13 +243,18 @@ public void completeDeleteTable(long tableId) {
                 String.format("Delete tablet assignment meta fail for table %s.", tableId));
     }
 
-    public void completeDeletePartition(long partitionId) {
+    public void completeDeletePartition(TablePartition tablePartition) {
         // final step for delete a partition.
         // 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),
-                String.format("Delete tablet assignment meta fail for partition %s.", partitionId));
+        ioExecutor.execute(
+                () -> {
+                    try {
+                        zookeeperClient.deletePartitionAssignment(tablePartition.getPartitionId());
+                    } catch (Exception e) {
+                        LOG.error("Fail to complete partition {} deletion.", tablePartition, e);
+                    }
+                });
     }
 
     /**

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

@@ -265,11 +265,7 @@ private void completeDeletePartition(TablePartition tablePartition) {
                         tablePartition.getTableId(), tablePartition.getPartitionId());
         replicaStateMachine.handleStateChanges(replicas, ReplicaState.NonExistentReplica);
         deleteRemoteDirectory(tablePartition);
-        try {
-            metadataManager.completeDeletePartition(tablePartition.getPartitionId());
-        } catch (Exception e) {
-            LOG.error("Fail to complete partition {} deletion.", tablePartition, e);
-        }
+        metadataManager.completeDeletePartition(tablePartition);
         coordinatorContext.removePartition(tablePartition);
     }