This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new 0c83c1b693c [Pipe] Restore sink progress under memory pressure
(#18640) (#18649)
0c83c1b693c is described below
commit 0c83c1b693c3a32889fe1b96a1ae5c91fc6b5a13
Author: Caideyipi <[email protected]>
AuthorDate: Thu Sep 17 11:13:30 2026 +0800
[Pipe] Restore sink progress under memory pressure (#18640) (#18649)
---
.../evolvable/batch/PipeTabletEventBatch.java | 7 +++---
.../batch/PipeTransferBatchReqBuilderTest.java | 25 ++++++++++++++++++++++
.../task/subtask/PipeAbstractSinkSubtask.java | 3 +--
.../commons/pipe/task/PipeSleepIntervalTest.java | 17 +++++++++++++++
4 files changed, 46 insertions(+), 6 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java
index de6055facfd..2e93a372349 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java
@@ -23,8 +23,8 @@ import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalExc
import org.apache.iotdb.commons.pipe.agent.task.progress.CommitterKey;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
+import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock;
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager;
-import org.apache.iotdb.db.pipe.resource.memory.PipeTabletMemoryBlock;
import
org.apache.iotdb.db.pipe.sink.protocol.thrift.async.IoTDBDataRegionAsyncSink;
import
org.apache.iotdb.db.storageengine.dataregion.wal.exception.WALPipeException;
import org.apache.iotdb.pipe.api.event.Event;
@@ -50,7 +50,7 @@ public abstract class PipeTabletEventBatch implements
AutoCloseable {
private long firstEventProcessingTime = Long.MIN_VALUE;
protected long totalBufferSize = 0;
- private final PipeTabletMemoryBlock allocatedMemoryBlock;
+ private final PipeMemoryBlock allocatedMemoryBlock;
private boolean shouldEmitOnMemoryPressure = false;
protected volatile boolean isClosed = false;
@@ -63,8 +63,7 @@ public abstract class PipeTabletEventBatch implements
AutoCloseable {
// limit in buffer size
this.maxBatchSizeInBytes = requestMaxBatchSizeInBytes;
- this.allocatedMemoryBlock =
-
PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(0);
+ this.allocatedMemoryBlock =
PipeDataNodeResourceManager.memory().forceAllocate(0);
if (recordMetric != null) {
this.recordMetric = recordMetric;
} else {
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilderTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilderTest.java
index 80d284cf0e6..bfa8f9cd9aa 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilderTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilderTest.java
@@ -22,6 +22,8 @@ package org.apache.iotdb.db.pipe.sink.payload.evolvable.batch;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
+import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
+import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
@@ -94,6 +96,29 @@ public class PipeTransferBatchReqBuilderTest {
}
}
+ @Test
+ public void testBatchMemoryIsNotCountedAsTabletMemory() throws Exception {
+ final PipeMemoryManager memoryManager =
PipeDataNodeResourceManager.memory();
+ final PipeRawTabletInsertionEvent event = createEvent(1);
+
+ try {
+ Assert.assertTrue(event.increaseReferenceCount(getClass().getName()));
+ final long tabletMemoryBeforeBatch =
memoryManager.getUsedMemorySizeInBytesOfTablets();
+ final long totalMemoryBeforeBatch =
memoryManager.getUsedMemorySizeInBytes();
+
+ try (final PipeTabletEventBatch batch =
+ new PipeTabletEventPlainBatch(Integer.MAX_VALUE, Long.MAX_VALUE,
null)) {
+ batch.onEvent(event);
+
+ Assert.assertEquals(
+ tabletMemoryBeforeBatch,
memoryManager.getUsedMemorySizeInBytesOfTablets());
+ Assert.assertTrue(memoryManager.getUsedMemorySizeInBytes() >
totalMemoryBeforeBatch);
+ }
+ } finally {
+ event.clearReferenceCount(getClass().getName());
+ }
+ }
+
private static PipeRawTabletInsertionEvent createEvent(final int value) {
final Tablet tablet =
new Tablet(
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java
index 2b4f807caa4..53415385f8b 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java
@@ -333,9 +333,8 @@ public abstract class PipeAbstractSinkSubtask extends
PipeReportableSubtask {
LOGGER::info,
e,
"Temporarily out of memory in pipe event transferring, will wait
for the memory to release.");
- } else {
- sleep4NonReportException();
}
+ sleep4NonReportException();
} else if (e instanceof PipeRuntimeSinkNonReportTimeConfigurableException)
{
if (lastExceptionTime == Long.MAX_VALUE) {
lastExceptionTime = System.currentTimeMillis();
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/task/PipeSleepIntervalTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/task/PipeSleepIntervalTest.java
index a9b72391b2b..015279acc98 100644
---
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/task/PipeSleepIntervalTest.java
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/task/PipeSleepIntervalTest.java
@@ -22,9 +22,11 @@ package org.apache.iotdb.commons.pipe.task;
import org.apache.iotdb.commons.conf.CommonConfig;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException;
+import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException;
import
org.apache.iotdb.commons.pipe.agent.task.subtask.PipeAbstractSinkSubtask;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
+import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.exception.PipeConnectionException;
import org.junit.After;
@@ -65,6 +67,10 @@ public class PipeSleepIntervalTest {
void sleepWithoutHighPriorityTask(final long sleepMillis) throws
InterruptedException {
sleepIfNoHighPriorityTask(sleepMillis);
}
+
+ void handle(final Event event, final Exception exception) {
+ handleException(event, exception);
+ }
}
private long oldPipeSinkSubtaskSleepIntervalInitMs;
@@ -136,4 +142,15 @@ public class PipeSleepIntervalTest {
Assert.assertTrue(System.currentTimeMillis() - startTime >= 15L);
}
}
+
+ @Test
+ public void testMemoryTimeoutRetryWaits() {
+ try (final TestSinkSubtask subtask = new TestSinkSubtask()) {
+ final long startTime = System.currentTimeMillis();
+ subtask.handle(null, new PipeRuntimeOutOfMemoryCriticalException("memory
unavailable"));
+ Assert.assertTrue(
+ System.currentTimeMillis() - startTime
+ >=
PipeConfig.getInstance().getPipeSinkSubtaskSleepIntervalInitMs());
+ }
+ }
}
