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 5ecd81d92b41..cc6b126384f6 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 @@ private void parseAndCollectEvent(final PipeTsFileInsertionEvent sourceEvent) th 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 8c26726bfb20..a95c99ebd52e 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 @@ protected boolean executeOnce() throws Exception { // 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 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 757411df9b97..631905501e8b 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 @@ -98,6 +98,13 @@ public class PipeTsFileInsertionEvent extends PipeInsertionEvent private final AtomicBoolean isClosed; private final AtomicReference eventParser; private final AtomicBoolean isTsFileParserMemoryReserved = new AtomicBoolean(false); + private final AtomicReference> tabletInsertionEventIterator = + new AtomicReference<>(); + private final AtomicReference 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. @@ -747,26 +754,68 @@ public interface TabletInsertionEventConsumer { public void consumeTabletInsertionEventsWithRetry( final TabletInsertionEventConsumer consumer, final String callerName) throws Exception { - int tabletEventCount = 0; try { - final Iterable iterable = toTabletInsertionEvents(); - final Iterator 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. + waitForResourceEnough4Parsing((long) ((1 + Math.random()) * 20 * 1000)); + + final PipeRawTabletInsertionEvent pendingEvent = pendingTabletInsertionEvent.get(); + if (pendingEvent != null) { + return pendingEvent; + } + + Iterator 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( @@ -788,21 +837,34 @@ private void consumeParsedTabletInsertionEventWithRetry( } 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; + final int shift = Math.min(30, Math.max(0, retryCount - 1)); + if (initialBackoffInMs > (maxBackoffInMs >> shift)) { + return maxBackoffInMs; + } + return Math.min(maxBackoffInMs, initialBackoffInMs << shift); + } + private void logParserRetryOnOutOfMemory( final String callerName, final int tabletEventCount, @@ -982,21 +1044,21 @@ private TsFileInsertionEventParser initEventParser() { } 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(); + } } } @@ -1009,6 +1071,11 @@ public long count(final boolean skipReportOnCommit) throws Exception { /** 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 4d4ed0c4b423..2224fc12fcc1 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 @@ private Map> processRow( public void process( final TsFileInsertionEvent tsFileInsertionEvent, final EventCollector eventCollector) throws Exception { - try { - if (tsFileInsertionEvent instanceof PipeTsFileInsertionEvent) { - final AtomicReference 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 fcaa0feb0581..f3c5dc4bca31 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.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 @@ protected abstract void processRow( public void process(TsFileInsertionEvent tsFileInsertionEvent, EventCollector eventCollector) throws Exception { if (shouldSplitFile) { - try { - if (tsFileInsertionEvent instanceof PipeTsFileInsertionEvent) { - final AtomicReference 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 17c375614cbb..06fbbcecbbf7 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 void transfer(TsFileInsertionEvent tsFileInsertionEvent) throws Exception 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 f5187ece3cf4..7a47520a8601 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 @@ -183,7 +183,7 @@ public void testScanParserReleasesTabletMemoryAfterRawTabletGenerated() throws E } @Test - public void testConsumeTabletInsertionEventsWithRetryReleasesParserOnOutOfMemory() + public void testConsumeTabletInsertionEventsWithRetryPreservesProgressOnOutOfMemory() throws Exception { nonalignedTsFile = TsFileGeneratorUtils.generateNonAlignedTsFile( @@ -233,8 +233,19 @@ public void testConsumeTabletInsertionEventsWithRetryReleasesParserOnOutOfMemory Assert.assertEquals("expected oom", exception.getMessage()); Assert.assertNotNull(parsedEventReference.get()); - Assert.assertTrue(parsedEventReference.get().isReleased()); - Assert.assertNull(getEventParser(event).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