Pipe: Optimized the memory occupation of pipe realtime source

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java

@@ -668,7 +668,6 @@ private void retryTransfer(final PipeTsFileInsertionEvent tsFileInsertionEvent)
         addFailureEventToRetryQueue(tsFileInsertionEvent, null);
       }
     } catch (final Exception e) {
-      tsFileInsertionEvent.decreaseReferenceCount(IoTDBDataRegionAsyncSink.class.getName(), false);
       addFailureEventToRetryQueue(tsFileInsertionEvent, e);
     }
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java

@@ -137,7 +137,7 @@
     return tsFile;
   }
 
   public void transfer(
       final IoTDBDataNodeAsyncClientManager clientManager,
       final AsyncPipeDataTransferServiceClient client)
       throws TException, IOException {
@@ -153,10 +153,7 @@
     }
 
     if (reader == null) {
-      reader =
-          Objects.nonNull(modFile)
-              ? new RandomAccessFile(modFile, "r")
-              : new RandomAccessFile(tsFile, "r");
+      reader = transferMod ? new RandomAccessFile(modFile, "r") : new RandomAccessFile(tsFile, "r");
     }
 
     this.clientManager = clientManager;

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java

@@ -427,6 +427,13 @@ protected void extractHeartbeat(final PipeRealtimeEvent event) {
   }
 
   protected void extractProgressReportEvent(final PipeRealtimeEvent event) {
+    // Remove any heartbeat events in front of this event to avoid OOM
+    // Since the batch and retry queue no longer need the heartbeat event to trigger
+    // And the progress report event can trigger the processor calculation because it's not reported
+    // yet
+    while (((PipeRealtimeEvent) pendingQueue.peekLast()).getEvent() instanceof PipeHeartbeatEvent) {
+      pendingQueue.pollLast();
+    }
     if (pendingQueue.peekLast() instanceof ProgressReportEvent) {
       final ProgressReportEvent oldEvent = (ProgressReportEvent) pendingQueue.peekLast();
       oldEvent.bindProgressIndex(

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

@@ -37,4 +37,8 @@ public UnboundedBlockingPendingQueue(final PipeEventCounter eventCounter) {
   public E peekLast() {
     return pendingDeque.peekLast();
   }
+
+  public E pollLast() {
+    return pendingDeque.pollLast();
+  }
 }

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java

@@ -766,7 +766,7 @@
   }
 
   private TPipeTransferResp checkNonFinalFileSeal(
       final File file, final String fileName, final long fileLength) throws IOException {
     if (!file.exists()) {
       final TSStatus status =
           RpcUtils.getStatus(
@@ -787,15 +787,15 @@
               String.format(
                   "Failed to seal file %s, because the length of file is not correct. "
                       + "The original file has length %s, but receiver file has length %s.",
-                  fileName, fileLength, writingFileWriter.length()));
+                  fileName, fileLength, file.length()));
       PipeLogger.log(
           LOGGER::warn,
           "Receiver id = %s: Failed to seal file %s, because the length of file is not correct. "
               + "The original file has length %s, but receiver file has length %s.",
           receiverId.get(),
           fileName,
           fileLength,
-          writingFileWriter.length());
+          file.length());
       return new TPipeTransferResp(status);
     }