[server] Handle uncaught exceptions in TabletServer during log recovery from residual data corresponding to already dropped tables (#1487)

Author: platinumhamburg

fluss-server/src/main/java/org/apache/fluss/server/TabletManagerBase.java

@@ -20,6 +20,7 @@
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.exception.KvStorageException;
 import org.apache.fluss.exception.LogStorageException;
+import org.apache.fluss.exception.SchemaNotExistException;
 import org.apache.fluss.metadata.PhysicalTablePath;
 import org.apache.fluss.metadata.SchemaInfo;
 import org.apache.fluss.metadata.TableBucket;
@@ -203,7 +204,7 @@ public static TableInfo getTableInfo(ZooKeeperClient zkClient, TablePath tablePa
         Optional<SchemaInfo> schemaInfoOpt = zkClient.getSchemaById(tablePath, schemaId);
         SchemaInfo schemaInfo;
         if (!schemaInfoOpt.isPresent()) {
-            throw new LogStorageException(
+            throw new SchemaNotExistException(
                     String.format(
                             "Failed to load table '%s': Table schema not found in zookeeper metadata.",
                             tablePath));

fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java

@@ -22,6 +22,7 @@
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.exception.FlussRuntimeException;
 import org.apache.fluss.exception.LogStorageException;
+import org.apache.fluss.exception.SchemaNotExistException;
 import org.apache.fluss.metadata.LogFormat;
 import org.apache.fluss.metadata.PhysicalTablePath;
 import org.apache.fluss.metadata.TableBucket;
@@ -189,24 +190,8 @@ private void loadLogs() {
             final boolean cleanShutdown = isCleanShutdown;
             // set runnable job.
             Runnable[] jobsForDir =
-                    tabletsToLoad.stream()
-                            .map(
-                                    tabletDir ->
-                                            (Runnable)
-                                                    () -> {
-                                                        LOG.debug("Loading log {}", tabletDir);
-                                                        try {
-                                                            loadLog(
-                                                                    tabletDir,
-                                                                    cleanShutdown,
-                                                                    finalRecoveryPoints,
-                                                                    conf,
-                                                                    clock);
-                                                        } catch (Exception e) {
-                                                            throw new FlussRuntimeException(e);
-                                                        }
-                                                    })
-                            .toArray(Runnable[]::new);
+                    createLogLoadingJobs(
+                            tabletsToLoad, cleanShutdown, finalRecoveryPoints, conf, clock);
 
             long startTime = System.currentTimeMillis();
 
@@ -471,6 +456,70 @@ public void shutdown() {
         LOG.info("Shut down LogManager complete.");
     }
 
+    /** Create runnable jobs for loading logs from tablet directories. */
+    private Runnable[] createLogLoadingJobs(
+            List<File> tabletsToLoad,
+            boolean cleanShutdown,
+            Map<TableBucket, Long> recoveryPoints,
+            Configuration conf,
+            Clock clock) {
+        Runnable[] jobs = new Runnable[tabletsToLoad.size()];
+        for (int i = 0; i < tabletsToLoad.size(); i++) {
+            final File tabletDir = tabletsToLoad.get(i);
+            jobs[i] = createLogLoadingJob(tabletDir, cleanShutdown, recoveryPoints, conf, clock);
+        }
+        return jobs;
+    }
+
+    /** Create a runnable job for loading log from a single tablet directory. */
+    private Runnable createLogLoadingJob(
+            File tabletDir,
+            boolean cleanShutdown,
+            Map<TableBucket, Long> recoveryPoints,
+            Configuration conf,
+            Clock clock) {
+        return new Runnable() {
+            @Override
+            public void run() {
+                LOG.debug("Loading log {}", tabletDir);
+                try {
+                    loadLog(tabletDir, cleanShutdown, recoveryPoints, conf, clock);
+                } catch (Exception e) {
+                    LOG.error("Fail to loadLog from {}", tabletDir, e);
+                    if (e instanceof SchemaNotExistException) {
+                        LOG.error(
+                                "schema not exist, table for {} has already been dropped, the residual data will be removed.",
+                                tabletDir,
+                                e);
+                        FileUtils.deleteDirectoryQuietly(tabletDir);
+
+                        // Also delete corresponding KV tablet directory if it exists
+                        try {
+                            Tuple2<PhysicalTablePath, TableBucket> pathAndBucket =
+                                    FlussPaths.parseTabletDir(tabletDir);
+                            File kvTabletDir =
+                                    FlussPaths.kvTabletDir(
+                                            dataDir, pathAndBucket.f0, pathAndBucket.f1);
+                            if (kvTabletDir.exists()) {
+                                LOG.info(
+                                        "Also removing corresponding KV tablet directory: {}",
+                                        kvTabletDir);
+                                FileUtils.deleteDirectoryQuietly(kvTabletDir);
+                            }
+                        } catch (Exception kvDeleteException) {
+                            LOG.warn(
+                                    "Failed to delete corresponding KV tablet directory for log {}: {}",
+                                    tabletDir,
+                                    kvDeleteException.getMessage());
+                        }
+                        return;
+                    }
+                    throw new FlussRuntimeException(e);
+                }
+            }
+        };
+    }
+
     @VisibleForTesting
     void checkpointRecoveryOffsets() {
         // Assuming TableBucket and LogTablet are actual types used in your application