Fix active load cleanup lifecycle (#17947) (#17963)

(cherry picked from commit c4574b6)

Author: Caideyipi

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java

@@ -4259,7 +4259,17 @@ public void setLoadActiveListeningFailDir(String loadActiveListeningFailDir) {
   }
 
   public String getLoadActiveListeningPipeDir() {
-    return loadActiveListeningPipeDir;
+    return loadActiveListeningPipeDir == null || Objects.equals(loadActiveListeningPipeDir, "")
+        ? extDir
+            + File.separator
+            + IoTDBConstant.LOAD_TSFILE_FOLDER_NAME
+            + File.separator
+            + IoTDBConstant.PIPE_FOLDER_NAME
+        : loadActiveListeningPipeDir;
+  }
+
+  public void setLoadActiveListeningPipeDir(String loadActiveListeningPipeDir) {
+    this.loadActiveListeningPipeDir = addDataHomeDir(loadActiveListeningPipeDir);
   }
 
   public String[] getLoadActiveListeningDirs() {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java

@@ -2484,6 +2484,9 @@ private void loadLoadTsFileProps(TrimProperties properties) throws IOException {
     conf.setLoadActiveListeningFailDir(
         properties.getProperty(
             "load_active_listening_fail_dir", conf.getLoadActiveListeningFailDir()));
+    conf.setLoadActiveListeningPipeDir(
+        properties.getProperty(
+            "load_active_listening_pipe_dir", conf.getLoadActiveListeningPipeDir()));
 
     final long loadActiveListeningCheckIntervalSeconds =
         Long.parseLong(
@@ -2614,6 +2617,9 @@ private void loadLoadTsFileHotModifiedProp(TrimProperties properties) throws IOE
         properties.getProperty(
             "load_active_listening_fail_dir",
             ConfigurationFileUtils.getConfigurationDefaultValue("load_active_listening_fail_dir")));
+    conf.setLoadActiveListeningPipeDir(
+        properties.getProperty(
+            "load_active_listening_pipe_dir", conf.getLoadActiveListeningPipeDir()));
 
     conf.setLoadTsFileSpiltPartitionMaxSize(
         Integer.parseInt(

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

@@ -310,6 +310,7 @@ public void start() throws StartupException {
     }
 
     asyncRecoverTsFileResource();
+    loadTsFileManager.start();
   }
 
   private void startTimedService() {
@@ -394,6 +395,7 @@ private void asyncRecoverTsFileResource() {
 
   @Override
   public void stop() {
+    loadTsFileManager.stop();
     for (DataRegion dataRegion : dataRegionMap.values()) {
       if (dataRegion != null) {
         CompactionScheduleTaskManager.getInstance().unregisterDataRegion(dataRegion);
@@ -412,6 +414,7 @@ public void stop() {
 
   @Override
   public void shutdown(long milliseconds) throws ShutdownException {
+    loadTsFileManager.stop();
     try {
       for (DataRegion dataRegion : dataRegionMap.values()) {
         if (dataRegion != null) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManager.java

@@ -125,9 +125,22 @@ public class LoadTsFileManager {
   public LoadTsFileManager() {
     registerCleanupTaskExecutor();
     recover();
+  }
+
+  public void start() {
     activeLoadAgent.start();
   }
 
+  public void stop() {
+    activeLoadAgent.stop();
+    synchronized (uuid2CleanupTask) {
+      uuid2CleanupTask.values().forEach(CleanupTask::cancel);
+      uuid2CleanupTask.clear();
+      cleanupTaskQueue.clear();
+    }
+    new HashSet<>(uuid2WriterManager.keySet()).forEach(this::forceCloseWriterManager);
+  }
+
   private long getCleanupTaskDelayInMs() {
     return CONFIG.getLoadCleanupTaskExecutionDelayTimeSeconds() * 1000L;
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadAgent.java

@@ -62,6 +62,12 @@ public synchronized void start() {
     activeLoadMetricsCollector.start();
   }
 
+  public synchronized void stop() {
+    activeLoadDirScanner.stop();
+    activeLoadMetricsCollector.stop();
+    activeLoadTsFileLoader.stop();
+  }
+
   /**
    * Clean up all listening directories for active load on DataNode first startup. This method will
    * clean up all files and subdirectories in the listening directories, including: 1. Pending

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadDirScanner.java

@@ -40,6 +40,7 @@
 import java.nio.file.attribute.BasicFileAttributes;
 import java.util.Arrays;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -54,6 +55,7 @@ public class ActiveLoadDirScanner extends ActiveLoadScheduledExecutorService {
   private static final String MODS = ".mods";
 
   private final AtomicReference<String[]> listeningDirsConfig = new AtomicReference<>();
+  private final AtomicReference<String> pipeListeningDirConfig = new AtomicReference<>();
   private final Set<String> listeningDirs = new CopyOnWriteArraySet<>();
 
   private final Set<String> noPermissionDirs = new CopyOnWriteArraySet<>();
@@ -201,8 +203,20 @@ private void hotReloadActiveLoadDirs() {
       } else {
         listeningDirs.clear();
       }
-      // Hot reload active load listening dir for pipe data sync
-      // Active load is always enabled for pipe data sync
+      if (!Objects.equals(
+          IOTDB_CONFIG.getLoadActiveListeningPipeDir(), pipeListeningDirConfig.get())) {
+        synchronized (this) {
+          if (!Objects.equals(
+              IOTDB_CONFIG.getLoadActiveListeningPipeDir(), pipeListeningDirConfig.get())) {
+            if (pipeListeningDirConfig.get() != null) {
+              listeningDirs.remove(pipeListeningDirConfig.get());
+            }
+            pipeListeningDirConfig.set(IOTDB_CONFIG.getLoadActiveListeningPipeDir());
+          }
+        }
+      }
+
+      // Active load is always enabled for pipe data sync.
       listeningDirs.add(IOTDB_CONFIG.getLoadActiveListeningPipeDir());
 
       // Create directories if not exists

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadScheduledExecutorService.java

@@ -45,14 +45,19 @@ public abstract class ActiveLoadScheduledExecutorService {
 
   private static final long MIN_EXECUTION_INTERVAL_SECONDS =
       IOTDB_CONFIG.getLoadActiveListeningCheckIntervalSeconds();
-  private final ScheduledExecutorService scheduledExecutorService;
+  private final ThreadName threadName;
+  private ScheduledExecutorService scheduledExecutorService;
   private Future<?> future;
 
   private final List<Pair<WrappedRunnable, Long>> jobs = new CopyOnWriteArrayList<>();
 
   protected ActiveLoadScheduledExecutorService(final ThreadName threadName) {
-    scheduledExecutorService =
-        IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor(threadName.name());
+    this.threadName = threadName;
+    scheduledExecutorService = newScheduledExecutorService();
+  }
+
+  private ScheduledExecutorService newScheduledExecutorService() {
+    return IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor(threadName.name());
   }
 
   public void register(Runnable runnable) {
@@ -73,6 +78,9 @@ public void runMayThrow() {
 
   public synchronized void start() {
     if (future == null) {
+      if (scheduledExecutorService.isShutdown()) {
+        scheduledExecutorService = newScheduledExecutorService();
+      }
       future =
           ScheduledExecutorUtil.safelyScheduleWithFixedDelay(
               scheduledExecutorService,
@@ -96,5 +104,12 @@ public synchronized void stop() {
       future = null;
       LOGGER.info("Active load periodical jobs executor is stopped successfully.");
     }
+    scheduledExecutorService.shutdownNow();
+    try {
+      scheduledExecutorService.awaitTermination(30, TimeUnit.SECONDS);
+    } catch (InterruptedException e) {
+      LOGGER.warn("{} still doesn't exit after 30s", threadName.getName());
+      Thread.currentThread().interrupt();
+    }
   }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoader.java

@@ -154,6 +154,25 @@ private void adjustExecutorIfNecessary() {
     }
   }
 
+  public void stop() {
+    final WrappedThreadPoolExecutor executor = activeLoadExecutor.getAndSet(null);
+    if (executor == null) {
+      return;
+    }
+
+    executor.shutdownNow();
+    try {
+      if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {
+        LOGGER.warn(
+            "{} still doesn't exit after 30s", ThreadName.ACTIVE_LOAD_TSFILE_LOADER.getName());
+      }
+    } catch (final InterruptedException e) {
+      LOGGER.warn(
+          "{} still doesn't exit after 30s", ThreadName.ACTIVE_LOAD_TSFILE_LOADER.getName());
+      Thread.currentThread().interrupt();
+    }
+  }
+
   private void tryLoadPendingTsFiles() {
     final IClientSession session =
         new InternalClientSession(
@@ -200,18 +219,22 @@ private Optional<ActiveLoadPendingQueue.ActiveLoadEntry> tryGetNextPendingFile()
         Math.max(1, IOTDB_CONFIG.getLoadActiveListeningCheckIntervalSeconds() << 1);
     long currentRetryTimes = 0;
 
-    while (true) {
+    while (!Thread.currentThread().isInterrupted()) {
       final ActiveLoadPendingQueue.ActiveLoadEntry entry = pendingQueue.dequeueFromPending();
       if (Objects.nonNull(entry)) {
         return Optional.of(entry);
       }
 
       LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(1));
+      if (Thread.currentThread().isInterrupted()) {
+        return Optional.empty();
+      }
 
       if (currentRetryTimes++ >= maxRetryTimes) {
         return Optional.empty();
       }
     }
+    return Optional.empty();
   }
 
   private TSStatus loadTsFile(

iotdb-core/datanode/src/test/resources/iotdb-system.properties

@@ -32,6 +32,7 @@ udf_lib_dir=target/ext/udf
 trigger_lib_dir=target/ext/trigger
 pipe_lib_dir=target/ext/pipe
 load_active_listening_dirs=target/ext/load/pending
+load_active_listening_pipe_dir=target/ext/load/pipe
 load_active_listening_fail_dir=target/ext/load/failed
 
 ####################

iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template

@@ -2062,6 +2062,18 @@ load_active_listening_enable=true
 # If its prefix is "/", then the path is absolute. Otherwise, it is relative.
 load_active_listening_dirs=ext/load/pending
 
+# The directory to be actively listened for tsfile loading from Pipe.
+# Only one directory can be configured.
+# effectiveMode: hot_reload
+# Datatype: String
+# For windows platform
+# If its prefix is a drive specifier followed by "\\", or if its prefix is "\\\\", then the path is absolute.
+# Otherwise, it is relative.
+# load_active_listening_pipe_dir=ext\\load\\pipe
+# For Linux platform
+# If its prefix is "/", then the path is absolute. Otherwise, it is relative.
+load_active_listening_pipe_dir=ext/load/pipe
+
 # The directory where tsfile are moved if the active listening mode fails to load them.
 # Only one directory can be configured.
 # effectiveMode: hot_reload