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 5ecd81d92b412..cc6b126384f63 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 8c26726bfb204..a95c99ebd52e9 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 757411df9b976..cb1b29c2711f1 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,70 @@ 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. 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 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 +839,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; + 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, @@ -982,21 +1046,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 +1073,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 4d4ed0c4b4237..2224fc12fcc11 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 fcaa0feb05814..f3c5dc4bca316 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 17c375614cbbb..06fbbcecbbf7d 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 f5187ece3cf43..7a1ade122c07e 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.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 void testScanParserReleasesTabletMemoryAfterRawTabletGenerated() throws E } @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 parsedEventReference = new AtomicReference<>(); @@ -233,6 +204,45 @@ public void testConsumeTabletInsertionEventsWithRetryReleasesParserOnOutOfMemory 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 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 void testConsumeTabletInsertionEventsWithRetryReleasesParserOnOutOfMemory @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 parsedEventReference = new AtomicReference<>(); @@ -293,6 +274,40 @@ public void testConsumeTabletInsertionEventsWithRetryKeepsParserForTransientOutO 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 =