Pipe: Optimized the log for failed events and completed tasks/pipe

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java

@@ -157,11 +157,24 @@
       if (Boolean.TRUE.equals(isPipeCompletedFromAgent)) {
 
         temporaryMeta.markDataNodeCompleted(nodeId);
+        LOGGER.info(
+            "Detected historical pipe completion report from DataNode {} for pipe {}. remainingEventCount: {}, remainingTime: {}, completedDataNodes: {}",
+            nodeId,
+            staticMeta.getPipeName(),
+            pipeHeartbeat.getRemainingEventCount(staticMeta),
+            pipeHeartbeat.getRemainingTime(staticMeta),
+            temporaryMeta.getCompletedDataNodeIds());
 
         final Set<Integer> uncompletedDataNodeIds =
             configManager.getNodeManager().getRegisteredDataNodeLocations().keySet();
         uncompletedDataNodeIds.removeAll(temporaryMeta.getCompletedDataNodeIds());
         if (uncompletedDataNodeIds.isEmpty()) {
+          LOGGER.info(
+              "All DataNodes reported historical pipe {} completed. globalRemainingEventCount: {}, globalRemainingTime: {}, staticMeta: {}",
+              staticMeta.getPipeName(),
+              temporaryMeta.getGlobalRemainingEvents(),
+              temporaryMeta.getGlobalRemainingTime(),
+              staticMeta);
           pipeTaskInfo.get().removePipeMeta(staticMeta.getPipeName());
           LOGGER.info(
               ManagerMessages.DETECTED_COMPLETION_OF_PIPE_STATIC_META_REMOVE_IT,

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

@@ -32,6 +32,7 @@
 import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent;
 import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
 import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
+import org.apache.iotdb.db.pipe.event.common.terminate.PipeTerminateEvent;
 import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
 import org.apache.iotdb.db.pipe.source.schemaregion.IoTDBSchemaRegionSource;
 import org.apache.iotdb.db.pipe.source.schemaregion.PipePlanTreePrivilegeParseVisitor;
@@ -138,12 +139,20 @@ private void parseAndCollectEvent(final PipeTsFileInsertionEvent sourceEvent) th
 
     if (skipParsing || !forceTabletFormat && canSkipParsing4TsFileEvent(sourceEvent)) {
       collectEvent(sourceEvent);
+      if (sourceEvent.isGeneratedByHistoricalExtractor()) {
+        PipeTerminateEvent.markHistoricalTsFileUnsplit(
+            sourceEvent.getPipeName(), sourceEvent.getCreationTime(), regionId);
+      }
       return;
     }
 
     try {
       sourceEvent.consumeTabletInsertionEventsWithRetry(
           this::collectParsedRawTableEvent, "PipeEventCollector::parseAndCollectEvent");
+      if (sourceEvent.isGeneratedByHistoricalExtractor()) {
+        PipeTerminateEvent.markHistoricalTsFileSplit(
+            sourceEvent.getPipeName(), sourceEvent.getCreationTime(), regionId);
+      }
     } finally {
       sourceEvent.close();
     }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java

@@ -36,6 +36,7 @@
 import org.apache.iotdb.db.pipe.event.UserDefinedEnrichedEvent;
 import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent;
 import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
+import org.apache.iotdb.db.pipe.event.common.terminate.PipeTerminateEvent;
 import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
 import org.apache.iotdb.db.pipe.metric.overview.PipeDataNodeSinglePipeMetrics;
 import org.apache.iotdb.db.pipe.metric.processor.PipeProcessorMetrics;
@@ -181,6 +182,12 @@ protected boolean executeOnce() throws Exception {
                     }
                   },
                   "PipeProcessorSubtask::executeOnce");
+              if (tsFileInsertionEvent.isGeneratedByHistoricalExtractor()) {
+                PipeTerminateEvent.markHistoricalTsFileSplit(
+                    tsFileInsertionEvent.getPipeName(),
+                    tsFileInsertionEvent.getCreationTime(),
+                    regionId);
+              }
               if (ex.get() != null) {
                 throw ex.get();
               }

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

@@ -33,9 +33,16 @@
 import org.apache.iotdb.db.pipe.agent.task.PipeDataNodeTask;
 import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Objects;
 import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 
 /**
  * The {@link PipeTerminateEvent} is an {@link EnrichedEvent} that controls the termination of pipe,
@@ -45,6 +52,8 @@
  */
 public class PipeTerminateEvent extends EnrichedEvent {
 
+  private static final Logger LOGGER = LoggerFactory.getLogger(PipeTerminateEvent.class);
+
   private final int dataRegionId;
 
   private final boolean shouldMark;
@@ -58,6 +67,9 @@
   // Do not use call run policy to avoid deadlock
   private static final ExecutorService terminateExecutor = createTerminateExecutor();
 
+  private static final ConcurrentMap<HistoricalTransferKey, HistoricalTransferSummaryCounter>
+      HISTORICAL_TRANSFER_SUMMARY_COUNTER_MAP = new ConcurrentHashMap<>();
+
   private static ExecutorService createTerminateExecutor() {
     final WrappedThreadPoolExecutor executor =
         new WrappedThreadPoolExecutor(
@@ -145,6 +157,18 @@
   }
 
   public void markCompleted() {
+    final HistoricalTransferSummary summary =
+        snapshotAndClearHistoricalTransferSummary(pipeName, creationTime, dataRegionId);
+    if (Objects.nonNull(summary)) {
+      LOGGER.info(
+          "Pipe {}@{}: terminate event committed for historical transfer. creationTime: {}, shouldMark: {}. {}",
+          pipeName,
+          dataRegionId,
+          creationTime,
+          shouldMark,
+          summary.toReportMessage());
+    }
+
     // To avoid deadlock
     if (shouldMark) {
       terminateExecutor.submit(
@@ -159,4 +183,159 @@
         + " - "
         + super.toString();
   }
+
+  public static void initializeHistoricalTransferSummary(
+      final String pipeName,
+      final long creationTime,
+      final int dataRegionId,
+      final long extractedHistoricalTsFileCount,
+      final long extractedHistoricalDeletionCount) {
+    HISTORICAL_TRANSFER_SUMMARY_COUNTER_MAP
+        .computeIfAbsent(
+            new HistoricalTransferKey(pipeName, creationTime, dataRegionId),
+            ignored -> new HistoricalTransferSummaryCounter())
+        .initialize(extractedHistoricalTsFileCount, extractedHistoricalDeletionCount);
+  }
+
+  public static void markHistoricalTsFileSkipped(
+      final String pipeName, final long creationTime, final int dataRegionId) {
+    getOrCreateHistoricalTransferSummaryCounter(pipeName, creationTime, dataRegionId)
+        .skippedHistoricalTsFileCount
+        .incrementAndGet();
+  }
+
+  public static void markHistoricalTsFileSplit(
+      final String pipeName, final long creationTime, final int dataRegionId) {
+    getOrCreateHistoricalTransferSummaryCounter(pipeName, creationTime, dataRegionId)
+        .splitHistoricalTsFileCount
+        .incrementAndGet();
+  }
+
+  public static void markHistoricalTsFileUnsplit(
+      final String pipeName, final long creationTime, final int dataRegionId) {
+    getOrCreateHistoricalTransferSummaryCounter(pipeName, creationTime, dataRegionId)
+        .unsplitHistoricalTsFileCount
+        .incrementAndGet();
+  }
+
+  public static HistoricalTransferSummary snapshotHistoricalTransferSummary(
+      final String pipeName, final long creationTime, final int dataRegionId) {
+    final HistoricalTransferSummaryCounter counter =
+        HISTORICAL_TRANSFER_SUMMARY_COUNTER_MAP.get(
+            new HistoricalTransferKey(pipeName, creationTime, dataRegionId));
+    return Objects.nonNull(counter) ? counter.snapshot() : null;
+  }
+
+  public static void clearHistoricalTransferSummary(
+      final String pipeName, final long creationTime, final int dataRegionId) {
+    HISTORICAL_TRANSFER_SUMMARY_COUNTER_MAP.remove(
+        new HistoricalTransferKey(pipeName, creationTime, dataRegionId));
+  }
+
+  private static HistoricalTransferSummary snapshotAndClearHistoricalTransferSummary(
+      final String pipeName, final long creationTime, final int dataRegionId) {
+    final HistoricalTransferSummaryCounter counter =
+        HISTORICAL_TRANSFER_SUMMARY_COUNTER_MAP.remove(
+            new HistoricalTransferKey(pipeName, creationTime, dataRegionId));
+    return Objects.nonNull(counter) ? counter.snapshot() : null;
+  }
+
+  private static HistoricalTransferSummaryCounter getOrCreateHistoricalTransferSummaryCounter(
+      final String pipeName, final long creationTime, final int dataRegionId) {
+    return HISTORICAL_TRANSFER_SUMMARY_COUNTER_MAP.computeIfAbsent(
+        new HistoricalTransferKey(pipeName, creationTime, dataRegionId),
+        ignored -> new HistoricalTransferSummaryCounter());
+  }
+
+  public static final class HistoricalTransferSummary {
+
+    private final long extractedHistoricalTsFileCount;
+    private final long skippedHistoricalTsFileCount;
+    private final long splitHistoricalTsFileCount;
+    private final long unsplitHistoricalTsFileCount;
+    private final long extractedHistoricalDeletionCount;
+
+    private HistoricalTransferSummary(
+        final long extractedHistoricalTsFileCount,
+        final long skippedHistoricalTsFileCount,
+        final long splitHistoricalTsFileCount,
+        final long unsplitHistoricalTsFileCount,
+        final long extractedHistoricalDeletionCount) {
+      this.extractedHistoricalTsFileCount = extractedHistoricalTsFileCount;
+      this.skippedHistoricalTsFileCount = skippedHistoricalTsFileCount;
+      this.splitHistoricalTsFileCount = splitHistoricalTsFileCount;
+      this.unsplitHistoricalTsFileCount = unsplitHistoricalTsFileCount;
+      this.extractedHistoricalDeletionCount = extractedHistoricalDeletionCount;
+    }
+
+    public String toReportMessage() {
+      return String.format(
+          "historical summary: extractedTsFileCount=%s, skippedTsFileCount=%s, splitTsFileCount=%s, unsplitTsFileCount=%s, deletionCount=%s",
+          extractedHistoricalTsFileCount,
+          skippedHistoricalTsFileCount,
+          splitHistoricalTsFileCount,
+          unsplitHistoricalTsFileCount,
+          extractedHistoricalDeletionCount);
+    }
+  }
+
+  private static final class HistoricalTransferSummaryCounter {
+
+    private final AtomicLong extractedHistoricalTsFileCount = new AtomicLong(0);
+    private final AtomicLong skippedHistoricalTsFileCount = new AtomicLong(0);
+    private final AtomicLong splitHistoricalTsFileCount = new AtomicLong(0);
+    private final AtomicLong unsplitHistoricalTsFileCount = new AtomicLong(0);
+    private final AtomicLong extractedHistoricalDeletionCount = new AtomicLong(0);
+
+    private void initialize(
+        final long extractedHistoricalTsFileCount, final long extractedHistoricalDeletionCount) {
+      this.extractedHistoricalTsFileCount.set(extractedHistoricalTsFileCount);
+      this.skippedHistoricalTsFileCount.set(0);
+      this.splitHistoricalTsFileCount.set(0);
+      this.unsplitHistoricalTsFileCount.set(0);
+      this.extractedHistoricalDeletionCount.set(extractedHistoricalDeletionCount);
+    }
+
+    private HistoricalTransferSummary snapshot() {
+      return new HistoricalTransferSummary(
+          extractedHistoricalTsFileCount.get(),
+          skippedHistoricalTsFileCount.get(),
+          splitHistoricalTsFileCount.get(),
+          unsplitHistoricalTsFileCount.get(),
+          extractedHistoricalDeletionCount.get());
+    }
+  }
+
+  private static final class HistoricalTransferKey {
+
+    private final String pipeName;
+    private final long creationTime;
+    private final int dataRegionId;
+
+    private HistoricalTransferKey(
+        final String pipeName, final long creationTime, final int dataRegionId) {
+      this.pipeName = pipeName;
+      this.creationTime = creationTime;
+      this.dataRegionId = dataRegionId;
+    }
+
+    @Override
+    public boolean equals(final Object obj) {
+      if (this == obj) {
+        return true;
+      }
+      if (!(obj instanceof HistoricalTransferKey)) {
+        return false;
+      }
+      final HistoricalTransferKey that = (HistoricalTransferKey) obj;
+      return creationTime == that.creationTime
+          && dataRegionId == that.dataRegionId
+          && Objects.equals(pipeName, that.pipeName);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(pipeName, creationTime, dataRegionId);
+    }
+  }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java

@@ -168,6 +168,8 @@
       new HashMap<>();
   private final Map<PersistentResource, Long> pendingResource2ReplicateIndexForIoTV2 =
       new HashMap<>();
+  private int extractedHistoricalTsFileCount = 0;
+  private int extractedHistoricalDeletionCount = 0;
 
   @Override
   public void validate(final PipeParameterValidator validator) {
@@ -488,6 +490,8 @@
       return;
     }
     hasBeenStarted = true;
+    extractedHistoricalTsFileCount = 0;
+    extractedHistoricalDeletionCount = 0;
 
     final DataRegion dataRegion =
         StorageEngine.getInstance().getDataRegion(new DataRegionId(dataRegionId));
@@ -521,6 +525,12 @@
                   ? Long.compare(o1.getFileStartTime(), o2.getFileStartTime())
                   : o1.getProgressIndex().topologicalCompareTo(o2.getProgressIndex()));
       pendingQueue = new ArrayDeque<>(originalResourceList);
+      PipeTerminateEvent.initializeHistoricalTransferSummary(
+          pipeName,
+          creationTime,
+          dataRegionId,
+          extractedHistoricalTsFileCount,
+          extractedHistoricalDeletionCount);
 
       LOGGER.info(
           DataNodePipeMessages.PIPE_FINISH_TO_SORT_ALL_EXTRACTED_RESOURCES,
@@ -649,6 +659,7 @@
                   return true;
                 }
               });
+      extractedHistoricalTsFileCount = filteredTsFileResources2TableNames.size();
 
       LOGGER.info(
           "Pipe {}@{}: finish to extract historical TsFile, extracted sequence file count {}/{}, "
@@ -798,6 +809,7 @@
                 })
             .collect(Collectors.toList());
     resourceList.addAll(allDeletionResources);
+    extractedHistoricalDeletionCount = allDeletionResources.size();
     LOGGER.info(
         DataNodePipeMessages.PIPE_FINISH_TO_EXTRACT_DELETIONS_EXTRACT_DELETIONS,
         pipeName,
@@ -841,6 +853,16 @@
   }
 
   private Event supplyTerminateEvent() {
+    final PipeTerminateEvent.HistoricalTransferSummary historicalTransferSummary =
+        PipeTerminateEvent.snapshotHistoricalTransferSummary(pipeName, creationTime, dataRegionId);
+    if (Objects.nonNull(historicalTransferSummary)) {
+      LOGGER.info(
+          "Pipe {}@{}: historical source has supplied all events, emitting terminate event. {}",
+          pipeName,
+          dataRegionId,
+          historicalTransferSummary.toReportMessage());
+    }
+
     final PipeTerminateEvent terminateEvent =
         new PipeTerminateEvent(
             pipeName,
@@ -867,6 +889,7 @@
     }
 
     filteredTsFileResources2TableNames.remove(resource);
+    PipeTerminateEvent.markHistoricalTsFileSkipped(pipeName, creationTime, dataRegionId);
     LOGGER.info(
         DataNodePipeMessages.PIPE_SKIP_HISTORICAL_TSFILE_BECAUSE_REALTIME_SOURCE,
         pipeName,
@@ -1074,6 +1097,9 @@
 
   @Override
   public synchronized void close() {
+    if (!isTerminateSignalSent) {
+      PipeTerminateEvent.clearHistoricalTransferSummary(pipeName, creationTime, dataRegionId);
+    }
     if (Objects.nonNull(pendingQueue)) {
       pendingQueue.forEach(
           resource -> {