Pipe: Exclude the tsFiles / insertNodes from transfer time metric which have not be sent (#16015)

Author: Caideyipi

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java

@@ -211,7 +211,10 @@ public boolean internallyDecreaseResourceReferenceCount(final String holderMessa
         PipeDataNodeAgent.task()
             .decreaseFloatingMemoryUsageInByte(pipeName, creationTime, ramBytesUsed());
         PipeDataNodeSinglePipeMetrics.getInstance()
-            .decreaseInsertNodeEventCount(pipeName, creationTime, System.nanoTime() - extractTime);
+            .decreaseInsertNodeEventCount(
+                pipeName,
+                creationTime,
+                shouldReportOnCommit ? System.nanoTime() - extractTime : -1);
       }
       insertNode = null;
     }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java

@@ -337,7 +337,10 @@ public boolean internallyDecreaseResourceReferenceCount(final String holderMessa
     } finally {
       if (Objects.nonNull(pipeName)) {
         PipeDataNodeSinglePipeMetrics.getInstance()
-            .decreaseTsFileEventCount(pipeName, creationTime, System.nanoTime() - extractTime);
+            .decreaseTsFileEventCount(
+                pipeName,
+                creationTime,
+                shouldReportOnCommit ? System.nanoTime() - extractTime : -1);
       }
     }
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeSinglePipeMetrics.java

@@ -269,8 +269,10 @@ public void decreaseInsertNodeEventCount(
             k -> new PipeDataNodeRemainingEventAndTimeOperator(pipeName, creationTime));
     operator.decreaseInsertNodeEventCount();
 
-    operator.getInsertNodeTransferTimer().update(transferTime, TimeUnit.NANOSECONDS);
-    PIPE_DATANODE_INSERTNODE_TRANSFER_TIME_HISTOGRAM.update(transferTime);
+    if (transferTime > 0) {
+      operator.getInsertNodeTransferTimer().update(transferTime, TimeUnit.NANOSECONDS);
+      PIPE_DATANODE_INSERTNODE_TRANSFER_TIME_HISTOGRAM.update(transferTime);
+    }
   }
 
   public void increaseRawTabletEventCount(final String pipeName, final long creationTime) {
@@ -305,8 +307,10 @@ public void decreaseTsFileEventCount(
             k -> new PipeDataNodeRemainingEventAndTimeOperator(pipeName, creationTime));
 
     operator.decreaseTsFileEventCount();
-    operator.getTsFileTransferTimer().update(transferTime, TimeUnit.NANOSECONDS);
-    PIPE_DATANODE_TSFILE_TRANSFER_TIME_HISTOGRAM.update(transferTime);
+    if (transferTime > 0) {
+      operator.getTsFileTransferTimer().update(transferTime, TimeUnit.NANOSECONDS);
+      PIPE_DATANODE_TSFILE_TRANSFER_TIME_HISTOGRAM.update(transferTime);
+    }
   }
 
   public void increaseHeartbeatEventCount(final String pipeName, final long creationTime) {

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/EnrichedEvent.java

@@ -202,16 +202,16 @@ public synchronized boolean decreaseReferenceCount(
     }
 
     if (referenceCount.get() == 1) {
+      if (!shouldReport) {
+        shouldReportOnCommit = false;
+      }
       // We assume that this function will not throw any exceptions.
       if (!internallyDecreaseResourceReferenceCount(holderMessage)) {
         LOGGER.warn(
             "resource reference count is decreased to 0, but failed to release the resource, EnrichedEvent: {}, stack trace: {}",
             coreReportMessage(),
             Thread.currentThread().getStackTrace());
       }
-      if (!shouldReport) {
-        shouldReportOnCommit = false;
-      }
       PipeEventCommitManager.getInstance().commit(this, committerKey);
     }