From 7f766f0ab6114c7731d42bb73137346fdfbafce1 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 22 Jul 2026 18:36:54 +0800 Subject: [PATCH] [Pipe] Resume TsFile parsing after retryable failures (#18261) * [Pipe] Resume TsFile parsing after retryable failures * [Pipe] Address parser retry review feedback (cherry picked from commit bb0f237271b985197f25ad2a999d8d48df7b20ea) --- .../task/connection/PipeEventCollector.java | 15 +-- .../tsfile/PipeTsFileInsertionEvent.java | 104 +++++++++++++--- .../aggregate/AggregateProcessor.java | 39 +++--- .../downsampling/DownSamplingProcessor.java | 40 +++--- .../protocol/websocket/WebSocketSink.java | 21 ++-- .../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 f0cc462161287..df72ccb830dbc 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 @@ 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/event/common/tsfile/PipeTsFileInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java index 98c660ea35138..046262dbd991e 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 dataContainer; 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 PipeConsensus' receiver side. // May be updated after it is flushed. Should be negative if not set. @@ -481,28 +488,73 @@ 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( "{}: 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 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 @@ 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, @@ -705,21 +770,21 @@ private TsFileInsertionDataContainer initDataContainer() { } 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 long count(final boolean skipReportOnCommit) throws Exception { /** 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 ff9ada22414ea..f12de14ebaa4c 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 @@ 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 7841e0199b250..fb3ee9812d9d5 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 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/TsFileInsertionDataContainerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionDataContainerTest.java index 862e2b553fe48..19079f4cf764e 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.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 void testScanContainerReleasesTabletMemoryAfterRawTabletGenerated() throw } @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 parsedEventReference = new AtomicReference<>(); @@ -226,6 +205,45 @@ public void testConsumeTabletInsertionEventsWithRetryReleasesContainerOnOutOfMem 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 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 void testConsumeTabletInsertionEventsWithRetryReleasesContainerOnOutOfMem @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 parsedEventReference = new AtomicReference<>(); @@ -278,6 +275,32 @@ 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 = 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 =