You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ju...@apache.org on 2026/10/22 11:06:15 UTC
(iotdb) 01/01: [Pipe] Unify TsFile parser allocation and resize quotas (#18984)
This is an automated email from the ASF dual-hosted git repository.
Caideyipi pushed a commit to branch cp-b626c602-dev-1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 59f37967386900f93789cfae11cfd6bf10a3e6a5
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Thu Oct 22 16:29:42 2026 +0800
[Pipe] Unify TsFile parser allocation and resize quotas (#18984)
(cherry picked from commit b626c602ead0ba8d96744eac8324d948c19a4e30)
---
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 6 +
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 6 +
.../TsFileInsertionEventParserMemoryManager.java | 15 +--
.../pipe/resource/memory/PipeMemoryBlockType.java | 1 +
.../db/pipe/resource/memory/PipeMemoryManager.java | 71 ++++++++--
.../memory/PipeTsFileParserMemoryBlock.java | 27 ++++
.../memory/PipeMemoryManagerResizeTest.java | 145 +++++++++++++++++++++
7 files changed, 254 insertions(+), 17 deletions(-)
diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index db51ff09b00..978279cc6d9 100644
--- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -23,6 +23,12 @@ public final class DataNodePipeMessages {
private DataNodePipeMessages() {}
+ public static final String
+ EXCEPTION_FORCERESIZE_FAILED_AFTER_ARG_RETRIES_TOTAL_ARG_BYTES_USED_ARG_BYTES_USEDTABLETS_ARG_BYTES_USEDTSFILES_ARG_BYTES_TABLETLIMIT_ARG_BYTES_TSFILELIMIT_ARG_BYTES_TABLETANDTSFILELIMIT_ARG_BYTES_PROTECTEDPARSERLIMIT_ARG_BYTES_REQUESTED_ARG_BYTES_311C8C7E =
+ "forceResize failed after %d retries: total=%d bytes, used=%d bytes, "
+ + "usedTablets=%d bytes, usedTsFiles=%d bytes, tabletLimit=%d bytes, tsFileLimit=%d bytes, "
+ + "tabletAndTsFileLimit=%d bytes, protectedParserLimit=%d bytes, requested=%d bytes";
+
// ===================== LEGACY PIPE RECEIVER RESPONSES =====================
public static final String MESSAGE_PIPE_DATA_TRANSPORT_ERROR_ARG_ECE6A238 =
diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index 053fc9cf69f..e3e72db09fb 100644
--- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -23,6 +23,12 @@ public final class DataNodePipeMessages {
private DataNodePipeMessages() {}
+ public static final String
+ EXCEPTION_FORCERESIZE_FAILED_AFTER_ARG_RETRIES_TOTAL_ARG_BYTES_USED_ARG_BYTES_USEDTABLETS_ARG_BYTES_USEDTSFILES_ARG_BYTES_TABLETLIMIT_ARG_BYTES_TSFILELIMIT_ARG_BYTES_TABLETANDTSFILELIMIT_ARG_BYTES_PROTECTEDPARSERLIMIT_ARG_BYTES_REQUESTED_ARG_BYTES_311C8C7E =
+ "forceResize 重试 %d 次后仍无法分配内存：total=%d bytes，used=%d bytes，"
+ + "usedTablets=%d bytes，usedTsFiles=%d bytes，tabletLimit=%d bytes，tsFileLimit=%d bytes，"
+ + "tabletAndTsFileLimit=%d bytes，protectedParserLimit=%d bytes，requested=%d bytes";
+
// ===================== LEGACY PIPE RECEIVER RESPONSES =====================
public static final String MESSAGE_PIPE_DATA_TRANSPORT_ERROR_ARG_ECE6A238 = "Pipe 数据传输错误：%s";
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserMemoryManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserMemoryManager.java
index abb1dfebe41..5b901ce475a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserMemoryManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserMemoryManager.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.db.pipe.event.common.tsfile.parser;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException;
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;
/** Allocates parser working memory from the pool owned by the caller. */
public interface TsFileInsertionEventParserMemoryManager {
@@ -41,7 +42,8 @@ public interface TsFileInsertionEventParserMemoryManager {
public TsFileInsertionEventParserMemoryBlock forceAllocateForTabletWithRetry(
final long sizeInBytes) {
return new PipeBlock(
- PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(sizeInBytes));
+ PipeDataNodeResourceManager.memory()
+ .forceAllocateForTsFileParserWithRetry(sizeInBytes));
}
@Override
@@ -66,15 +68,10 @@ public interface TsFileInsertionEventParserMemoryManager {
@Override
public void forceResize(final long newSizeInBytes) {
final long oldSize = delegate.getMemoryUsageInBytes();
- if (!PipeDataNodeResourceManager.memory().tryResize(delegate, newSizeInBytes)) {
+ final PipeMemoryManager memoryManager = PipeDataNodeResourceManager.memory();
+ if (!memoryManager.tryResize(delegate, newSizeInBytes)) {
throw new PipeRuntimeOutOfMemoryCriticalException(
- String.format(
- "forceResize: failed to allocate memory after %d retries, total memory size %d "
- + "bytes, used memory size %d bytes, requested memory size %d bytes",
- 0,
- PipeDataNodeResourceManager.memory().getTotalNonFloatingMemorySizeInBytes(),
- PipeDataNodeResourceManager.memory().getUsedMemorySizeInBytes(),
- newSizeInBytes - oldSize));
+ memoryManager.getResizeFailureMessage(0, newSizeInBytes - oldSize));
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryBlockType.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryBlockType.java
index 846fc7dd1ce..78fd9aff670 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryBlockType.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryBlockType.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.db.pipe.resource.memory;
public enum PipeMemoryBlockType {
NORMAL,
TABLET,
+ TS_FILE_PARSER,
TS_FILE,
BATCH,
WAL
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java
index 8fc3d49536a..a41e1d86cb1 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.db.pipe.resource.memory;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
import org.apache.iotdb.db.pipe.metric.overview.PipeResourceMetrics;
import org.apache.iotdb.db.pipe.resource.memory.strategy.ThresholdAllocationStrategy;
@@ -132,6 +133,35 @@ public class PipeMemoryManager {
* totalNonFloatingMemorySizeInBytes;
}
+ private boolean isEnoughForTsFileParserMemory(
+ final long extraMemoryInBytes, final long totalNonFloatingMemorySizeInBytes) {
+ return (double) usedMemorySizeInBytesOfTablets
+ + (double) usedMemorySizeInBytesOfTsFiles
+ + (double) extraMemoryInBytes
+ < EXCEED_PROTECT_THRESHOLD
+ * allowedMaxMemorySizeInBytesOfTabletsAndTsFiles(totalNonFloatingMemorySizeInBytes);
+ }
+
+ /** Reports both global availability and the category limits that can reject a resize. */
+ public synchronized String getResizeFailureMessage(final int retries, final long requestedSize) {
+ final long totalNonFloatingMemorySizeInBytes = getTotalNonFloatingMemorySizeInBytes();
+ final double combinedLimit =
+ allowedMaxMemorySizeInBytesOfTabletsAndTsFiles(totalNonFloatingMemorySizeInBytes);
+ return String.format(
+ DataNodePipeMessages
+ .EXCEPTION_FORCERESIZE_FAILED_AFTER_ARG_RETRIES_TOTAL_ARG_BYTES_USED_ARG_BYTES_USEDTABLETS_ARG_BYTES_USEDTSFILES_ARG_BYTES_TABLETLIMIT_ARG_BYTES_TSFILELIMIT_ARG_BYTES_TABLETANDTSFILELIMIT_ARG_BYTES_PROTECTEDPARSERLIMIT_ARG_BYTES_REQUESTED_ARG_BYTES_311C8C7E,
+ retries,
+ totalNonFloatingMemorySizeInBytes,
+ usedMemorySizeInBytes,
+ usedMemorySizeInBytesOfTablets,
+ usedMemorySizeInBytesOfTsFiles,
+ Math.round(allowedMaxMemorySizeInBytesOfTablets(totalNonFloatingMemorySizeInBytes)),
+ Math.round(allowedMaxMemorySizeInBytesOfTsTiles(totalNonFloatingMemorySizeInBytes)),
+ Math.round(combinedLimit),
+ Math.round(EXCEED_PROTECT_THRESHOLD * combinedLimit),
+ requestedSize);
+ }
+
private static long getTsFileParserMemorySizeInBytes() {
return Math.max(
PipeConfig.getInstance().getTsFileParserMemory(), MEMORY_ALLOCATE_MIN_SIZE_IN_BYTES);
@@ -537,6 +567,10 @@ public class PipeMemoryManager {
private boolean isHardEnoughForResizing(
final PipeMemoryBlock block, final long extraMemoryInBytes) {
+ if (block instanceof PipeTsFileParserMemoryBlock) {
+ return isEnoughForTsFileParserMemory(
+ extraMemoryInBytes, getTotalNonFloatingMemorySizeInBytes());
+ }
if (!(block instanceof PipeTabletMemoryBlock) && !(block instanceof PipeTsFileMemoryBlock)) {
return true;
}
@@ -663,6 +697,28 @@ public class PipeMemoryManager {
}
}
+ /** Allocates TsFile parser memory against the protected tablet + TsFile combined quota. */
+ public synchronized PipeTsFileParserMemoryBlock forceAllocateForTsFileParserWithRetry(
+ final long parserMemorySizeInBytes) {
+ if (!PIPE_MEMORY_MANAGEMENT_ENABLED) {
+ return new PipeTsFileParserMemoryBlock(0);
+ }
+
+ final PipeTsFileParserMemoryBlock block =
+ (PipeTsFileParserMemoryBlock) registerMemoryBlock(0, PipeMemoryBlockType.TS_FILE_PARSER);
+ if (parserMemorySizeInBytes == 0) {
+ return block;
+ }
+
+ try {
+ forceResize(block, parserMemorySizeInBytes);
+ return block;
+ } catch (final RuntimeException e) {
+ release(block);
+ throw e;
+ }
+ }
+
public PipeModelFixedMemoryBlock forceAllocateForModelFixedMemoryBlock(
long fixedSizeInBytes, PipeMemoryBlockType type)
throws PipeRuntimeOutOfMemoryCriticalException {
@@ -790,18 +846,14 @@ public class PipeMemoryManager {
if (force) {
throw new PipeRuntimeOutOfMemoryCriticalException(
- String.format(
- "forceResize: failed to allocate memory after %d retries, "
- + "total memory size %d bytes, used memory size %d bytes, "
- + "requested memory size %d bytes",
- memoryAllocateMaxRetries,
- getTotalNonFloatingMemorySizeInBytes(),
- usedMemorySizeInBytes,
- sizeInBytes));
+ getResizeFailureMessage(memoryAllocateMaxRetries, sizeInBytes));
}
}
private boolean tryResizeInternal(final PipeMemoryBlock block, final long targetSize) {
+ if (block.isReleased()) {
+ return false;
+ }
if (!PIPE_MEMORY_MANAGEMENT_ENABLED) {
block.setMemoryUsageInBytes(targetSize);
return true;
@@ -977,6 +1029,9 @@ public class PipeMemoryManager {
case TABLET:
returnedMemoryBlock = new PipeTabletMemoryBlock(sizeInBytes);
break;
+ case TS_FILE_PARSER:
+ returnedMemoryBlock = new PipeTsFileParserMemoryBlock(sizeInBytes);
+ break;
case TS_FILE:
returnedMemoryBlock = new PipeTsFileMemoryBlock(sizeInBytes);
break;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeTsFileParserMemoryBlock.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeTsFileParserMemoryBlock.java
new file mode 100644
index 00000000000..5258d5216b5
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeTsFileParserMemoryBlock.java
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.iotdb.db.pipe.resource.memory;
+
+/** Parser working memory charged to TsFiles and admitted against the protected combined quota. */
+public class PipeTsFileParserMemoryBlock extends PipeTsFileMemoryBlock {
+
+ public PipeTsFileParserMemoryBlock(final long memoryUsageInBytes) {
+ super(memoryUsageInBytes);
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerResizeTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerResizeTest.java
index 08effc87e86..ecb0cf8fbe6 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerResizeTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerResizeTest.java
@@ -28,6 +28,13 @@ import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
public class PipeMemoryManagerResizeTest {
@@ -98,6 +105,144 @@ public class PipeMemoryManagerResizeTest {
}
}
+ @Test
+ public void testTsFileParserAllocationAndResizeShareProtectedCombinedLimit() {
+ final AtomicLong floatingMemoryUsageInBytes = new AtomicLong(0);
+ final PipeMemoryManager manager =
+ new PipeMemoryManager(TOTAL_MEMORY_SIZE_IN_BYTES, floatingMemoryUsageInBytes::get);
+ final PipeTsFileParserMemoryBlock parser = manager.forceAllocateForTsFileParserWithRetry(1000);
+
+ try {
+ Assert.assertTrue(manager.tryResize(parser, 1100));
+ Assert.assertFalse(manager.tryResize(parser, 1140));
+ Assert.assertEquals(1100, parser.getMemoryUsageInBytes());
+ Assert.assertEquals(0, manager.getUsedMemorySizeInBytesOfTablets());
+ Assert.assertEquals(1100, manager.getUsedMemorySizeInBytesOfTsFiles());
+ Assert.assertEquals(1100, manager.getUsedMemorySizeInBytes());
+
+ final PipeRuntimeOutOfMemoryCriticalException allocationException =
+ Assert.assertThrows(
+ PipeRuntimeOutOfMemoryCriticalException.class,
+ () -> manager.forceAllocateForTsFileParserWithRetry(40));
+ Assert.assertTrue(
+ allocationException.getMessage(),
+ allocationException.getMessage().contains("usedTsFiles=1100 bytes"));
+ Assert.assertTrue(
+ allocationException.getMessage(),
+ allocationException.getMessage().contains("tabletLimit=900 bytes"));
+ Assert.assertTrue(
+ allocationException.getMessage().contains("tabletAndTsFileLimit=1200 bytes"));
+ Assert.assertTrue(
+ allocationException.getMessage().contains("protectedParserLimit=1140 bytes"));
+ // Parser working memory must not consume the tablet-only quota: the generated tablet
+ // still needs to be retained and sent while the parser owns its decoded pages/chunks.
+ final PipeTabletMemoryBlock output = manager.forceAllocateForTabletWithRetry(50);
+ try {
+ Assert.assertTrue(manager.tryResize(output, 90));
+ Assert.assertEquals(90, manager.getUsedMemorySizeInBytesOfTablets());
+ Assert.assertEquals(1100, manager.getUsedMemorySizeInBytesOfTsFiles());
+ } finally {
+ manager.release(output);
+ }
+ } finally {
+ manager.release(parser);
+ }
+
+ Assert.assertEquals(0, manager.getUsedMemorySizeInBytes());
+ }
+
+ @Test
+ public void testZeroSizedTsFileParserCanGrowBeyondTabletLimitAndRelease() {
+ final PipeMemoryManager manager = new PipeMemoryManager(TOTAL_MEMORY_SIZE_IN_BYTES, () -> 0);
+ final PipeTsFileParserMemoryBlock parser = manager.forceAllocateForTsFileParserWithRetry(0);
+ try {
+ Assert.assertTrue(manager.tryResize(parser, 1000));
+ Assert.assertEquals(1000, manager.getUsedMemorySizeInBytesOfTsFiles());
+ Assert.assertEquals(1000, manager.getUsedMemorySizeInBytes());
+
+ Assert.assertTrue(manager.tryResize(parser, 100));
+ Assert.assertEquals(100, manager.getUsedMemorySizeInBytesOfTsFiles());
+ manager.release(parser);
+ Assert.assertTrue(parser.isReleased());
+ Assert.assertEquals(0, manager.getUsedMemorySizeInBytesOfTsFiles());
+ Assert.assertEquals(0, manager.getUsedMemorySizeInBytesOfTablets());
+ Assert.assertEquals(0, manager.getUsedMemorySizeInBytes());
+ } finally {
+ manager.release(parser);
+ }
+ }
+
+ @Test
+ public void testTsFileParserSharesQuotaWithTabletsAndTsFilesAndHonorsGlobalLimit() {
+ final PipeMemoryManager manager = new PipeMemoryManager(TOTAL_MEMORY_SIZE_IN_BYTES, () -> 0);
+ final PipeTabletMemoryBlock tablet = manager.forceAllocateForTabletWithRetry(200);
+ final PipeTsFileMemoryBlock tsFile = manager.forceAllocateForTsFileWithRetry(500);
+ final PipeTsFileParserMemoryBlock parser = manager.forceAllocateForTsFileParserWithRetry(400);
+ final PipeMemoryBlock sink = manager.forceAllocate(900);
+ try {
+ Assert.assertFalse(manager.tryResize(parser, 440));
+ Assert.assertFalse(manager.tryResize(parser, 401));
+ Assert.assertEquals(200, manager.getUsedMemorySizeInBytesOfTablets());
+ Assert.assertEquals(900, manager.getUsedMemorySizeInBytesOfTsFiles());
+ Assert.assertEquals(2000, manager.getUsedMemorySizeInBytes());
+
+ manager.release(sink);
+ Assert.assertTrue(manager.tryResize(parser, 439));
+ Assert.assertFalse(manager.tryResize(parser, 440));
+ Assert.assertEquals(200, manager.getUsedMemorySizeInBytesOfTablets());
+ Assert.assertEquals(939, manager.getUsedMemorySizeInBytesOfTsFiles());
+ Assert.assertEquals(1139, manager.getUsedMemorySizeInBytes());
+ } finally {
+ manager.release(tablet);
+ manager.release(tsFile);
+ manager.release(parser);
+ manager.release(sink);
+ }
+ Assert.assertEquals(0, manager.getUsedMemorySizeInBytesOfTablets());
+ Assert.assertEquals(0, manager.getUsedMemorySizeInBytesOfTsFiles());
+ Assert.assertEquals(0, manager.getUsedMemorySizeInBytes());
+ }
+
+ @Test
+ public void testConcurrentTsFileParserAllocationsCannotOvershootQuota() throws Exception {
+ final PipeMemoryManager manager = new PipeMemoryManager(TOTAL_MEMORY_SIZE_IN_BYTES, () -> 0);
+ final ExecutorService executor = Executors.newFixedThreadPool(4);
+ final CountDownLatch start = new CountDownLatch(1);
+ final List<Future<PipeTsFileParserMemoryBlock>> futures = new ArrayList<>();
+ final List<PipeTsFileParserMemoryBlock> allocated = new ArrayList<>();
+ try {
+ for (int i = 0; i < 20; i++) {
+ futures.add(
+ executor.submit(
+ () -> {
+ start.await();
+ try {
+ return manager.forceAllocateForTsFileParserWithRetry(300);
+ } catch (final PipeRuntimeOutOfMemoryCriticalException e) {
+ return null;
+ }
+ }));
+ }
+ start.countDown();
+ for (final Future<PipeTsFileParserMemoryBlock> future : futures) {
+ final PipeTsFileParserMemoryBlock block = future.get(10, TimeUnit.SECONDS);
+ if (block != null) {
+ allocated.add(block);
+ }
+ }
+ Assert.assertEquals(3, allocated.size());
+ Assert.assertEquals(900, manager.getUsedMemorySizeInBytes());
+ Assert.assertEquals(0, manager.getUsedMemorySizeInBytesOfTablets());
+ Assert.assertEquals(900, manager.getUsedMemorySizeInBytesOfTsFiles());
+ } finally {
+ executor.shutdownNow();
+ allocated.forEach(manager::release);
+ }
+ Assert.assertEquals(0, manager.getUsedMemorySizeInBytes());
+ Assert.assertEquals(0, manager.getUsedMemorySizeInBytesOfTsFiles());
+ Assert.assertEquals(0, manager.getUsedMemorySizeInBytesOfTablets());
+ }
+
@Test
public void testTabletResizeLeavesMemoryForSinkForwardProgress() {
final PipeMemoryManager manager = new PipeMemoryManager();