Fix: send set configuration only to target nodes and harden compaction schedule interruption handling (#17469)

Author: shuwenwei

integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBSetConfigurationTableIT.java

@@ -109,6 +109,9 @@ public void testSetConfiguration() {
         statement.execute(
             "set configuration inner_compaction_candidate_file_num='1',max_cross_compaction_candidate_file_num='1' on "
                 + dnId);
+        if (dnId == 5) {
+          statement.execute("set configuration compaction_schedule_thread_num='2' on 5");
+        }
       }
     } catch (Exception e) {
       Assert.fail(e.getMessage());
@@ -131,6 +134,17 @@ public void testSetConfiguration() {
               "enable_cross_space_compaction=false",
               "inner_compaction_candidate_file_num=1",
               "max_cross_compaction_candidate_file_num=1"));
+      boolean scheduleThreadNumChanged =
+          checkConfigFileContains(
+              dnId,
+              EnvFactory.getEnv().getDataNodeWrapperList().get(i),
+              "compaction_schedule_thread_num=2");
+      if (scheduleThreadNumChanged && dnId != 5) {
+        Assert.fail();
+      }
+      if (!scheduleThreadNumChanged && dnId == 5) {
+        Assert.fail();
+      }
     }
   }
 

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java

@@ -1052,7 +1052,7 @@ public List<TSStatus> setConfiguration(TSetConfigurationReq req) {
     if (!targetDataNodes.isEmpty()) {
       DataNodeAsyncRequestContext<Object, TSStatus> clientHandler =
           new DataNodeAsyncRequestContext<>(
-              CnToDnAsyncRequestType.SET_CONFIGURATION, req, dataNodeLocationMap);
+              CnToDnAsyncRequestType.SET_CONFIGURATION, req, targetDataNodes);
       CnToDnInternalServiceAsyncRequestManager.getInstance()
           .sendAsyncRequestWithRetry(clientHandler);
       responseList.addAll(clientHandler.getResponseList());

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java

@@ -3705,7 +3705,11 @@ public void executeTTLCheckForObjectFiles() throws InterruptedException {
       if (!regionObjectDir.isDirectory()) {
         continue;
       }
-      CompactionUtils.executeTTLCheckObjectFilesForTableModel(regionObjectDir, databaseName);
+      try {
+        CompactionUtils.executeTTLCheckObjectFilesForTableModel(regionObjectDir, databaseName);
+      } catch (Exception e) {
+        logger.error("Failed to execute object ttl check", e);
+      }
     }
     CompactionMetrics.getInstance()
         .updateTTLCheckForObjectFileCost(System.currentTimeMillis() - startTime);

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionUtils.java

@@ -69,9 +69,11 @@
 import javax.annotation.Nullable;
 
 import java.io.File;
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
+import java.nio.file.NoSuchFileException;
 import java.nio.file.attribute.BasicFileAttributes;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -595,7 +597,10 @@ private static void recursiveTTLCheckForTableDir(
           checkTTLAndDeleteExpiredObjectFile(currentFile, basicFileAttributes, lowerBoundInMS);
           return;
         }
-      } catch (IOException ignored) {
+      } catch (FileNotFoundException | NoSuchFileException ignored) {
+        // may be deleted by other thread
+      } catch (IOException e) {
+        logger.warn("Failed to read file attributes: {}", currentFile, e);
       }
     }
     File[] children = currentFile.listFiles();
@@ -606,8 +611,16 @@ private static void recursiveTTLCheckForTableDir(
     // block-aligned and reflects allocated directory entry blocks.
     acquireCompactionReadRate(currentFile.length());
     for (File child : children) {
-      recursiveTTLCheckForTableDir(
-          child, depth + 1, maxObjectFileDepth, canDistinguishDirectoryByFileName, lowerBoundInMS);
+      try {
+        recursiveTTLCheckForTableDir(
+            child,
+            depth + 1,
+            maxObjectFileDepth,
+            canDistinguishDirectoryByFileName,
+            lowerBoundInMS);
+      } catch (Exception e) {
+        logger.warn("Failed to check table dir: {}", child, e);
+      }
     }
   }
 
@@ -637,7 +650,10 @@ private static void checkTTLAndDeleteExpiredObjectFile(
       FileMetrics.getInstance().decreaseObjectFileNum(1);
       FileMetrics.getInstance().decreaseObjectFileSize(attributes.size());
       logger.info("Remove object file {}, size is {}(byte)", file.getPath(), attributes.size());
-    } catch (Exception ignored) {
+    } catch (FileNotFoundException | NoSuchFileException ignored) {
+      // may be deleted by other thread
+    } catch (Exception e) {
+      logger.warn("Failed to delete expired object file: {}", file, e);
     }
   }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskManager.java

@@ -65,6 +65,7 @@ public class CompactionScheduleTaskManager implements IService {
       ConcurrentHashMap.newKeySet();
   private ReentrantLock lock = new ReentrantLock();
   private volatile boolean init = false;
+  private volatile boolean isStoppingAllScheduleTask = false;
 
   @Override
   public void start() throws StartupException {
@@ -76,8 +77,13 @@ public void start() throws StartupException {
     logger.info("Compaction schedule task manager started.");
   }
 
+  public boolean isStoppingAllScheduleTask() {
+    return isStoppingAllScheduleTask;
+  }
+
   public void stopCompactionScheduleTasks() throws InterruptedException {
     lock.lock();
+    isStoppingAllScheduleTask = true;
     try {
       for (Future<Void> task : submitCompactionScheduleTaskFutures) {
         task.cancel(true);
@@ -121,6 +127,7 @@ public void checkAndMayApplyConfigurationChange() throws InterruptedException {
 
   public void startScheduleTasks() {
     lock.lock();
+    isStoppingAllScheduleTask = false;
     try {
       // compaction selector
       for (int workerId = 0; workerId < compactionSelectorNum; workerId++) {
@@ -144,6 +151,7 @@ public void startScheduleTasks() {
   @Override
   public void stop() {
     lock.lock();
+    isStoppingAllScheduleTask = true;
     try {
       if (!init) {
         return;
@@ -160,6 +168,7 @@ public void stop() {
   @Override
   public void waitAndStop(long milliseconds) {
     lock.lock();
+    isStoppingAllScheduleTask = true;
     try {
       if (!init) {
         return;

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskWorker.java

@@ -72,9 +72,26 @@ public Void call() {
           dataRegion.executeCompaction();
         }
       } catch (InterruptedException ignored) {
+        boolean isStoppedByUser =
+            CompactionScheduleTaskManager.getInstance().isStoppingAllScheduleTask();
         logger.info(
-            "[CompactionScheduleTaskWorker-{}] compaction schedule is interrupted", workerId);
-        return null;
+            "[CompactionScheduleTaskWorker-{}] compaction schedule is interrupted, isStopByUser: {}",
+            workerId,
+            isStoppedByUser);
+        if (isStoppedByUser) {
+          return null;
+        }
+      } catch (Exception e) {
+        logger.error(
+            "[CompactionScheduleTaskWorker-{}] Failed to execute compaction schedule task",
+            workerId,
+            e);
+      } catch (Throwable t) {
+        logger.error(
+            "[CompactionScheduleTaskWorker-{}] Failed to execute compaction schedule task and cannot recover",
+            workerId,
+            t);
+        throw t;
       }
     }
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/TTLScheduleTask.java

@@ -78,8 +78,21 @@ public Void call() throws Exception {
           }
         }
       } catch (InterruptedException ignored) {
-        logger.info("[TTLCheckTask-{}] TTL checker is interrupted", workerId);
-        return null;
+        boolean isStoppedByUser =
+            CompactionScheduleTaskManager.getInstance().isStoppingAllScheduleTask();
+        logger.info(
+            "[TTLCheckTask-{}] TTL checker is interrupted, isStoppedByUser: {}",
+            workerId,
+            isStoppedByUser);
+        if (isStoppedByUser) {
+          return null;
+        }
+      } catch (Exception e) {
+        logger.error("[TTLCheckTask-{}] Failed to execute ttl check", workerId, e);
+      } catch (Throwable t) {
+        logger.error(
+            "[TTLCheckTask-{}] Failed to execute ttl check and cannot recover", workerId, t);
+        throw t;
       }
     }
   }