Pipe: Made historical extractor do not extract TsFiles that have not started to close

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java

@@ -156,18 +156,6 @@ private void restartPipeToReloadResourceIfNeeded(final PipeMeta pipeMeta) {
       return;
     }
 
-    if (PIPE_NAME_TO_LAST_RESTART_TIME_MAP.isEmpty()) {
-      LOGGER.info(
-          "Flushing storage engine before restarting pipe {}.",
-          pipeMeta.getStaticMeta().getPipeName());
-      final long currentTime = System.currentTimeMillis();
-      StorageEngine.getInstance().syncCloseAllProcessor();
-      WALManager.getInstance().syncDeleteOutdatedFilesInWALNodes();
-      LOGGER.info(
-          "Finished flushing storage engine, time cost: {} ms.",
-          System.currentTimeMillis() - currentTime);
-    }
-
     restartStuckPipe(pipeMeta);
     LOGGER.info(
         "Reloaded resource for stopped pipe {} before starting it.",
@@ -562,21 +550,7 @@ protected void collectPipeMetaListInternal(
   ///////////////////////// Restart Logic /////////////////////////
 
   public void restartAllStuckPipes() {
-    final List<String> removedPipeName = removeOutdatedPipeInfoFromLastRestartTimeMap();
-    if (!removedPipeName.isEmpty()) {
-      final long currentTime = System.currentTimeMillis();
-      LOGGER.info(
-          "Pipes {} now can dynamically adjust their extraction strategies. "
-              + "Start to flush storage engine to trigger the adjustment.",
-          removedPipeName);
-      StorageEngine.getInstance().syncCloseAllProcessor();
-      WALManager.getInstance().syncDeleteOutdatedFilesInWALNodes();
-      LOGGER.info(
-          "Finished flushing storage engine, time cost: {} ms.",
-          System.currentTimeMillis() - currentTime);
-      LOGGER.info("Skipping restarting pipes this round because of the dynamic flushing.");
-      return;
-    }
+    removeOutdatedPipeInfoFromLastRestartTimeMap();
 
     if (!tryWriteLockWithTimeOut(5)) {
       return;

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionExtractor.java

@@ -51,6 +51,7 @@
 import org.apache.iotdb.db.pipe.resource.tsfile.PipeTsFileResourceManager;
 import org.apache.iotdb.db.storageengine.StorageEngine;
 import org.apache.iotdb.db.storageengine.dataregion.DataRegion;
+import org.apache.iotdb.db.storageengine.dataregion.memtable.TsFileProcessor;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileManager;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.generator.TsFileNameGenerator;
@@ -612,9 +613,13 @@ private void extractTsFiles(
                       // Some resource is marked as deleted but not removed from the list.
                       !resource.isDeleted()
                           && (
-                          // Some resource may not be closed due to the control of
-                          // PIPE_MIN_FLUSH_INTERVAL_IN_MS. We simply ignore them.
+                          // If the tsFile is not already marked closing, it is not captured by the
+                          // pipe realtime module. Thus, we can wait for the realtime sync module to
+                          // handle this, to avoid blocking the pipe sync process.
                           !resource.isClosed()
+                                  && Optional.ofNullable(resource.getProcessor())
+                                      .map(TsFileProcessor::alreadyMarkedClosing)
+                                      .orElse(true)
                               || mayTsFileContainUnprocessedData(resource)
                                   && isTsFileResourceOverlappedWithTimeRange(resource)
                                   && isTsFileGeneratedAfterExtractionTimeLowerBound(resource)
@@ -629,9 +634,13 @@ && mayTsFileResourceOverlappedWithPattern(resource)))
                       // Some resource is marked as deleted but not removed from the list.
                       !resource.isDeleted()
                           && (
-                          // Some resource may not be closed due to the control of
-                          // PIPE_MIN_FLUSH_INTERVAL_IN_MS. We simply ignore them.
+                          // If the tsFile is not already marked closing, it is not captured by the
+                          // pipe realtime module. Thus, we can wait for the realtime sync module to
+                          // handle this, to avoid blocking the pipe sync process.
                           !resource.isClosed()
+                                  && Optional.ofNullable(resource.getProcessor())
+                                      .map(TsFileProcessor::alreadyMarkedClosing)
+                                      .orElse(true)
                               || mayTsFileContainUnprocessedData(resource)
                                   && isTsFileResourceOverlappedWithTimeRange(resource)
                                   && isTsFileGeneratedAfterExtractionTimeLowerBound(resource)
@@ -687,10 +696,10 @@ private boolean mayTsFileContainUnprocessedData(final TsFileResource resource) {
       // For consensus pipe, we only focus on the progressIndex that is generated from local write
       // instead of replication or something else.
       ProgressIndex dedicatedProgressIndex =
-          tryToExtractLocalProgressIndexForIoTV2(resource.getMaxProgressIndexAfterClose());
+          tryToExtractLocalProgressIndexForIoTV2(resource.getMaxProgressIndex());
       return greaterThanStartIndex(dedicatedProgressIndex);
     }
-    return greaterThanStartIndex(resource.getMaxProgressIndexAfterClose());
+    return greaterThanStartIndex(resource.getMaxProgressIndex());
   }
 
   private boolean greaterThanStartIndex(ProgressIndex progressIndex) {

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

@@ -1294,18 +1294,18 @@ public Future<?> asyncClose() {
       IMemTable tmpMemTable = workMemTable == null ? new NotifyFlushMemTable() : workMemTable;
 
       try {
+        // When invoke closing TsFile after insert data to memTable, we shouldn't flush until invoke
+        // flushing memTable in System module.
+        Future<?> future = addAMemtableIntoFlushingList(tmpMemTable);
+        shouldClose = true;
+
         PipeInsertionDataNodeListener.getInstance()
             .listenToTsFile(
                 dataRegionInfo.getDataRegion().getDataRegionId(),
                 dataRegionInfo.getDataRegion().getDatabaseName(),
                 tsFileResource,
                 false,
                 tmpMemTable.isTotallyGeneratedByPipe());
-
-        // When invoke closing TsFile after insert data to memTable, we shouldn't flush until invoke
-        // flushing memTable in System module.
-        Future<?> future = addAMemtableIntoFlushingList(tmpMemTable);
-        shouldClose = true;
         return future;
       } catch (Exception e) {
         logger.error(

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java

@@ -473,9 +473,8 @@ protected boolean createPipe(final PipeMeta pipeMetaFromCoordinator) throws Ille
     // Create pipe tasks
     final Map<Integer, PipeTask> pipeTasks = buildPipeTasks(pipeMetaFromCoordinator);
 
-    // Trigger create() method for each pipe task by parallel stream
     final long startTime = System.currentTimeMillis();
-    pipeTasks.values().parallelStream().forEach(PipeTask::create);
+    pipeTasks.values().forEach(PipeTask::create);
     LOGGER.info(
         "Create all pipe tasks on Pipe {} successfully within {} ms",
         pipeName,
@@ -528,9 +527,8 @@ protected boolean dropPipe(final String pipeName, final long creationTime) {
       return false;
     }
 
-    // Trigger drop() method for each pipe task by parallel stream
     final long startTime = System.currentTimeMillis();
-    pipeTasks.values().parallelStream().forEach(PipeTask::drop);
+    pipeTasks.values().forEach(PipeTask::drop);
     LOGGER.info(
         "Drop all pipe tasks on Pipe {} successfully within {} ms",
         pipeName,
@@ -567,9 +565,8 @@ protected boolean dropPipe(final String pipeName) {
       return false;
     }
 
-    // Trigger drop() method for each pipe task by parallel stream
     final long startTime = System.currentTimeMillis();
-    pipeTasks.values().parallelStream().forEach(PipeTask::drop);
+    pipeTasks.values().forEach(PipeTask::drop);
     LOGGER.info(
         "Drop all pipe tasks on Pipe {} successfully within {} ms",
         pipeName,
@@ -600,9 +597,8 @@ protected void startPipe(final String pipeName, final long creationTime) {
       return;
     }
 
-    // Trigger start() method for each pipe task by parallel stream
     final long startTime = System.currentTimeMillis();
-    pipeTasks.values().parallelStream().forEach(PipeTask::start);
+    pipeTasks.values().forEach(PipeTask::start);
     LOGGER.info(
         "Start all pipe tasks on Pipe {} successfully within {} ms",
         pipeName,
@@ -639,9 +635,8 @@ private void stopPipe(final String pipeName, final long creationTime) {
       return;
     }
 
-    // Trigger stop() method for each pipe task by parallel stream
     final long startTime = System.currentTimeMillis();
-    pipeTasks.values().parallelStream().forEach(PipeTask::stop);
+    pipeTasks.values().forEach(PipeTask::stop);
     LOGGER.info(
         "Stop all pipe tasks on Pipe {} successfully within {} ms",
         pipeName,