Pipe:Fix DataNodeShutdownHook waiting report logic and add capture history file log (#15952)

* Pipe:Fix DataNodeShutdownHook waiting report logic and add capture history file log

* fix

Author: luoluoyuyu

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

@@ -703,13 +703,23 @@ private boolean mayTsFileContainUnprocessedData(final TsFileResource resource) {
       // instead of replication or something else.
       ProgressIndex dedicatedProgressIndex =
           tryToExtractLocalProgressIndexForIoTV2(resource.getMaxProgressIndexAfterClose());
-      return greaterThanStartIndex(dedicatedProgressIndex);
+      return greaterThanStartIndex(resource, dedicatedProgressIndex);
     }
-    return greaterThanStartIndex(resource.getMaxProgressIndexAfterClose());
+    return greaterThanStartIndex(resource, resource.getMaxProgressIndexAfterClose());
   }
 
-  private boolean greaterThanStartIndex(ProgressIndex progressIndex) {
-    return !startIndex.isAfter(progressIndex) && !startIndex.equals(progressIndex);
+  private boolean greaterThanStartIndex(PersistentResource resource, ProgressIndex progressIndex) {
+    if (!startIndex.isAfter(progressIndex) && !startIndex.equals(progressIndex)) {
+      LOGGER.info(
+          "Pipe {}@{}: resource {} meets mayTsFileContainUnprocessedData condition, extractor progressIndex: {}, resource ProgressIndex: {}",
+          pipeName,
+          dataRegionId,
+          resource,
+          startIndex,
+          progressIndex);
+      return true;
+    }
+    return false;
   }
 
   private boolean mayTsFileResourceOverlappedWithPattern(final TsFileResource resource) {
@@ -809,7 +819,7 @@ private void extractDeletions(
               if (pipeName.startsWith(PipeStaticMeta.CONSENSUS_PIPE_PREFIX)) {
                 toBeCompared = tryToExtractLocalProgressIndexForIoTV2(toBeCompared);
               }
-              return !greaterThanStartIndex(toBeCompared);
+              return !greaterThanStartIndex(resource, toBeCompared);
             })
         .forEach(DeletionResource::decreaseReference);
     // Get deletions that should be sent.
@@ -821,7 +831,7 @@ private void extractDeletions(
                   if (pipeName.startsWith(PipeStaticMeta.CONSENSUS_PIPE_PREFIX)) {
                     toBeCompared = tryToExtractLocalProgressIndexForIoTV2(toBeCompared);
                   }
-                  return greaterThanStartIndex(toBeCompared);
+                  return greaterThanStartIndex(resource, toBeCompared);
                 })
             .collect(Collectors.toList());
     resourceList.addAll(allDeletionResources);

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java

@@ -128,7 +128,7 @@ public void run() {
           PipeDataNodeSinglePipeMetrics.getInstance().remainingEventAndTimeOperatorMap.entrySet()) {
         boolean timeout = false;
         while (true) {
-          if (entry.getValue().getRemainingNonHeartbeatEvents() > 0) {
+          if (entry.getValue().getRemainingNonHeartbeatEvents() == 0) {
             logger.info(
                 "Successfully waited for pipe {} to finish.", entry.getValue().getPipeName());
             break;