Pipe: Enable flushing storage engine after last terminate event reported (#15465)

Co-authored-by: Steve Yurong Su <rong@apache.org>

Author: Caideyipi

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeDataNodeRuntimeAgent.java

@@ -35,6 +35,7 @@
 import org.apache.iotdb.commons.utils.TestOnly;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
 import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
+import org.apache.iotdb.db.pipe.event.common.terminate.PipeTerminateEvent;
 import org.apache.iotdb.db.pipe.extractor.schemaregion.SchemaRegionListeningQueue;
 import org.apache.iotdb.db.pipe.resource.PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
@@ -89,6 +90,11 @@ public synchronized void start() throws StartupException {
         "PipeTaskAgent#restartAllStuckPipes",
         PipeDataNodeAgent.task()::restartAllStuckPipes,
         PipeConfig.getInstance().getPipeStuckRestartIntervalSeconds());
+    registerPeriodicalJob(
+        "PipeTaskAgent#flushDataRegionIfNeeded",
+        PipeTerminateEvent::flushDataRegionIfNeeded,
+        PipeConfig.getInstance().getPipeFlushAfterLastTerminateSeconds());
+
     pipePeriodicalJobExecutor.start();
 
     if (PipeConfig.getInstance().getPipeEventReferenceTrackingEnabled()) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/terminate/PipeTerminateEvent.java

@@ -19,17 +19,24 @@
 
 package org.apache.iotdb.db.pipe.event.common.terminate;
 
+import org.apache.iotdb.common.rpc.thrift.TFlushReq;
 import org.apache.iotdb.commons.consensus.index.ProgressIndex;
 import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
 import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
+import org.apache.iotdb.commons.pipe.config.PipeConfig;
 import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
 import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
 import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
 import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
 import org.apache.iotdb.db.pipe.agent.task.PipeDataNodeTask;
 import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
+import org.apache.iotdb.db.storageengine.StorageEngine;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicLong;
 
 /**
  * The {@link PipeTerminateEvent} is an {@link EnrichedEvent} that controls the termination of pipe,
@@ -38,6 +45,41 @@
  * be discarded.
  */
 public class PipeTerminateEvent extends EnrichedEvent {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(PipeTerminateEvent.class);
+
+  private static final AtomicLong PROGRESS_REPORT_COUNT = new AtomicLong(0);
+  private static final AtomicLong LAST_PROGRESS_REPORT_TIME = new AtomicLong(0);
+
+  public static void flushDataRegionIfNeeded() {
+    if (PROGRESS_REPORT_COUNT.get() > 0
+        && PROGRESS_REPORT_COUNT.get()
+            > PipeConfig.getInstance().getPipeFlushAfterTerminateCount()) {
+      flushDataRegion();
+      return;
+    }
+
+    if (LAST_PROGRESS_REPORT_TIME.get() > 0
+        && System.currentTimeMillis() - LAST_PROGRESS_REPORT_TIME.get()
+            > PipeConfig.getInstance().getPipeFlushAfterLastTerminateSeconds() * 1000L) {
+      flushDataRegion();
+    }
+  }
+
+  private static void flushDataRegion() {
+    try {
+      StorageEngine.getInstance().operateFlush(new TFlushReq());
+      PROGRESS_REPORT_COUNT.set(0);
+      LAST_PROGRESS_REPORT_TIME.set(0);
+      LOGGER.info("Force flush all data regions because of last progress report time.");
+    } catch (final Exception e) {
+      LOGGER.warn(
+          "Failed to flush all data regions, please check the error message: {}",
+          e.getMessage(),
+          e);
+    }
+  }
+
   private final int dataRegionId;
 
   public PipeTerminateEvent(
@@ -106,6 +148,9 @@ public boolean mayEventPathsOverlappedWithPattern() {
 
   @Override
   public void reportProgress() {
+    PROGRESS_REPORT_COUNT.incrementAndGet();
+    LAST_PROGRESS_REPORT_TIME.set(System.currentTimeMillis());
+
     // To avoid deadlock
     CompletableFuture.runAsync(
         () -> PipeDataNodeAgent.task().markCompleted(pipeName, dataRegionId));

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java

@@ -273,6 +273,8 @@ public class CommonConfig {
   private long pipeStuckRestartIntervalSeconds = 120;
   private long pipeStuckRestartMinIntervalMs = 5 * 60 * 1000L; // 5 minutes
   private boolean pipeEpochKeepTsFileAfterStuckRestartEnabled = false;
+  private long pipeFlushAfterLastTerminateSeconds = 30;
+  private long pipeFlushAfterTerminateCount = 30;
   private long pipeStorageEngineFlushTimeIntervalMs = Long.MAX_VALUE;
 
   private int pipeMetaReportMaxLogNumPerRound = 10;
@@ -1478,6 +1480,31 @@ public void setPipeStorageEngineFlushTimeIntervalMs(long pipeStorageEngineFlushT
         "pipeStorageEngineFlushTimeIntervalMs is set to {}", pipeStorageEngineFlushTimeIntervalMs);
   }
 
+  public long getPipeFlushAfterLastTerminateSeconds() {
+    return pipeFlushAfterLastTerminateSeconds;
+  }
+
+  public void setPipeFlushAfterLastTerminateSeconds(long pipeFlushAfterLastTerminateSeconds) {
+    if (this.pipeFlushAfterLastTerminateSeconds == pipeFlushAfterLastTerminateSeconds) {
+      return;
+    }
+    this.pipeFlushAfterLastTerminateSeconds = pipeFlushAfterLastTerminateSeconds;
+    logger.info(
+        "pipeFlushAfterLastTerminateSeconds is set to {}", pipeFlushAfterLastTerminateSeconds);
+  }
+
+  public long getPipeFlushAfterTerminateCount() {
+    return pipeFlushAfterTerminateCount;
+  }
+
+  public void setPipeFlushAfterTerminateCount(long pipeFlushAfterTerminateCount) {
+    if (this.pipeFlushAfterTerminateCount == pipeFlushAfterTerminateCount) {
+      return;
+    }
+    this.pipeFlushAfterTerminateCount = pipeFlushAfterTerminateCount;
+    logger.info("pipeFlushAfterTerminateCount is set to {}", pipeFlushAfterTerminateCount);
+  }
+
   public int getPipeMetaReportMaxLogNumPerRound() {
     return pipeMetaReportMaxLogNumPerRound;
   }

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java

@@ -303,6 +303,14 @@ public boolean isPipeEpochKeepTsFileAfterStuckRestartEnabled() {
     return COMMON_CONFIG.isPipeEpochKeepTsFileAfterStuckRestartEnabled();
   }
 
+  public long getPipeFlushAfterTerminateCount() {
+    return COMMON_CONFIG.getPipeFlushAfterTerminateCount();
+  }
+
+  public long getPipeFlushAfterLastTerminateSeconds() {
+    return COMMON_CONFIG.getPipeFlushAfterLastTerminateSeconds();
+  }
+
   public long getPipeStorageEngineFlushTimeIntervalMs() {
     return COMMON_CONFIG.getPipeStorageEngineFlushTimeIntervalMs();
   }
@@ -530,6 +538,8 @@ public void printAllConfigs() {
     LOGGER.info(
         "PipeEpochKeepTsFileAfterStuckRestartEnabled: {}",
         isPipeEpochKeepTsFileAfterStuckRestartEnabled());
+    LOGGER.info("PipeFlushAfterTerminateCount: {}", getPipeFlushAfterTerminateCount());
+    LOGGER.info("PipeFlushAfterLastTerminateSeconds: {}", getPipeFlushAfterLastTerminateSeconds());
     LOGGER.info(
         "PipeStorageEngineFlushTimeIntervalMs: {}", getPipeStorageEngineFlushTimeIntervalMs());
 

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java

@@ -431,6 +431,16 @@ public static void loadPipeInternalConfig(CommonConfig config, TrimProperties pr
             properties.getProperty(
                 "pipe_stuck_restart_min_interval_ms",
                 String.valueOf(config.getPipeStuckRestartMinIntervalMs()))));
+    config.setPipeFlushAfterLastTerminateSeconds(
+        Long.parseLong(
+            properties.getProperty(
+                "pipe_flush_after_last_terminate_seconds",
+                String.valueOf(config.getPipeFlushAfterLastTerminateSeconds()))));
+    config.setPipeFlushAfterTerminateCount(
+        Long.parseLong(
+            properties.getProperty(
+                "pipe_flush_after_terminate_count",
+                String.valueOf(config.getPipeFlushAfterTerminateCount()))));
     config.setPipeEpochKeepTsFileAfterStuckRestartEnabled(
         Boolean.parseBoolean(
             properties.getProperty(