This is an automated email from the ASF dual-hosted git repository.
jt2594838 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 bb0f237271b [Pipe] Resume TsFile parsing after retryable failures
(#18261)
bb0f237271b is described below
commit bb0f237271b985197f25ad2a999d8d48df7b20ea
Author: Caideyipi <[email protected]>
AuthorDate: Wed Jul 22 18:36:54 2026 +0800
[Pipe] Resume TsFile parsing after retryable failures (#18261)
* [Pipe] Resume TsFile parsing after retryable failures
* [Pipe] Address parser retry review feedback
---
.../agent/task/connection/PipeEventCollector.java | 15 +--
.../subtask/processor/PipeProcessorSubtask.java | 41 +++---
.../common/tsfile/PipeTsFileInsertionEvent.java | 103 ++++++++++++---
.../processor/aggregate/AggregateProcessor.java | 39 +++---
.../downsampling/DownSamplingProcessor.java | 40 +++---
.../sink/protocol/websocket/WebSocketSink.java | 21 ++--
.../pipe/event/TsFileInsertionEventParserTest.java | 139 ++++++++++++---------
7 files changed, 233 insertions(+), 165 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 5ecd81d92b4..cc6b126384f 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
@@ -147,15 +147,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/agent/task/subtask/processor/PipeProcessorSubtask.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java
index 8c26726bfb2..a95c99ebd52 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java
@@ -172,29 +172,24 @@ public class PipeProcessorSubtask extends
PipeReportableSubtask {
// We have to parse the privilege first, to avoid passing
no-privilege data to processor
if (event instanceof PipeTsFileInsertionEvent
&& ((PipeTsFileInsertionEvent) event).shouldParse4Privilege()) {
- try (final PipeTsFileInsertionEvent tsFileInsertionEvent =
- (PipeTsFileInsertionEvent) event) {
- final AtomicReference<Exception> ex = new AtomicReference<>();
- tsFileInsertionEvent.consumeTabletInsertionEventsWithRetry(
- event1 -> {
- try {
- pipeProcessor.process(event1, outputEventCollector);
- } catch (PipeRuntimeOutOfMemoryCriticalException e) {
- throw e;
- } catch (Exception e) {
- ex.set(e);
- }
- },
- "PipeProcessorSubtask::executeOnce");
- if (tsFileInsertionEvent.isGeneratedByHistoricalExtractor()) {
- PipeTerminateEvent.markHistoricalTsFileSplit(
- tsFileInsertionEvent.getPipeName(),
- tsFileInsertionEvent.getCreationTime(),
- regionId);
- }
- if (ex.get() != null) {
- throw ex.get();
- }
+ final PipeTsFileInsertionEvent tsFileInsertionEvent =
(PipeTsFileInsertionEvent) event;
+ tsFileInsertionEvent.consumeTabletInsertionEventsWithRetry(
+ event1 -> {
+ try {
+ pipeProcessor.process(event1, outputEventCollector);
+ } catch (PipeRuntimeOutOfMemoryCriticalException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new PipeException(e.getMessage(), e);
+ }
+ },
+ "PipeProcessorSubtask::executeOnce");
+ tsFileInsertionEvent.close();
+ if (tsFileInsertionEvent.isGeneratedByHistoricalExtractor()) {
+ PipeTerminateEvent.markHistoricalTsFileSplit(
+ tsFileInsertionEvent.getPipeName(),
+ tsFileInsertionEvent.getCreationTime(),
+ regionId);
}
} else {
pipeProcessor.process((TsFileInsertionEvent) event,
outputEventCollector);
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 db0d1b261ab..988f57ce5cb 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
@@ -99,6 +99,13 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
private final AtomicBoolean isClosed;
private final AtomicReference<TsFileInsertionEventParser> eventParser;
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 IoTConsensusV2'
receiver side.
// May be updated after it is flushed. Should be negative if not set.
@@ -748,26 +755,70 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
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(
DataNodePipeMessages.FAILED_TO_ALLOCATE_MEMORY_FOR_PARSING_TSFILE,
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(DataNodePipeMessages.PIPE_SKIPPING_TEMPORARY_TSFILE_S_PARSING_WHICH,
tsFile);
+ return null;
+ }
+ iterator = initEventParser().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(
@@ -789,21 +840,34 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
}
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,
@@ -1013,21 +1077,21 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
}
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();
+ }
}
}
@@ -1040,6 +1104,11 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
/** Release the resource of {@link TsFileInsertionEventParser}. */
@Override
public void close() {
+ tabletInsertionEventIterator.set(null);
+ releaseParsedTabletEvent(pendingTabletInsertionEvent.getAndSet(null));
+ parsedTabletInsertionEventCount.set(0);
+ parsedPointCountForCount.set(0);
+ isTsFileParsingCompleted.set(false);
eventParser.getAndUpdate(
parser -> {
if (Objects.nonNull(parser)) {
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 4d4ed0c4b42..2224fc12fcc 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
@@ -542,32 +542,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 17c375614cb..06fbbcecbbf 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
@@ -145,18 +145,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/TsFileInsertionEventParserTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionEventParserTest.java
index f5187ece3cf..7a1ade122c0 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionEventParserTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionEventParserTest.java
@@ -44,6 +44,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;
@@ -183,40 +184,10 @@ public class TsFileInsertionEventParserTest {
}
@Test
- public void
testConsumeTabletInsertionEventsWithRetryReleasesParserOnOutOfMemory()
+ 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 =
IDeviceID.Factory.DEFAULT_FACTORY.create("root.testsg.d0");
- resource.updateStartTime(deviceID, 0);
- resource.updateEndTime(deviceID, 9);
-
final PipeTsFileInsertionEvent event =
- new PipeTsFileInsertionEvent(
- false,
- "root",
- resource,
- null,
- false,
- false,
- false,
- null,
- null,
- 0,
- null,
- new PrefixTreePattern("root"),
- null,
- null,
- null,
- null,
- true,
- Long.MIN_VALUE,
- Long.MAX_VALUE);
+
createPipeTsFileInsertionEventForRetryTest("nonaligned-consume-oom.tsfile");
final AtomicReference<PipeRawTabletInsertionEvent> parsedEventReference =
new AtomicReference<>();
@@ -233,6 +204,45 @@ public class TsFileInsertionEventParserTest {
Assert.assertEquals("expected oom", exception.getMessage());
Assert.assertNotNull(parsedEventReference.get());
+ Assert.assertFalse(parsedEventReference.get().isReleased());
+ Assert.assertNotNull(getEventParser(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(getEventParser(event).get());
+
+ event.close();
+
Assert.assertTrue(parsedEventReference.get().isReleased());
Assert.assertNull(getEventParser(event).get());
}
@@ -240,37 +250,8 @@ public class TsFileInsertionEventParserTest {
@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 =
IDeviceID.Factory.DEFAULT_FACTORY.create("root.testsg.d0");
- resource.updateStartTime(deviceID, 0);
- resource.updateEndTime(deviceID, 9);
-
final PipeTsFileInsertionEvent event =
- new PipeTsFileInsertionEvent(
- false,
- "root",
- resource,
- null,
- false,
- false,
- false,
- null,
- null,
- 0,
- null,
- new PrefixTreePattern("root"),
- null,
- null,
- null,
- null,
- true,
- Long.MIN_VALUE,
- Long.MAX_VALUE);
+
createPipeTsFileInsertionEventForRetryTest("nonaligned-consume-transient-oom.tsfile");
final AtomicInteger retryCount = new AtomicInteger(0);
final AtomicReference<PipeRawTabletInsertionEvent> parsedEventReference =
new AtomicReference<>();
@@ -293,6 +274,40 @@ public class TsFileInsertionEventParserTest {
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 =
IDeviceID.Factory.DEFAULT_FACTORY.create("root.testsg.d0");
+ resource.updateStartTime(deviceID, 0);
+ resource.updateEndTime(deviceID, 9);
+
+ return new PipeTsFileInsertionEvent(
+ false,
+ "root",
+ resource,
+ null,
+ false,
+ false,
+ false,
+ null,
+ null,
+ 0,
+ null,
+ new PrefixTreePattern("root"),
+ null,
+ null,
+ null,
+ null,
+ true,
+ Long.MIN_VALUE,
+ Long.MAX_VALUE);
+ }
+
@Test
public void
testScanParserSplitNonAlignedSinglePageChunkByEstimatedPageMemory() throws
Exception {
final long originalPipeMaxReaderChunkSize =
