[bug][coordinator/kv] delete remote kv dir for table on coordinator server

fluss-common/src/main/java/com/alibaba/fluss/utils/FlussPaths.java

@@ -21,6 +21,7 @@
 import com.alibaba.fluss.fs.FsPath;
 import com.alibaba.fluss.metadata.PhysicalTablePath;
 import com.alibaba.fluss.metadata.TableBucket;
+import com.alibaba.fluss.metadata.TablePath;
 import com.alibaba.fluss.remote.RemoteLogSegment;
 import com.alibaba.fluss.utils.types.Tuple2;
 
@@ -586,6 +587,35 @@ public static FsPath remoteKvTabletDir(
         return new FsPath(remoteTableDir, String.valueOf(tableBucket.getBucket()));
     }
 
+    /**
+     * Returns the remote directory path for storing kv snapshot files or log segments file of
+     * table.
+     *
+     * <p>The path contract:
+     *
+     * <pre>
+     * Remote kv table dir
+     * {$remote.data.dir}/kv/{databaseName}/{tableName}-{tableId}
+     *
+     * Remote log table dir.
+     * {$remote.data.dir}/log/{databaseName}/{tableName}-{tableId}
+     *
+     * </pre>
+     *
+     * @param remoteKvOrLogBaseDir - the remote kv snapshots or log segments root dir of specific
+     *     table.
+     * @param tablePath - table path.
+     * @param tableId - table id.
+     */
+    public static FsPath remoteTableDir(
+            FsPath remoteKvOrLogBaseDir, TablePath tablePath, long tableId) {
+        return new FsPath(
+                remoteKvOrLogBaseDir,
+                String.format(
+                        "%s/%s-%s",
+                        tablePath.getDatabaseName(), tablePath.getTableName(), tableId));
+    }
+
     /**
      * Returns the remote directory path for storing kv snapshot exclusive files (manifest and
      * CURRENT files).

fluss-server/src/main/java/com/alibaba/fluss/server/coordinator/CoordinatorEventProcessor.java

@@ -141,13 +141,15 @@ public class CoordinatorEventProcessor implements EventProcessor {
 
     public CoordinatorEventProcessor(
             ZooKeeperClient zooKeeperClient,
+            RemoteStorageHandler remoteStorageHandler,
             ServerMetadataCache serverMetadataCache,
             CoordinatorChannelManager coordinatorChannelManager,
             CompletedSnapshotStoreManager completedSnapshotStoreManager,
             AutoPartitionManager autoPartitionManager,
             CoordinatorMetricGroup coordinatorMetricGroup) {
         this(
                 zooKeeperClient,
+                remoteStorageHandler,
                 serverMetadataCache,
                 coordinatorChannelManager,
                 new CoordinatorContext(),
@@ -158,6 +160,7 @@ public CoordinatorEventProcessor(
 
     public CoordinatorEventProcessor(
             ZooKeeperClient zooKeeperClient,
+            RemoteStorageHandler remoteStorageHandler,
             ServerMetadataCache serverMetadataCache,
             CoordinatorChannelManager coordinatorChannelManager,
             CoordinatorContext coordinatorContext,
@@ -186,7 +189,8 @@ public CoordinatorEventProcessor(
                         metaDataManager,
                         coordinatorContext,
                         replicaStateMachine,
-                        tableBucketStateMachine);
+                        tableBucketStateMachine,
+                        remoteStorageHandler);
         this.tableChangeWatcher = new TableChangeWatcher(zooKeeperClient, coordinatorEventManager);
         this.tabletServerChangeWatcher =
                 new TabletServerChangeWatcher(zooKeeperClient, coordinatorEventManager);

fluss-server/src/main/java/com/alibaba/fluss/server/coordinator/CoordinatorServer.java

@@ -180,9 +180,11 @@ protected void startServices() throws Exception {
             // up.
             // in HA for coordinator server, the processor also need to know the leader node during
             // start up
+            RemoteStorageHandler remoteStorageHandler = new RemoteStorageHandler(conf);
             this.coordinatorEventProcessor =
                     new CoordinatorEventProcessor(
                             zkClient,
+                            remoteStorageHandler,
                             metadataCache,
                             coordinatorChannelManager,
                             bucketSnapshotManager,

fluss-server/src/main/java/com/alibaba/fluss/server/coordinator/RemoteStorageHandler.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright (c) 2024 Alibaba Group Holding Ltd.
+ *
+ * Licensed 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 com.alibaba.fluss.server.coordinator;
+
+import com.alibaba.fluss.config.Configuration;
+import com.alibaba.fluss.fs.FileSystem;
+import com.alibaba.fluss.fs.FsPath;
+import com.alibaba.fluss.metadata.TablePath;
+import com.alibaba.fluss.utils.FlussPaths;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+public class RemoteStorageHandler {
+
+    private static final Logger LOG = LoggerFactory.getLogger(RemoteStorageHandler.class);
+
+    private final FsPath remoteKvDir;
+    private final FileSystem remoteFileSystem;
+
+    public RemoteStorageHandler(Configuration configuration) throws IOException {
+        this.remoteKvDir = FlussPaths.remoteKvDir(configuration);
+        this.remoteFileSystem = remoteKvDir.getFileSystem();
+    }
+
+    public void createTableKvDir(TablePath tablePath, long tableId) throws IOException {
+        remoteFileSystem.mkdirs(tableKvDir(tablePath, tableId));
+    }
+
+    public void deleteTable(TablePath tablePath, long tableId) {
+        deleteDir(tableKvDir(tablePath, tableId));
+        // todo delete log segments file dirs of table.
+    }
+
+    public boolean isTableKvDirExists(TablePath tablePath, long tableId) throws IOException {
+        return isDirExists(tableKvDir(tablePath, tableId));
+    }
+
+    private boolean isDirExists(FsPath fsPath) throws IOException {
+        return remoteFileSystem.exists(fsPath);
+    }
+
+    private void deleteDir(FsPath fsPath) {
+        try {
+            if (isDirExists(fsPath)) {
+                remoteFileSystem.delete(fsPath, true);
+                LOG.info("Delete table's remote dir {} success.", fsPath);
+            }
+        } catch (IOException e) {
+            LOG.error("Delete table's remote dir {} failed.", fsPath, e);
+        }
+    }
+
+    private FsPath tableKvDir(TablePath tablePath, long tableId) {
+        return FlussPaths.remoteTableDir(remoteKvDir, tablePath, tableId);
+    }
+}

fluss-server/src/main/java/com/alibaba/fluss/server/coordinator/TableManager.java

@@ -41,6 +41,7 @@ public class TableManager {
     private static final Logger LOG = LoggerFactory.getLogger(TableManager.class);
 
     private final MetaDataManager metaDataManager;
+    private final RemoteStorageHandler remoteStorageHandler;
     private final CoordinatorContext coordinatorContext;
     private final ReplicaStateMachine replicaStateMachine;
     private final TableBucketStateMachine tableBucketStateMachine;
@@ -49,8 +50,10 @@ public TableManager(
             MetaDataManager metaDataManager,
             CoordinatorContext coordinatorContext,
             ReplicaStateMachine replicaStateMachine,
-            TableBucketStateMachine tableBucketStateMachine) {
+            TableBucketStateMachine tableBucketStateMachine,
+            RemoteStorageHandler remoteStorageHandler) {
         this.metaDataManager = metaDataManager;
+        this.remoteStorageHandler = remoteStorageHandler;
         this.coordinatorContext = coordinatorContext;
         this.replicaStateMachine = replicaStateMachine;
         this.tableBucketStateMachine = tableBucketStateMachine;
@@ -244,6 +247,8 @@ private void completeDeleteTable(long tableId) {
         replicaStateMachine.handleStateChanges(replicas, ReplicaState.NonExistentReplica);
         try {
             metaDataManager.completeDeleteTable(tableId);
+            TablePath tablePath = coordinatorContext.getTablePathById(tableId);
+            remoteStorageHandler.deleteTable(tablePath, tableId);
         } catch (Exception e) {
             LOG.error("Fail to complete table deletion for table {}.", tableId, e);
         }

fluss-server/src/test/java/com/alibaba/fluss/server/coordinator/CoordinatorEventProcessorTest.java

@@ -17,6 +17,7 @@
 package com.alibaba.fluss.server.coordinator;
 
 import com.alibaba.fluss.cluster.ServerNode;
+import com.alibaba.fluss.config.ConfigOptions;
 import com.alibaba.fluss.config.Configuration;
 import com.alibaba.fluss.exception.FencedLeaderEpochException;
 import com.alibaba.fluss.exception.InvalidCoordinatorException;
@@ -62,6 +63,7 @@
 import org.junit.jupiter.api.extension.RegisterExtension;
 import org.junit.jupiter.api.io.TempDir;
 
+import java.io.IOException;
 import java.nio.file.Path;
 import java.time.Duration;
 import java.util.Arrays;
@@ -115,6 +117,8 @@ class CoordinatorEventProcessorTest {
     private CompletedSnapshotStoreManager completedSnapshotStoreManager;
     private AutoPartitionManager autoPartitionManager;
 
+    private RemoteStorageHandler remoteStorageHandler;
+
     @BeforeAll
     static void baseBeforeAll() throws Exception {
         zookeeperClient =
@@ -130,16 +134,20 @@ static void baseBeforeAll() throws Exception {
     }
 
     @BeforeEach
-    void beforeEach() {
+    void beforeEach() throws IOException {
         serverMetadataCache = new ServerMetadataCacheImpl();
         // set a test channel manager for the context
         testCoordinatorChannelManager = new TestCoordinatorChannelManager();
         completedSnapshotStoreManager = new CompletedSnapshotStoreManager(1, 1, zookeeperClient);
         autoPartitionManager =
                 new AutoPartitionManager(serverMetadataCache, zookeeperClient, new Configuration());
+        Configuration conf = new Configuration();
+        conf.setString(ConfigOptions.REMOTE_DATA_DIR, "/tmp/fluss/remote-data");
+        remoteStorageHandler = new RemoteStorageHandler(conf);
         eventProcessor =
                 new CoordinatorEventProcessor(
                         zookeeperClient,
+                        remoteStorageHandler,
                         serverMetadataCache,
                         testCoordinatorChannelManager,
                         completedSnapshotStoreManager,
@@ -199,6 +207,7 @@ void testCreateAndDropTable() throws Exception {
         eventProcessor =
                 new CoordinatorEventProcessor(
                         zookeeperClient,
+                        remoteStorageHandler,
                         serverMetadataCache,
                         testCoordinatorChannelManager,
                         completedSnapshotStoreManager,
@@ -389,6 +398,7 @@ void testServerBecomeOnlineAndOfflineLine() throws Exception {
         eventProcessor =
                 new CoordinatorEventProcessor(
                         zookeeperClient,
+                        remoteStorageHandler,
                         serverMetadataCache,
                         testCoordinatorChannelManager,
                         completedSnapshotStoreManager,
@@ -437,6 +447,7 @@ void testRestartTriggerReplicaToOffline() throws Exception {
         eventProcessor =
                 new CoordinatorEventProcessor(
                         zookeeperClient,
+                        remoteStorageHandler,
                         serverMetadataCache,
                         testCoordinatorChannelManager,
                         completedSnapshotStoreManager,
@@ -623,6 +634,7 @@ void testCreateAndDropPartition() throws Exception {
         eventProcessor =
                 new CoordinatorEventProcessor(
                         zookeeperClient,
+                        remoteStorageHandler,
                         serverMetadataCache,
                         testCoordinatorChannelManager,
                         completedSnapshotStoreManager,

fluss-server/src/test/java/com/alibaba/fluss/server/coordinator/RemoteStorageHandlerTest.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright (c) 2024 Alibaba Group Holding Ltd.
+ *
+ * Licensed 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 com.alibaba.fluss.server.coordinator;
+
+import com.alibaba.fluss.config.ConfigOptions;
+import com.alibaba.fluss.config.Configuration;
+import com.alibaba.fluss.metadata.TablePath;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.nio.file.Path;
+
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+
+public class RemoteStorageHandlerTest {
+
+    private static @TempDir Path tempDir;
+
+    private static RemoteStorageHandler remoteStorageHandler;
+
+    @BeforeAll
+    static void before() throws IOException {
+        Configuration conf = new Configuration();
+        conf.setString(ConfigOptions.REMOTE_DATA_DIR, tempDir.toString());
+        remoteStorageHandler = new RemoteStorageHandler(conf);
+    }
+
+    @Test
+    void testDeleteRemoteByTablePath() throws IOException {
+        TablePath tablePath = TablePath.of("ods", "tbl");
+        remoteStorageHandler.createTableKvDir(tablePath, 0);
+        assertThat(remoteStorageHandler.isTableKvDirExists(tablePath, 0)).isTrue();
+        remoteStorageHandler.deleteTable(tablePath, 0);
+        assertThat(remoteStorageHandler.isTableKvDirExists(tablePath, 0)).isFalse();
+    }
+}