This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new b5199bd4832 Pipe: Optimized the memory occupation of pipe realtime
source (#17450)
b5199bd4832 is described below
commit b5199bd4832a571b5133d181fe13f4fd8a1f953a
Author: Caideyipi <[email protected]>
AuthorDate: Thu Apr 9 17:29:07 2026 +0800
Pipe: Optimized the memory occupation of pipe realtime source (#17450)
* fix
* fix
* fix
---
.../pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java | 1 -
.../protocol/thrift/async/handler/PipeTransferTsFileHandler.java | 5 +----
.../source/dataregion/realtime/PipeRealtimeDataRegionSource.java | 7 +++++++
.../pipe/agent/task/connection/UnboundedBlockingPendingQueue.java | 4 ++++
.../org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java | 4 ++--
5 files changed, 14 insertions(+), 7 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
index 1a4d7cf3863..d19639310cf 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
@@ -668,7 +668,6 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink {
addFailureEventToRetryQueue(tsFileInsertionEvent, null);
}
} catch (final Exception e) {
-
tsFileInsertionEvent.decreaseReferenceCount(IoTDBDataRegionAsyncSink.class.getName(),
false);
addFailureEventToRetryQueue(tsFileInsertionEvent, e);
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java
index 6726c9a6701..3eaaa94c416 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java
@@ -153,10 +153,7 @@ public class PipeTransferTsFileHandler extends
PipeTransferTrackableHandler {
}
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;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
index 1e47a48e500..718529243d9 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
@@ -427,6 +427,13 @@ public abstract class PipeRealtimeDataRegionSource
implements PipeExtractor {
}
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(
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/connection/UnboundedBlockingPendingQueue.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/connection/UnboundedBlockingPendingQueue.java
index 785e89cfb9a..43fa64c158e 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/connection/UnboundedBlockingPendingQueue.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/connection/UnboundedBlockingPendingQueue.java
@@ -37,4 +37,8 @@ public class UnboundedBlockingPendingQueue<E extends Event>
extends BlockingPend
public E peekLast() {
return pendingDeque.peekLast();
}
+
+ public E pollLast() {
+ return pendingDeque.pollLast();
+ }
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java
index 1105a21d8ac..8140348676a 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java
@@ -798,7 +798,7 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
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. "
@@ -806,7 +806,7 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
receiverId.get(),
fileName,
fileLength,
- writingFileWriter.length());
+ file.length());
return new TPipeTransferResp(status);
}
