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 106d7bf3c15 [Pipe] Resume TsFile parsing after retryable failures
(#18261) (#18283)
106d7bf3c15 is described below
commit 106d7bf3c15f1e4469f7e37fa0194596c07406b8
Author: Caideyipi <[email protected]>
AuthorDate: Thu Jul 23 09:35:32 2026 +0800
[Pipe] Resume TsFile parsing after retryable failures (#18261) (#18283)
* [Pipe] Resume TsFile parsing after retryable failures
* [Pipe] Address parser retry review feedback
(cherry picked from commit bb0f237271b985197f25ad2a999d8d48df7b20ea)
---
.../agent/task/connection/PipeEventCollector.java | 15 ++-
.../common/tsfile/PipeTsFileInsertionEvent.java | 104 ++++++++++++++++---
.../processor/aggregate/AggregateProcessor.java | 39 ++++---
.../downsampling/DownSamplingProcessor.java | 40 ++++---
.../sink/protocol/websocket/WebSocketSink.java | 21 ++--
.../event/TsFileInsertionDataContainerTest.java | 115 ++++++++++++---------
6 files changed, 208 insertions(+), 126 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
index f0cc4621612..df72ccb830d 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
@@ -139,15 +139,12 @@ public class PipeEventCollector implements EventCollector
{
return;
}
- try {
- sourceEvent.consumeTabletInsertionEventsWithRetry(
- this::collectParsedRawTableEvent,
"PipeEventCollector::parseAndCollectEvent");
- if (sourceEvent.isGeneratedByHistoricalExtractor()) {
- PipeTerminateEvent.markHistoricalTsFileSplit(
- sourceEvent.getPipeName(), sourceEvent.getCreationTime(),
regionId);
- }
- } finally {
- sourceEvent.close();
+ sourceEvent.consumeTabletInsertionEventsWithRetry(
+ this::collectParsedRawTableEvent,
"PipeEventCollector::parseAndCollectEvent");
+ sourceEvent.close();
+ if (sourceEvent.isGeneratedByHistoricalExtractor()) {
+ PipeTerminateEvent.markHistoricalTsFileSplit(
+ sourceEvent.getPipeName(), sourceEvent.getCreationTime(), regionId);
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
index 98c660ea351..046262dbd99 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
@@ -86,6 +86,13 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent
protected final AtomicBoolean isClosed;
protected final AtomicReference<TsFileInsertionDataContainer> dataContainer;
private final AtomicBoolean isTsFileParserMemoryReserved = new
AtomicBoolean(false);
+ private final AtomicReference<Iterator<TabletInsertionEvent>>
tabletInsertionEventIterator =
+ new AtomicReference<>();
+ private final AtomicReference<PipeRawTabletInsertionEvent>
pendingTabletInsertionEvent =
+ new AtomicReference<>();
+ private final AtomicInteger parsedTabletInsertionEventCount = new
AtomicInteger(0);
+ private final AtomicBoolean isTsFileParsingCompleted = new
AtomicBoolean(false);
+ private final AtomicLong parsedPointCountForCount = new AtomicLong(0);
// The point count of the TsFile. Used for metrics on PipeConsensus'
receiver side.
// May be updated after it is flushed. Should be negative if not set.
@@ -481,28 +488,73 @@ public class PipeTsFileInsertionEvent extends
EnrichedEvent
public void consumeTabletInsertionEventsWithRetry(
final TabletInsertionEventConsumer consumer, final String callerName)
throws Exception {
- int tabletEventCount = 0;
try {
- final Iterable<TabletInsertionEvent> iterable =
toTabletInsertionEvents();
- final Iterator<TabletInsertionEvent> iterator = iterable.iterator();
- while (iterator.hasNext()) {
- final TabletInsertionEvent parsedEvent = iterator.next();
- tabletEventCount++;
+ while (true) {
+ final PipeRawTabletInsertionEvent parsedEvent =
+ getNextTabletInsertionEventFromSavedProgress();
+ if (parsedEvent == null) {
+ isTsFileParsingCompleted.set(true);
+ releaseTsFileParserMemoryIfReserved();
+ return;
+ }
consumeParsedTabletInsertionEventWithRetry(
- consumer, callerName, tabletEventCount, parsedEvent);
+ consumer, callerName, parsedTabletInsertionEventCount.get(),
parsedEvent);
+ pendingTabletInsertionEvent.compareAndSet(parsedEvent, null);
}
} catch (final PipeRuntimeOutOfMemoryCriticalException e) {
- close();
+ // Yield the active parser slot to the next pipe while retaining the
iterator and current
+ // tablet. The next retry resumes from this exact tablet instead of
reparsing the TsFile.
+ releaseTsFileParserMemoryIfReserved();
LOGGER.warn(
"{}: failed to allocate memory for parsing TsFile {}, tablet event
no. {}, will release parser memory and retry the TsFile event later.",
callerName,
getTsFile(),
- tabletEventCount,
+ parsedTabletInsertionEventCount.get(),
e);
throw e;
+ } catch (final Exception e) {
+ releaseTsFileParserMemoryIfReserved();
+ throw e;
}
}
+ private PipeRawTabletInsertionEvent
getNextTabletInsertionEventFromSavedProgress()
+ throws Exception {
+ if (isTsFileParsingCompleted.get()) {
+ return null;
+ }
+
+ // Reacquire parser memory after a previous failure yielded the active
parser slot. This wait
+ // is already bounded to 20-40 seconds, while the exponential backoff
below is only for retrying
+ // the current tablet without yielding its parser slot.
+ waitForResourceEnough4Parsing((long) ((1 + Math.random()) * 20 * 1000));
+
+ final PipeRawTabletInsertionEvent pendingEvent =
pendingTabletInsertionEvent.get();
+ if (pendingEvent != null) {
+ return pendingEvent;
+ }
+
+ Iterator<TabletInsertionEvent> iterator =
tabletInsertionEventIterator.get();
+ if (iterator == null) {
+ if (!waitForTsFileClose()) {
+ LOGGER.warn(
+ "Pipe skipping temporary TsFile's parsing which shouldn't be
transferred: {}", tsFile);
+ return null;
+ }
+ iterator = initDataContainer().toTabletInsertionEvents().iterator();
+ tabletInsertionEventIterator.set(iterator);
+ }
+
+ if (!iterator.hasNext()) {
+ return null;
+ }
+
+ final PipeRawTabletInsertionEvent nextEvent =
(PipeRawTabletInsertionEvent) iterator.next();
+ pendingTabletInsertionEvent.set(nextEvent);
+ parsedTabletInsertionEventCount.incrementAndGet();
+ return nextEvent;
+ }
+
private void consumeParsedTabletInsertionEventWithRetry(
final TabletInsertionEventConsumer consumer,
final String callerName,
@@ -522,21 +574,34 @@ public class PipeTsFileInsertionEvent extends
EnrichedEvent
}
if (memoryManager.shouldReleaseTsFileParserOnOutOfMemory(
firstOutOfMemoryTimeInMs, ++retryCount)) {
- releaseParsedTabletEvent(parsedEvent);
throw e;
}
logParserRetryOnOutOfMemory(callerName, tabletEventCount, retryCount,
e);
try {
-
Thread.sleep(PipeConfig.getInstance().getPipeMemoryAllocateRetryIntervalInMs());
+ Thread.sleep(getParserRetryBackoffInMs(retryCount));
} catch (final InterruptedException interruptedException) {
Thread.currentThread().interrupt();
- releaseParsedTabletEvent(parsedEvent);
throw e;
}
}
}
}
+ private long getParserRetryBackoffInMs(final int retryCount) {
+ final long initialBackoffInMs =
+ Math.max(1,
PipeConfig.getInstance().getPipeMemoryAllocateRetryIntervalInMs());
+ final int maxRetries = Math.max(1,
PipeConfig.getInstance().getPipeMemoryAllocateMaxRetries());
+ final long maxBackoffInMs =
+ initialBackoffInMs > Long.MAX_VALUE / maxRetries
+ ? Long.MAX_VALUE
+ : initialBackoffInMs * maxRetries;
+ long backoffInMs = initialBackoffInMs;
+ for (int retry = 1; retry < retryCount && backoffInMs < maxBackoffInMs;
retry++) {
+ backoffInMs = backoffInMs >= maxBackoffInMs - backoffInMs ?
maxBackoffInMs : backoffInMs << 1;
+ }
+ return backoffInMs;
+ }
+
private void logParserRetryOnOutOfMemory(
final String callerName,
final int tabletEventCount,
@@ -705,21 +770,21 @@ public class PipeTsFileInsertionEvent extends
EnrichedEvent
}
public long count(final boolean skipReportOnCommit) throws Exception {
- AtomicLong count = new AtomicLong();
-
if (shouldParseTime()) {
try {
consumeTabletInsertionEventsWithRetry(
event -> {
- count.addAndGet(event.count());
+ parsedPointCountForCount.addAndGet(event.count());
if (skipReportOnCommit) {
event.skipReportOnCommit();
}
},
"PipeTsFileInsertionEvent::count");
- return count.get();
+ return parsedPointCountForCount.getAndSet(0);
} finally {
- close();
+ if (isTsFileParsingCompleted.get()) {
+ close();
+ }
}
}
@@ -732,6 +797,11 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent
/** Release the resource of {@link TsFileInsertionDataContainer}. */
@Override
public void close() {
+ tabletInsertionEventIterator.set(null);
+ releaseParsedTabletEvent(pendingTabletInsertionEvent.getAndSet(null));
+ parsedTabletInsertionEventCount.set(0);
+ parsedPointCountForCount.set(0);
+ isTsFileParsingCompleted.set(false);
dataContainer.getAndUpdate(
container -> {
if (Objects.nonNull(container)) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/AggregateProcessor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/AggregateProcessor.java
index ff9ada22414..f12de14ebaa 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/AggregateProcessor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/AggregateProcessor.java
@@ -516,32 +516,29 @@ public class AggregateProcessor implements PipeProcessor {
public void process(
final TsFileInsertionEvent tsFileInsertionEvent, final EventCollector
eventCollector)
throws Exception {
- try {
- if (tsFileInsertionEvent instanceof PipeTsFileInsertionEvent) {
- final AtomicReference<Exception> ex = new AtomicReference<>();
- ((PipeTsFileInsertionEvent) tsFileInsertionEvent)
- .consumeTabletInsertionEventsWithRetry(
- event -> {
- try {
- process(event, eventCollector);
- } catch (PipeRuntimeOutOfMemoryCriticalException e) {
- throw e;
- } catch (Exception e) {
- ex.set(e);
- }
- },
- "AggregateProcessor::process");
- if (ex.get() != null) {
- throw ex.get();
- }
- } else {
+ if (tsFileInsertionEvent instanceof PipeTsFileInsertionEvent) {
+ ((PipeTsFileInsertionEvent) tsFileInsertionEvent)
+ .consumeTabletInsertionEventsWithRetry(
+ event -> {
+ try {
+ process(event, eventCollector);
+ } catch (PipeRuntimeOutOfMemoryCriticalException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new PipeException(e.getMessage(), e);
+ }
+ },
+ "AggregateProcessor::process");
+ tsFileInsertionEvent.close();
+ } else {
+ try {
for (final TabletInsertionEvent tabletInsertionEvent :
tsFileInsertionEvent.toTabletInsertionEvents()) {
process(tabletInsertionEvent, eventCollector);
}
+ } finally {
+ tsFileInsertionEvent.close();
}
- } finally {
- tsFileInsertionEvent.close();
}
// The timeProgressIndex shall only be reported by the output events
// whose progressIndex is bounded with tablet events
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/DownSamplingProcessor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/DownSamplingProcessor.java
index fcaa0feb058..f3c5dc4bca3 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/DownSamplingProcessor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/DownSamplingProcessor.java
@@ -36,6 +36,7 @@ import
org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
+import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.tsfile.common.constant.TsFileConstant;
@@ -149,32 +150,29 @@ public abstract class DownSamplingProcessor implements
PipeProcessor {
public void process(TsFileInsertionEvent tsFileInsertionEvent,
EventCollector eventCollector)
throws Exception {
if (shouldSplitFile) {
- try {
- if (tsFileInsertionEvent instanceof PipeTsFileInsertionEvent) {
- final AtomicReference<Exception> ex = new AtomicReference<>();
- ((PipeTsFileInsertionEvent) tsFileInsertionEvent)
- .consumeTabletInsertionEventsWithRetry(
- event -> {
- try {
- process(event, eventCollector);
- } catch (PipeRuntimeOutOfMemoryCriticalException e) {
- throw e;
- } catch (Exception e) {
- ex.set(e);
- }
- },
- "DownSamplingProcessor::process");
- if (ex.get() != null) {
- throw ex.get();
- }
- } else {
+ if (tsFileInsertionEvent instanceof PipeTsFileInsertionEvent) {
+ ((PipeTsFileInsertionEvent) tsFileInsertionEvent)
+ .consumeTabletInsertionEventsWithRetry(
+ event -> {
+ try {
+ process(event, eventCollector);
+ } catch (PipeRuntimeOutOfMemoryCriticalException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new PipeException(e.getMessage(), e);
+ }
+ },
+ "DownSamplingProcessor::process");
+ tsFileInsertionEvent.close();
+ } else {
+ try {
for (final TabletInsertionEvent tabletInsertionEvent :
tsFileInsertionEvent.toTabletInsertionEvents()) {
process(tabletInsertionEvent, eventCollector);
}
+ } finally {
+ tsFileInsertionEvent.close();
}
- } finally {
- tsFileInsertionEvent.close();
}
} else {
eventCollector.collect(tsFileInsertionEvent);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/websocket/WebSocketSink.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/websocket/WebSocketSink.java
index 7841e0199b2..fb3ee9812d9 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/websocket/WebSocketSink.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/websocket/WebSocketSink.java
@@ -140,18 +140,15 @@ public class WebSocketSink implements PipeConnector,
PipeConnectorWithEventDisca
return;
}
- try {
- ((PipeTsFileInsertionEvent) tsFileInsertionEvent)
- .consumeTabletInsertionEventsWithRetry(
- event -> {
- // Skip report if any tablet events is added
- ((PipeTsFileInsertionEvent)
tsFileInsertionEvent).skipReportOnCommit();
- transfer(event);
- },
- "WebSocketConnector::transfer");
- } finally {
- tsFileInsertionEvent.close();
- }
+ ((PipeTsFileInsertionEvent) tsFileInsertionEvent)
+ .consumeTabletInsertionEventsWithRetry(
+ event -> {
+ // Skip report if any tablet events is added
+ ((PipeTsFileInsertionEvent)
tsFileInsertionEvent).skipReportOnCommit();
+ transfer(event);
+ },
+ "WebSocketConnector::transfer");
+ tsFileInsertionEvent.close();
}
@Override
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionDataContainerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionDataContainerTest.java
index 862e2b553fe..19079f4cf76 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionDataContainerTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionDataContainerTest.java
@@ -41,6 +41,7 @@ import
org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import
org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
import org.apache.iotdb.pipe.api.access.Row;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
+import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.tsfile.common.conf.TSFileConfig;
import org.apache.tsfile.common.conf.TSFileDescriptor;
@@ -184,32 +185,10 @@ public class TsFileInsertionDataContainerTest {
}
@Test
- public void
testConsumeTabletInsertionEventsWithRetryReleasesContainerOnOutOfMemory()
+ public void
testConsumeTabletInsertionEventsWithRetryPreservesProgressOnOutOfMemory()
throws Exception {
- nonalignedTsFile =
- TsFileGeneratorUtils.generateNonAlignedTsFile(
- "nonaligned-consume-oom.tsfile", 1, 1, 10, 0, 100, 10, 10);
- resource = new TsFileResource(nonalignedTsFile);
- resource.setStatusForTest(TsFileResourceStatus.NORMAL);
-
- // The TsFile generator only creates the file, so mark the resource
non-empty explicitly.
- final IDeviceID deviceID = new PlainDeviceID("root.testsg.d0");
- resource.updateStartTime(deviceID, 0);
- resource.updateEndTime(deviceID, 9);
-
final PipeTsFileInsertionEvent event =
- new PipeTsFileInsertionEvent(
- resource,
- null,
- false,
- false,
- false,
- null,
- 0,
- null,
- new PrefixPipePattern("root"),
- Long.MIN_VALUE,
- Long.MAX_VALUE);
+
createPipeTsFileInsertionEventForRetryTest("nonaligned-consume-oom.tsfile");
final AtomicReference<PipeRawTabletInsertionEvent> parsedEventReference =
new AtomicReference<>();
@@ -226,6 +205,45 @@ public class TsFileInsertionDataContainerTest {
Assert.assertEquals("expected oom", exception.getMessage());
Assert.assertNotNull(parsedEventReference.get());
+ Assert.assertFalse(parsedEventReference.get().isReleased());
+ Assert.assertNotNull(getDataContainer(event).get());
+
+ final PipeRawTabletInsertionEvent eventAtFailure =
parsedEventReference.get();
+ event.consumeTabletInsertionEventsWithRetry(
+ parsedEvent -> {
+ Assert.assertSame(eventAtFailure, parsedEvent);
+ parsedEvent.clearReferenceCount(getClass().getName());
+ },
+ "test");
+
+ Assert.assertTrue(eventAtFailure.isReleased());
+ event.close();
+ }
+
+ @Test
+ public void
testConsumeTabletInsertionEventsWithRetryReleasesProgressWhenClosed()
+ throws Exception {
+ final PipeTsFileInsertionEvent event =
+
createPipeTsFileInsertionEventForRetryTest("nonaligned-consume-close.tsfile");
+ final AtomicReference<PipeRawTabletInsertionEvent> parsedEventReference =
+ new AtomicReference<>();
+
+ Assert.assertThrows(
+ PipeException.class,
+ () ->
+ event.consumeTabletInsertionEventsWithRetry(
+ parsedEvent -> {
+ parsedEventReference.set(parsedEvent);
+ throw new PipeException("expected failure");
+ },
+ "test"));
+
+ Assert.assertNotNull(parsedEventReference.get());
+ Assert.assertFalse(parsedEventReference.get().isReleased());
+ Assert.assertNotNull(getDataContainer(event).get());
+
+ event.close();
+
Assert.assertTrue(parsedEventReference.get().isReleased());
Assert.assertNull(getDataContainer(event).get());
}
@@ -233,29 +251,8 @@ public class TsFileInsertionDataContainerTest {
@Test
public void
testConsumeTabletInsertionEventsWithRetryKeepsParserForTransientOutOfMemory()
throws Exception {
- nonalignedTsFile =
- TsFileGeneratorUtils.generateNonAlignedTsFile(
- "nonaligned-consume-transient-oom.tsfile", 1, 1, 10, 0, 100, 10,
10);
- resource = new TsFileResource(nonalignedTsFile);
- resource.setStatusForTest(TsFileResourceStatus.NORMAL);
-
- final IDeviceID deviceID = new PlainDeviceID("root.testsg.d0");
- resource.updateStartTime(deviceID, 0);
- resource.updateEndTime(deviceID, 9);
-
final PipeTsFileInsertionEvent event =
- new PipeTsFileInsertionEvent(
- resource,
- null,
- false,
- false,
- false,
- null,
- 0,
- null,
- new PrefixPipePattern("root"),
- Long.MIN_VALUE,
- Long.MAX_VALUE);
+
createPipeTsFileInsertionEventForRetryTest("nonaligned-consume-transient-oom.tsfile");
final AtomicInteger retryCount = new AtomicInteger(0);
final AtomicReference<PipeRawTabletInsertionEvent> parsedEventReference =
new AtomicReference<>();
@@ -278,6 +275,32 @@ public class TsFileInsertionDataContainerTest {
event.close();
}
+ private PipeTsFileInsertionEvent
createPipeTsFileInsertionEventForRetryTest(final String fileName)
+ throws Exception {
+ nonalignedTsFile =
+ TsFileGeneratorUtils.generateNonAlignedTsFile(fileName, 1, 1, 10, 0,
100, 10, 10);
+ resource = new TsFileResource(nonalignedTsFile);
+ resource.setStatusForTest(TsFileResourceStatus.NORMAL);
+
+ // The TsFile generator only creates the file, so mark the resource
non-empty explicitly.
+ final IDeviceID deviceID = new PlainDeviceID("root.testsg.d0");
+ resource.updateStartTime(deviceID, 0);
+ resource.updateEndTime(deviceID, 9);
+
+ return new PipeTsFileInsertionEvent(
+ resource,
+ null,
+ false,
+ false,
+ false,
+ null,
+ 0,
+ null,
+ new PrefixPipePattern("root"),
+ Long.MIN_VALUE,
+ Long.MAX_VALUE);
+ }
+
@Test
public void
testScanParserSplitNonAlignedSinglePageChunkByEstimatedPageMemory() throws
Exception {
final long originalPipeMaxReaderChunkSize =
