diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 520a300c8e197..4576632e0b1c8 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -21,6 +21,9 @@ public final class DataNodePipeMessages { + public static final String LOG_FAILED_TO_RESOLVE_TRANSFER_EXCEPTION_A4F5397A = + "Failed to resolve transfer exception."; + // ===================== CONSENSUS ===================== public static final String CLOSING_DELETION_RESOURCE_MANAGER_FOR = @@ -2314,6 +2317,12 @@ private DataNodePipeMessages() {} public static final String PIPE_EXCEPTION_FORCERESIZE_FAILED_TO_ALLOCATE_MEMORY_AFTER_D_RETRIES_TOTAL_8C6948BC = "forceResize: failed to allocate memory after %d retries, total memory size %d bytes, used " + "memory size %d bytes, requested memory size %d bytes"; + public static final String + EXCEPTION_UNSUPPORTED_BATCH_TYPE_ARG_WHEN_TRANSFERRING_TABLET_INSERTION_EVENT_66153E12 = + "Unsupported batch type %s when transferring tablet insertion event."; + public static final String + EXCEPTION_FAILED_TO_TRANSFER_TSFILE_BATCH_BECAUSE_NO_TSFILE_WAS_GENERATED_FOR_ARG_CC60CCEB = + "Failed to transfer TsFile batch because no TsFile was generated for %s."; public static final String PIPE_EXCEPTION_FAILED_TO_GET_HARDLINK_OR_COPIED_FILE_IN_PIPE_DIR_FOR_FILE_F009D86E = "failed to get hardlink or copied file in pipe dir for file %s, it is not a tsfile, mod file " + "or resource file"; diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 2a43385342fa8..0c807a8d2ae61 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -21,6 +21,9 @@ public final class DataNodePipeMessages { + public static final String LOG_FAILED_TO_RESOLVE_TRANSFER_EXCEPTION_A4F5397A = + "解析 transfer exception 失败。"; + // ===================== CONSENSUS ===================== public static final String CLOSING_DELETION_RESOURCE_MANAGER_FOR = "正在关闭 {} 的删除资源管理器..."; @@ -2145,6 +2148,12 @@ private DataNodePipeMessages() {} "forceAllocate：重试 %d 次后仍无法分配内存，总内存大小 %d bytes，已用内存大小 %d bytes，请求内存大小 %d bytes"; public static final String PIPE_EXCEPTION_FORCERESIZE_FAILED_TO_ALLOCATE_MEMORY_AFTER_D_RETRIES_TOTAL_8C6948BC = "forceResize：重试 %d 次后仍无法分配内存，总内存大小 %d bytes，已用内存大小 %d bytes，请求内存大小 %d bytes"; + public static final String + EXCEPTION_UNSUPPORTED_BATCH_TYPE_ARG_WHEN_TRANSFERRING_TABLET_INSERTION_EVENT_66153E12 = + "传输 tablet insertion event 时不支持 batch 类型 %s。"; + public static final String + EXCEPTION_FAILED_TO_TRANSFER_TSFILE_BATCH_BECAUSE_NO_TSFILE_WAS_GENERATED_FOR_ARG_CC60CCEB = + "无法传输 TsFile batch，因为没有为 %s 生成 TsFile。"; public static final String PIPE_EXCEPTION_FAILED_TO_GET_HARDLINK_OR_COPIED_FILE_IN_PIPE_DIR_FOR_FILE_F009D86E = "获取 pipe 目录中文件 %s 的 hardlink 或复制文件失败；该文件不是 tsfile、mod 文件或 resource 文件"; public static final String PIPE_EXCEPTION_PIPEPLANTOSTATEMENTVISITOR_DOES_NOT_SUPPORT_VISITING_GENERAL_452AAA60 = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/statement/PipeStatementInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/statement/PipeStatementInsertionEvent.java index 9f812d90b3c8b..859c29b34c822 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/statement/PipeStatementInsertionEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/statement/PipeStatementInsertionEvent.java @@ -21,6 +21,7 @@ import org.apache.iotdb.commons.consensus.index.ProgressIndex; import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex; +import org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException; import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta; import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern; import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern; @@ -91,8 +92,17 @@ public PipeStatementInsertionEvent( @Override public boolean internallyIncreaseResourceReferenceCount(String holderMessage) { - PipeDataNodeResourceManager.memory() - .forceResize(allocatedMemoryBlock, statement.ramBytesUsed() + INSTANCE_SIZE); + final long targetSize = statement.ramBytesUsed() + INSTANCE_SIZE; + if (!PipeDataNodeResourceManager.memory().tryResize(allocatedMemoryBlock, targetSize)) { + throw new PipeRuntimeOutOfMemoryCriticalException( + String.format( + DataNodePipeMessages + .PIPE_EXCEPTION_FORCERESIZE_FAILED_TO_ALLOCATE_MEMORY_AFTER_D_RETRIES_TOTAL_8C6948BC, + 0, + PipeDataNodeResourceManager.memory().getTotalNonFloatingMemorySizeInBytes(), + PipeDataNodeResourceManager.memory().getUsedMemorySizeInBytes(), + targetSize - allocatedMemoryBlock.getMemoryUsageInBytes())); + } if (Objects.nonNull(pipeName)) { PipeDataNodeSinglePipeMetrics.getInstance() .increaseRawTabletEventCount(pipeName, creationTime); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java index dc2ab1d381fd8..caab7381c8321 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java @@ -23,6 +23,7 @@ import org.apache.iotdb.commons.consensus.index.ProgressIndex; import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex; import org.apache.iotdb.commons.exception.IllegalPathException; +import org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException; import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta; import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern; import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern; @@ -258,10 +259,17 @@ public PipeRawTabletInsertionEvent( @Override public boolean internallyIncreaseResourceReferenceCount(final String holderMessage) { - PipeDataNodeResourceManager.memory() - .forceResize( - allocatedMemoryBlock, - PipeMemoryWeightUtil.calculateTabletSizeInBytes(tablet) + INSTANCE_SIZE); + final long targetSize = PipeMemoryWeightUtil.calculateTabletSizeInBytes(tablet) + INSTANCE_SIZE; + if (!PipeDataNodeResourceManager.memory().tryResize(allocatedMemoryBlock, targetSize)) { + throw new PipeRuntimeOutOfMemoryCriticalException( + String.format( + DataNodePipeMessages + .PIPE_EXCEPTION_FORCERESIZE_FAILED_TO_ALLOCATE_MEMORY_AFTER_D_RETRIES_TOTAL_8C6948BC, + 0, + PipeDataNodeResourceManager.memory().getTotalNonFloatingMemorySizeInBytes(), + PipeDataNodeResourceManager.memory().getUsedMemorySizeInBytes(), + targetSize - allocatedMemoryBlock.getMemoryUsageInBytes())); + } if (Objects.nonNull(pipeName)) { PipeDataNodeSinglePipeMetrics.getInstance() .increaseRawTabletEventCount(pipeName, creationTime); 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 335278e2f3790..aebdec7d3d603 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 @@ -108,6 +108,19 @@ public class PipeTsFileInsertionEvent extends PipeInsertionEvent new AtomicReference<>(); private final AtomicReference pendingTabletInsertionEvent = new AtomicReference<>(); + // Only one caller may advance the parser at a time. close() deliberately does not acquire this + // monitor: a parser consumer may be blocked in a user callback, while close() still needs to + // detach the parser and release resources promptly. + private final Object tabletConsumptionLock = new Object(); + // Guarded by eventParser. A close() increments this generation so an invocation that was waiting + // for parser memory cannot install a parser after the parser state has been reset. + private long parserStateGeneration; + // Guarded by eventParser. The pending tablet remains owned by the consumer until its callback + // returns, even when close() races with that callback. + private PipeRawTabletInsertionEvent consumingPendingTabletInsertionEvent; + // Guarded by eventParser. Set by close() when it detached a tablet that is still in a consumer + // callback; the callback clears and releases it after returning. + private boolean pendingTabletReleaseDeferred; private final AtomicInteger parsedTabletInsertionEventCount = new AtomicInteger(0); private final AtomicBoolean isTsFileParsingCompleted = new AtomicBoolean(false); private final AtomicLong parsedPointCountForCount = new AtomicLong(0); @@ -418,17 +431,35 @@ public boolean internallyIncreaseResourceReferenceCount(final String holderMessa extractTime = System.nanoTime(); final String pipeTsFileResourcePipeName = PipeTsFileResourceManager.getPipeTsFileResourcePipeName(pipeName, creationTime); + final File originalTsFile = tsFile; + final File originalModFile = modFile; + File increasedTsFile = null; + boolean increased = false; try { - tsFile = + increasedTsFile = PipeDataNodeResourceManager.tsfile() .increaseFileReference(tsFile, true, pipeTsFileResourcePipeName); + tsFile = increasedTsFile; if (isWithMod) { modFile = PipeDataNodeResourceManager.tsfile() .increaseFileReference(modFile, false, pipeTsFileResourcePipeName); } + increased = true; return true; } catch (final Exception e) { + // A TsFile reference may have been acquired before the mod-file reference failed. Roll it + // back and restore the original paths so a later retry starts from a valid source file. + if (increasedTsFile != null) { + try { + PipeDataNodeResourceManager.tsfile() + .decreaseFileReference(increasedTsFile, pipeTsFileResourcePipeName); + } catch (final Exception rollbackException) { + e.addSuppressed(rollbackException); + } + } + tsFile = originalTsFile; + modFile = originalModFile; LOGGER.warn( String.format( DataNodePipeMessages.INCREASE_REFERENCE_COUNT_TSFILE_OR_MODFILE_ERROR_HOLDER_FMT, @@ -438,9 +469,13 @@ public boolean internallyIncreaseResourceReferenceCount(final String holderMessa e); return false; } finally { - if (Objects.nonNull(pipeName)) { - PipeDataNodeSinglePipeMetrics.getInstance() - .increaseTsFileEventCount(pipeName, creationTime); + if (increased && Objects.nonNull(pipeName)) { + try { + PipeDataNodeSinglePipeMetrics.getInstance() + .increaseTsFileEventCount(pipeName, creationTime); + } catch (final Exception e) { + LOGGER.warn(DataNodePipeMessages.DECREASE_REFERENCE_COUNT_FOR_TSFILE_ERROR, tsFile, e); + } } } } @@ -449,32 +484,62 @@ public boolean internallyIncreaseResourceReferenceCount(final String holderMessa public boolean internallyDecreaseResourceReferenceCount(final String holderMessage) { final String pipeTsFileResourcePipeName = PipeTsFileResourceManager.getPipeTsFileResourcePipeName(pipeName, creationTime); + boolean isSuccessful = true; try { PipeDataNodeResourceManager.tsfile() .decreaseFileReference(tsFile, pipeTsFileResourcePipeName); - if (isWithMod) { + } catch (final Exception e) { + isSuccessful = false; + LOGGER.warn( + String.format( + DataNodePipeMessages.DECREASE_REFERENCE_COUNT_TSFILE_ERROR_HOLDER_FMT, + tsFile, + holderMessage), + e); + } + + // Keep each cleanup independent. In particular, a failed TsFile path resolution must not + // prevent releasing the mod-file reference or the parser memory. + if (isWithMod) { + try { PipeDataNodeResourceManager.tsfile() .decreaseFileReference(modFile, pipeTsFileResourcePipeName); + } catch (final Exception e) { + isSuccessful = false; + LOGGER.warn( + String.format( + DataNodePipeMessages.DECREASE_REFERENCE_COUNT_TSFILE_ERROR_HOLDER_FMT, + modFile, + holderMessage), + e); } + } + + try { close(); - return true; } catch (final Exception e) { + isSuccessful = false; LOGGER.warn( String.format( DataNodePipeMessages.DECREASE_REFERENCE_COUNT_TSFILE_ERROR_HOLDER_FMT, - tsFile.getPath(), + tsFile, holderMessage), e); - return false; } finally { if (Objects.nonNull(pipeName)) { - PipeDataNodeSinglePipeMetrics.getInstance() - .decreaseTsFileEventCount( - pipeName, - creationTime, - shouldReportOnCommit ? System.nanoTime() - extractTime : -1); + try { + PipeDataNodeSinglePipeMetrics.getInstance() + .decreaseTsFileEventCount( + pipeName, + creationTime, + shouldReportOnCommit ? System.nanoTime() - extractTime : -1); + } catch (final Exception e) { + isSuccessful = false; + LOGGER.warn(DataNodePipeMessages.DECREASE_REFERENCE_COUNT_FOR_TSFILE_ERROR, tsFile, e); + } } } + return isSuccessful; } @Override @@ -802,82 +867,165 @@ public void consumeTabletInsertionEventsWithRetry( final String callerName, final PipeProcessorSubtaskExecutionGuard processorExecutionGuard) throws Exception { - try { - while (true) { - processorExecutionGuard.check(); - final PipeRawTabletInsertionEvent parsedEvent = - getNextTabletInsertionEventFromSavedProgress(processorExecutionGuard); - if (parsedEvent == null) { - isTsFileParsingCompleted.set(true); - releaseTsFileParserMemoryIfReserved(); - return; + synchronized (tabletConsumptionLock) { + final long parserStateGeneration; + synchronized (eventParser) { + parserStateGeneration = this.parserStateGeneration; + } + + try { + while (true) { + processorExecutionGuard.check(); + final PipeRawTabletInsertionEvent parsedEvent = + getNextTabletInsertionEventFromSavedProgress( + processorExecutionGuard, parserStateGeneration); + if (parsedEvent == null) { + final boolean isCurrentGeneration; + synchronized (eventParser) { + isCurrentGeneration = parserStateGeneration == this.parserStateGeneration; + if (isCurrentGeneration) { + isTsFileParsingCompleted.set(true); + } + } + releaseTsFileParserMemoryIfReserved(); + if (!isCurrentGeneration) { + return; + } + return; + } + + boolean consumed = false; + try { + processorExecutionGuard.check(); + consumeParsedTabletInsertionEventWithRetry( + consumer, + callerName, + parsedTabletInsertionEventCount.get(), + parsedEvent, + processorExecutionGuard); + consumed = true; + } finally { + finishConsumingTabletInsertionEvent(parsedEvent, consumed); + } + + synchronized (eventParser) { + if (parserStateGeneration != this.parserStateGeneration) { + return; + } + } + processorExecutionGuard.check(); } - processorExecutionGuard.check(); - consumeParsedTabletInsertionEventWithRetry( - consumer, + } catch (final PipeProcessorSubtaskYieldException e) { + releaseTsFileParserMemoryIfReserved(); + if (!processorExecutionGuard.isCurrentInvocationValid()) { + cancelTsFileParserMemoryReservationIfPending(); + } + throw e; + } catch (final PipeRuntimeOutOfMemoryCriticalException e) { + // 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(), parsedTabletInsertionEventCount.get(), - parsedEvent, - processorExecutionGuard); - pendingTabletInsertionEvent.compareAndSet(parsedEvent, null); - processorExecutionGuard.check(); - } - } catch (final PipeProcessorSubtaskYieldException e) { - releaseTsFileParserMemoryIfReserved(); - if (!processorExecutionGuard.isCurrentInvocationValid()) { - cancelTsFileParserMemoryReservationIfPending(); + e); + throw e; + } catch (final Exception e) { + releaseTsFileParserMemoryIfReserved(); + throw e; } - throw e; - } catch (final PipeRuntimeOutOfMemoryCriticalException e) { - // 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(), - parsedTabletInsertionEventCount.get(), - e); - throw e; - } catch (final Exception e) { - releaseTsFileParserMemoryIfReserved(); - throw e; } } private PipeRawTabletInsertionEvent getNextTabletInsertionEventFromSavedProgress( - final PipeProcessorSubtaskExecutionGuard processorExecutionGuard) throws Exception { - if (isTsFileParsingCompleted.get()) { - return null; + final PipeProcessorSubtaskExecutionGuard processorExecutionGuard, + final long expectedParserStateGeneration) + throws Exception { + synchronized (eventParser) { + if (expectedParserStateGeneration != parserStateGeneration + || isTsFileParsingCompleted.get()) { + return null; + } } // Reacquire parser memory after a previous failure yielded the active parser slot. Processor // subtasks use non-blocking admission here, while other callers retain the bounded wait. - reserveResource4Parsing(processorExecutionGuard); - - final PipeRawTabletInsertionEvent pendingEvent = pendingTabletInsertionEvent.get(); - if (pendingEvent != null) { - return pendingEvent; + if (!reserveResource4Parsing(processorExecutionGuard, expectedParserStateGeneration)) { + return null; } - Iterator iterator = tabletInsertionEventIterator.get(); - if (iterator == null) { - if (!waitForTsFileClose(processorExecutionGuard)) { - LOGGER.warn(DataNodePipeMessages.PIPE_SKIPPING_TEMPORARY_TSFILE_S_PARSING_WHICH, tsFile); + synchronized (eventParser) { + if (expectedParserStateGeneration != parserStateGeneration + || isTsFileParsingCompleted.get()) { return null; } - iterator = initEventParser().toTabletInsertionEvents().iterator(); - tabletInsertionEventIterator.set(iterator); + + final PipeRawTabletInsertionEvent pendingEvent = pendingTabletInsertionEvent.get(); + if (pendingEvent != null) { + markTabletInsertionEventAsConsuming(pendingEvent); + return pendingEvent; + } } - if (!iterator.hasNext()) { + if (!waitForTsFileClose(processorExecutionGuard)) { + LOGGER.warn(DataNodePipeMessages.PIPE_SKIPPING_TEMPORARY_TSFILE_S_PARSING_WHICH, tsFile); return null; } - final PipeRawTabletInsertionEvent nextEvent = (PipeRawTabletInsertionEvent) iterator.next(); - pendingTabletInsertionEvent.set(nextEvent); - parsedTabletInsertionEventCount.incrementAndGet(); - return nextEvent; + synchronized (eventParser) { + if (expectedParserStateGeneration != parserStateGeneration + || isTsFileParsingCompleted.get()) { + return null; + } + + Iterator iterator = tabletInsertionEventIterator.get(); + if (iterator == null) { + iterator = initEventParser().toTabletInsertionEvents().iterator(); + if (expectedParserStateGeneration != parserStateGeneration) { + return null; + } + tabletInsertionEventIterator.set(iterator); + } + + if (!iterator.hasNext()) { + return null; + } + + final PipeRawTabletInsertionEvent nextEvent = (PipeRawTabletInsertionEvent) iterator.next(); + pendingTabletInsertionEvent.set(nextEvent); + parsedTabletInsertionEventCount.incrementAndGet(); + markTabletInsertionEventAsConsuming(nextEvent); + return nextEvent; + } + } + + private void markTabletInsertionEventAsConsuming(final PipeRawTabletInsertionEvent event) { + consumingPendingTabletInsertionEvent = event; + } + + private void finishConsumingTabletInsertionEvent( + final PipeRawTabletInsertionEvent event, final boolean consumed) { + PipeRawTabletInsertionEvent eventToRelease = null; + synchronized (eventParser) { + if (consumingPendingTabletInsertionEvent != event) { + return; + } + + consumingPendingTabletInsertionEvent = null; + if (consumed && pendingTabletInsertionEvent.get() == event) { + pendingTabletInsertionEvent.compareAndSet(event, null); + } + if (pendingTabletReleaseDeferred) { + pendingTabletReleaseDeferred = false; + eventToRelease = event; + } + } + + if (eventToRelease != null) { + releaseParsedTabletEvent(eventToRelease); + } } private void consumeParsedTabletInsertionEventWithRetry( @@ -986,8 +1134,19 @@ public Iterable toTabletInsertionEvents(final long timeout LOGGER.warn(DataNodePipeMessages.PIPE_SKIPPING_TEMPORARY_TSFILE_S_PARSING_WHICH, tsFile); return Collections.emptyList(); } - waitForResourceEnough4Parsing(timeoutMs); - return initEventParser().toTabletInsertionEvents(); + final long parserStateGeneration; + synchronized (eventParser) { + parserStateGeneration = this.parserStateGeneration; + } + if (!waitForResourceEnough4Parsing(timeoutMs, parserStateGeneration)) { + return Collections.emptyList(); + } + synchronized (eventParser) { + if (parserStateGeneration != this.parserStateGeneration) { + return Collections.emptyList(); + } + return initEventParser().toTabletInsertionEvents(); + } } catch (final Exception e) { close(); @@ -1014,20 +1173,28 @@ public Iterable toTabletInsertionEvents(final long timeout } } - private void reserveResource4Parsing( - final PipeProcessorSubtaskExecutionGuard processorExecutionGuard) + private boolean reserveResource4Parsing( + final PipeProcessorSubtaskExecutionGuard processorExecutionGuard, + final long expectedParserStateGeneration) throws InterruptedException { if (!processorExecutionGuard.isEnabled()) { - waitForResourceEnough4Parsing((long) ((1 + Math.random()) * 20 * 1000)); - return; + return waitForResourceEnough4Parsing( + (long) ((1 + Math.random()) * 20 * 1000), expectedParserStateGeneration); } processorExecutionGuard.check(); + if (!isParserStateGenerationCurrent(expectedParserStateGeneration)) { + return false; + } final PipeMemoryManager memoryManager = PipeDataNodeResourceManager.memory(); if (tryReserveTsFileParserMemory(memoryManager)) { try { processorExecutionGuard.check(); - return; + if (isParserStateGenerationCurrent(expectedParserStateGeneration)) { + return true; + } + releaseTsFileParserMemoryIfReserved(); + return false; } catch (final PipeProcessorSubtaskYieldException e) { releaseTsFileParserMemoryIfReserved(); throw e; @@ -1038,19 +1205,36 @@ private void reserveResource4Parsing( cancelTsFileParserMemoryReservationIfPending(); processorExecutionGuard.check(); } + if (!isParserStateGenerationCurrent(expectedParserStateGeneration)) { + cancelTsFileParserMemoryReservationIfPending(); + return false; + } processorExecutionGuard.yieldIfParserNotAdmitted(); + return false; } - private void waitForResourceEnough4Parsing(final long timeoutMs) throws InterruptedException { + private boolean waitForResourceEnough4Parsing( + final long timeoutMs, final long expectedParserStateGeneration) throws InterruptedException { final PipeMemoryManager memoryManager = PipeDataNodeResourceManager.memory(); + if (!isParserStateGenerationCurrent(expectedParserStateGeneration)) { + return false; + } if (tryReserveTsFileParserMemory(memoryManager)) { - return; + if (isParserStateGenerationCurrent(expectedParserStateGeneration)) { + return true; + } + releaseTsFileParserMemoryIfReserved(); + return false; } final long startTime = System.currentTimeMillis(); long lastRecordTime = startTime; while (!tryReserveTsFileParserMemory(memoryManager)) { + if (!isParserStateGenerationCurrent(expectedParserStateGeneration)) { + cancelTsFileParserMemoryReservationIfPending(); + return false; + } final long currentTime = System.currentTimeMillis(); final long elapsedRecordTimeInMs = currentTime - lastRecordTime; final long waitTimeInMs = currentTime - startTime; @@ -1090,6 +1274,17 @@ private void waitForResourceEnough4Parsing(final long timeoutMs) throws Interrup DataNodePipeMessages.WAIT_FOR_MEMORY_ENOUGH_FOR_PARSING_FOR, resource != null ? resource.getTsFilePath() : "tsfile", waitTimeSeconds); + if (isParserStateGenerationCurrent(expectedParserStateGeneration)) { + return true; + } + releaseTsFileParserMemoryIfReserved(); + return false; + } + + private boolean isParserStateGenerationCurrent(final long expectedParserStateGeneration) { + synchronized (eventParser) { + return expectedParserStateGeneration == parserStateGeneration; + } } private boolean tryReserveTsFileParserMemory(final PipeMemoryManager memoryManager) { @@ -1118,10 +1313,12 @@ private void releaseTsFileParserMemoryIfReserved() { } public void cancelTsFileParserMemoryReservationIfPending() { - if (!isTsFileParserMemoryReserved.get()) { - PipeDataNodeResourceManager.memory() - .cancelTsFileParserMemoryReservation( - pipeName, creationTime, dataRegionId, tsFileParserMemoryReservationKey); + synchronized (isTsFileParserMemoryReserved) { + if (!isTsFileParserMemoryReserved.get()) { + PipeDataNodeResourceManager.memory() + .cancelTsFileParserMemoryReservation( + pipeName, creationTime, dataRegionId, tsFileParserMemoryReservationKey); + } } } @@ -1136,26 +1333,33 @@ public boolean isGeneratedByHistoricalExtractor() { private TsFileInsertionEventParser initEventParser() { try { - eventParser.compareAndSet( - null, - new TsFileInsertionEventParserProvider( - pipeName, - creationTime, - tsFile, - treePattern, - tablePattern, - startTime, - endTime, - pipeTaskMeta, - // Do not parse privilege if it should not be parsed - // To avoid renaming of the tsFile database - shouldParse4Privilege - ? new UserEntity(Long.parseLong(userId), userName, cliHostname) - : null, - this, - tsFileParser) - .provide(isWithMod)); - return eventParser.get(); + synchronized (eventParser) { + final TsFileInsertionEventParser parser = eventParser.get(); + if (parser != null) { + return parser; + } + + final TsFileInsertionEventParser createdParser = + new TsFileInsertionEventParserProvider( + pipeName, + creationTime, + tsFile, + treePattern, + tablePattern, + startTime, + endTime, + pipeTaskMeta, + // Do not parse privilege if it should not be parsed + // To avoid renaming of the tsFile database + shouldParse4Privilege + ? new UserEntity(Long.parseLong(userId), userName, cliHostname) + : null, + this, + tsFileParser) + .provide(isWithMod); + eventParser.set(createdParser); + return createdParser; + } } catch (final Exception e) { close(); @@ -1194,20 +1398,57 @@ public long count(final boolean skipReportOnCommit) throws Exception { /** Release the resource of {@link TsFileInsertionEventParser}. */ @Override public void close() { - cancelTsFileParserMemoryReservationIfPending(); - tabletInsertionEventIterator.set(null); - releaseParsedTabletEvent(pendingTabletInsertionEvent.getAndSet(null)); - parsedTabletInsertionEventCount.set(0); - parsedPointCountForCount.set(0); - isTsFileParsingCompleted.set(false); - eventParser.getAndUpdate( - parser -> { - if (Objects.nonNull(parser)) { - parser.close(); - } - return null; - }); - releaseTsFileParserMemoryIfReserved(); + // Every cleanup step is best effort and independent. A parser/reader close failure must not + // strand its reservation or the pending tablet event. + try { + cancelTsFileParserMemoryReservationIfPending(); + } catch (final Exception e) { + LOGGER.warn(DataNodePipeMessages.DECREASE_REFERENCE_COUNT_FOR_TSFILE_ERROR, tsFile, e); + } + + final PipeRawTabletInsertionEvent detachedPendingEvent; + final boolean pendingEventIsBeingConsumed; + final TsFileInsertionEventParser parserToClose; + synchronized (eventParser) { + ++parserStateGeneration; + tabletInsertionEventIterator.set(null); + detachedPendingEvent = pendingTabletInsertionEvent.getAndSet(null); + parsedTabletInsertionEventCount.set(0); + parsedPointCountForCount.set(0); + isTsFileParsingCompleted.set(false); + + parserToClose = eventParser.getAndSet(null); + pendingEventIsBeingConsumed = + detachedPendingEvent != null + && detachedPendingEvent == consumingPendingTabletInsertionEvent; + if (pendingEventIsBeingConsumed) { + // The consumer owns this tablet until its callback returns. It must not observe a released + // tablet merely because another thread closed the source event. + pendingTabletReleaseDeferred = true; + } + } + + if (detachedPendingEvent != null && !pendingEventIsBeingConsumed) { + try { + releaseParsedTabletEvent(detachedPendingEvent); + } catch (final Exception e) { + LOGGER.warn(DataNodePipeMessages.DECREASE_REFERENCE_COUNT_FOR_TSFILE_ERROR, tsFile, e); + } + } + + if (parserToClose != null) { + try { + parserToClose.close(); + } catch (final Exception e) { + LOGGER.warn(DataNodePipeMessages.DECREASE_REFERENCE_COUNT_FOR_TSFILE_ERROR, tsFile, e); + } + } + + try { + releaseTsFileParserMemoryIfReserved(); + } catch (final Exception e) { + LOGGER.warn(DataNodePipeMessages.DECREASE_REFERENCE_COUNT_FOR_TSFILE_ERROR, tsFile, e); + } } /////////////////////////// Object /////////////////////////// @@ -1299,24 +1540,39 @@ protected void finalizeResource() { PipeDataNodeResourceManager.memory() .cancelTsFileParserMemoryReservation( pipeName, creationTime, dataRegionId, tsFileParserMemoryReservationKey); - final String pipeTsFileResourcePipeName = - PipeTsFileResourceManager.getPipeTsFileResourcePipeName(pipeName, creationTime); - // decrease reference count + } catch (final Exception e) { + LOGGER.warn(DataNodePipeMessages.DECREASE_REFERENCE_COUNT_FOR_TSFILE_ERROR, tsFile, e); + } + + final String pipeTsFileResourcePipeName = + PipeTsFileResourceManager.getPipeTsFileResourcePipeName(pipeName, creationTime); + try { PipeDataNodeResourceManager.tsfile() .decreaseFileReference(tsFile, pipeTsFileResourcePipeName); - if (isWithMod) { + } catch (final Exception e) { + LOGGER.warn(DataNodePipeMessages.DECREASE_REFERENCE_COUNT_FOR_TSFILE_ERROR, tsFile, e); + } + if (isWithMod) { + try { PipeDataNodeResourceManager.tsfile() .decreaseFileReference(modFile, pipeTsFileResourcePipeName); + } catch (final Exception e) { + LOGGER.warn(DataNodePipeMessages.DECREASE_REFERENCE_COUNT_FOR_TSFILE_ERROR, modFile, e); } + } - // close event parser - eventParser.getAndUpdate( - parser -> { - if (Objects.nonNull(parser)) { - parser.close(); - } - return null; - }); + synchronized (eventParser) { + final TsFileInsertionEventParser parser = eventParser.get(); + if (parser != null) { + try { + parser.close(); + eventParser.compareAndSet(parser, null); + } catch (final Exception e) { + LOGGER.warn(DataNodePipeMessages.DECREASE_REFERENCE_COUNT_FOR_TSFILE_ERROR, tsFile, e); + } + } + } + try { synchronized (isTsFileParserMemoryReserved) { if (isTsFileParserMemoryReserved.compareAndSet(true, false)) { PipeDataNodeResourceManager.memory() @@ -1324,8 +1580,7 @@ protected void finalizeResource() { } } } catch (final Exception e) { - LOGGER.warn( - DataNodePipeMessages.DECREASE_REFERENCE_COUNT_FOR_TSFILE_ERROR, tsFile.getPath(), e); + LOGGER.warn(DataNodePipeMessages.DECREASE_REFERENCE_COUNT_FOR_TSFILE_ERROR, tsFile, e); } } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserMemoryManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserMemoryManager.java index 68669ce4b55d8..11a31ff77aef0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserMemoryManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserMemoryManager.java @@ -19,6 +19,8 @@ package org.apache.iotdb.db.pipe.event.common.tsfile.parser; +import org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException; +import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock; @@ -64,7 +66,17 @@ public long getMemoryUsageInBytes() { @Override public void forceResize(final long newSizeInBytes) { - PipeDataNodeResourceManager.memory().forceResize(delegate, newSizeInBytes); + final long oldSize = delegate.getMemoryUsageInBytes(); + if (!PipeDataNodeResourceManager.memory().tryResize(delegate, newSizeInBytes)) { + throw new PipeRuntimeOutOfMemoryCriticalException( + String.format( + DataNodePipeMessages + .PIPE_EXCEPTION_FORCERESIZE_FAILED_TO_ALLOCATE_MEMORY_AFTER_D_RETRIES_TOTAL_8C6948BC, + 0, + PipeDataNodeResourceManager.memory().getTotalNonFloatingMemorySizeInBytes(), + PipeDataNodeResourceManager.memory().getUsedMemorySizeInBytes(), + newSizeInBytes - oldSize)); + } } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java index 90a45c1542ec3..4e43cc9c35c94 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java @@ -691,6 +691,24 @@ public void forceResize(final PipeMemoryBlock block, final long targetSize) { resize(block, targetSize, true); } + /** + * Attempts a single resize without waiting for other pipe tasks to release memory. + * + *

This is intended for callers that hold payload/batch locks and can actively release memory + * after a failed attempt. Waiting in that situation can prevent the caller itself from making + * forward progress. + */ + public synchronized boolean tryResize(final PipeMemoryBlock block, final long targetSize) { + if (targetSize < 0) { + return false; + } + if (block == null || block.isReleased()) { + LOGGER.warn(DataNodePipeMessages.FORCERESIZE_CANNOT_RESIZE_A_NULL_OR_RELEASED); + return false; + } + return tryResizeInternal(block, targetSize); + } + public synchronized void resize( final PipeMemoryBlock block, final long targetSize, final boolean force) { if (block == null || block.isReleased()) { @@ -698,61 +716,18 @@ public synchronized void resize( return; } - if (!PIPE_MEMORY_MANAGEMENT_ENABLED) { - block.setMemoryUsageInBytes(targetSize); + if (tryResizeInternal(block, targetSize)) { return; } - final long oldSize = block.getMemoryUsageInBytes(); - - if (oldSize >= targetSize) { - memoryBlock.release(oldSize - targetSize); - if (block instanceof PipeTabletMemoryBlock) { - usedMemorySizeInBytesOfTablets -= oldSize - targetSize; - } - if (block instanceof PipeTsFileMemoryBlock) { - usedMemorySizeInBytesOfTsFiles -= oldSize - targetSize; - } - block.setMemoryUsageInBytes(targetSize); - - // If no memory is used in the block, we can remove it from the allocated blocks. - if (targetSize == 0) { - allocatedBlocks.remove(block); - } - - notifyNextTsFileParserMemoryReservationInternal(); - this.notifyAll(); - return; - } - - long sizeInBytes = targetSize - oldSize; + final long sizeInBytes = targetSize - block.getMemoryUsageInBytes(); final int memoryAllocateMaxRetries = PIPE_CONFIG.getPipeMemoryAllocateMaxRetries(); for (int i = 1; i <= memoryAllocateMaxRetries; i++) { - // Dynamically resized data-structure blocks must obey the same admission thresholds as - // blocks allocated with a non-zero initial size. Otherwise they can exhaust the pool and - // prevent downstream consumers from allocating the memory needed to release them. - if (isHardEnoughForResizing(block, sizeInBytes) - && getTotalNonFloatingMemorySizeInBytes() - memoryBlock.getUsedMemoryInBytes() - >= sizeInBytes) { - memoryBlock.forceAllocateWithoutLimitation(sizeInBytes); - if (oldSize == 0) { - // If the memory block is not registered, we need to register it first. - // Otherwise, the memory usage will be inconsistent. - // See registerMemoryBlock for more details. - allocatedBlocks.add(block); - } - if (block instanceof PipeTabletMemoryBlock) { - usedMemorySizeInBytesOfTablets += sizeInBytes; - } - if (block instanceof PipeTsFileMemoryBlock) { - usedMemorySizeInBytesOfTsFiles += sizeInBytes; - } - block.setMemoryUsageInBytes(targetSize); - return; - } - try { tryShrinkUntilFreeMemorySatisfy(sizeInBytes); + if (tryResizeInternal(block, targetSize)) { + return; + } this.wait(PIPE_CONFIG.getPipeMemoryAllocateRetryIntervalInMs()); } catch (InterruptedException e) { Thread.currentThread().interrupt(); @@ -773,6 +748,58 @@ && getTotalNonFloatingMemorySizeInBytes() - memoryBlock.getUsedMemoryInBytes() } } + private boolean tryResizeInternal(final PipeMemoryBlock block, final long targetSize) { + if (!PIPE_MEMORY_MANAGEMENT_ENABLED) { + block.setMemoryUsageInBytes(targetSize); + return true; + } + + final long oldSize = block.getMemoryUsageInBytes(); + if (oldSize >= targetSize) { + final long releasedSize = oldSize - targetSize; + memoryBlock.release(releasedSize); + if (block instanceof PipeTabletMemoryBlock) { + usedMemorySizeInBytesOfTablets -= releasedSize; + } + if (block instanceof PipeTsFileMemoryBlock) { + usedMemorySizeInBytesOfTsFiles -= releasedSize; + } + block.setMemoryUsageInBytes(targetSize); + + if (targetSize == 0) { + allocatedBlocks.remove(block); + } + + notifyNextTsFileParserMemoryReservationInternal(); + this.notifyAll(); + return true; + } + + final long sizeInBytes = targetSize - oldSize; + // Dynamically resized data-structure blocks must obey the same admission thresholds as blocks + // allocated with a non-zero initial size. Otherwise they can exhaust the pool and prevent + // downstream consumers from allocating the memory needed to release them. + if (!isHardEnoughForResizing(block, sizeInBytes) + || getTotalNonFloatingMemorySizeInBytes() - memoryBlock.getUsedMemoryInBytes() + < sizeInBytes) { + return false; + } + + memoryBlock.forceAllocateWithoutLimitation(sizeInBytes); + if (oldSize == 0) { + // Zero-sized blocks are registered lazily on their first successful expansion. + allocatedBlocks.add(block); + } + if (block instanceof PipeTabletMemoryBlock) { + usedMemorySizeInBytesOfTablets += sizeInBytes; + } + if (block instanceof PipeTsFileMemoryBlock) { + usedMemorySizeInBytesOfTsFiles += sizeInBytes; + } + block.setMemoryUsageInBytes(targetSize); + return true; + } + /** * Allocate a {@link PipeMemoryBlock} for pipe only if memory used after allocation is less than * the specified threshold. diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFilePublicResource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFilePublicResource.java index fe54ee48b547a..1ec61549040c9 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFilePublicResource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFilePublicResource.java @@ -38,7 +38,6 @@ import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.Objects; public class PipeTsFilePublicResource extends PipeTsFileResource { private static final Logger LOGGER = LoggerFactory.getLogger(PipeTsFilePublicResource.class); @@ -53,24 +52,15 @@ public PipeTsFilePublicResource(File hardlinkOrCopiedFile) { } @Override - public void close() { + public synchronized void close() { super.close(); - - if (deviceMeasurementsMap != null) { - deviceMeasurementsMap = null; - } - - if (deviceIsAlignedMap != null) { - deviceIsAlignedMap = null; - } - - if (measurementDataTypeMap != null) { - measurementDataTypeMap = null; - } - - if (allocatedMemoryBlock != null) { - allocatedMemoryBlock.close(); - allocatedMemoryBlock = null; + deviceMeasurementsMap = null; + deviceIsAlignedMap = null; + measurementDataTypeMap = null; + final PipeMemoryBlock block = allocatedMemoryBlock; + allocatedMemoryBlock = null; + if (block != null) { + block.close(); } } @@ -114,46 +104,53 @@ synchronized boolean cacheDeviceIsAlignedMapIfAbsent(final File tsFile) throws I // See if pipe memory is sufficient to be allocated for TsFileSequenceReader. // Only allocate when pipe memory used is less than 50%, because memory here // is hard to shrink and may consume too much memory. - allocatedMemoryBlock = + final PipeMemoryBlock readerMemoryBlock = PipeDataNodeResourceManager.memory() .forceAllocateIfSufficient( PipeConfig.getInstance().getPipeMemoryAllocateForTsFileSequenceReaderInBytes(), MEMORY_SUFFICIENT_THRESHOLD); - if (allocatedMemoryBlock == null) { + if (readerMemoryBlock == null) { LOGGER.info( DataNodePipeMessages.FAILED_TO_CACHEDEVICEISALIGNEDMAPIFABSENT_FOR_TSFILE_BECAUSE_MEMORY, tsFile.getPath()); return false; } + final Map cachedDeviceIsAlignedMap = new HashMap<>(); long memoryRequiredInBytes = 0L; - try (TsFileSequenceReader sequenceReader = - new TsFileSequenceReader(tsFile.getPath(), true, false)) { - deviceIsAlignedMap = new HashMap<>(); - final TsFileDeviceIterator deviceIsAlignedIterator = - sequenceReader.getAllDevicesIteratorWithIsAligned(); - while (deviceIsAlignedIterator.hasNext()) { - final Pair deviceIsAlignedPair = deviceIsAlignedIterator.next(); - deviceIsAlignedMap.put(deviceIsAlignedPair.getLeft(), deviceIsAlignedPair.getRight()); + try { + try (TsFileSequenceReader sequenceReader = + new TsFileSequenceReader(tsFile.getPath(), true, false)) { + final TsFileDeviceIterator deviceIsAlignedIterator = + sequenceReader.getAllDevicesIteratorWithIsAligned(); + while (deviceIsAlignedIterator.hasNext()) { + final Pair deviceIsAlignedPair = deviceIsAlignedIterator.next(); + cachedDeviceIsAlignedMap.put( + deviceIsAlignedPair.getLeft(), deviceIsAlignedPair.getRight()); + } } - memoryRequiredInBytes += PipeMemoryWeightUtil.memoryOfIDeviceId2Bool(deviceIsAlignedMap); + memoryRequiredInBytes += + PipeMemoryWeightUtil.memoryOfIDeviceId2Bool(cachedDeviceIsAlignedMap); + } finally { + // The reader block is temporary and must never become the persistent metadata block. + readerMemoryBlock.close(); } - // Release memory of TsFileSequenceReader. - allocatedMemoryBlock.close(); - allocatedMemoryBlock = null; // Allocate again for the cached objects. - allocatedMemoryBlock = + final PipeMemoryBlock cachedMemoryBlock = PipeDataNodeResourceManager.memory() .forceAllocateIfSufficient(memoryRequiredInBytes, MEMORY_SUFFICIENT_THRESHOLD); - if (allocatedMemoryBlock == null) { + if (cachedMemoryBlock == null) { LOGGER.info( DataNodePipeMessages.PIPETSFILERESOURCE_FAILED_TO_CACHE_OBJECTS_FOR_TSFILE, tsFile.getPath()); - deviceIsAlignedMap = null; return false; } + // Publish the map only after its accounting block has been acquired. Readers never observe + // a partially built map or a map without a corresponding memory reservation. + deviceIsAlignedMap = cachedDeviceIsAlignedMap; + allocatedMemoryBlock = cachedMemoryBlock; LOGGER.info( DataNodePipeMessages.PIPETSFILERESOURCE_CACHED_DEVICEISALIGNEDMAP_FOR_TSFILE, tsFile.getPath()); @@ -166,65 +163,75 @@ synchronized boolean cacheObjectsIfAbsent(final File tsFile) throws IOException return true; } else { // Recalculate it again because only deviceIsAligned map is cached - allocatedMemoryBlock.close(); + final PipeMemoryBlock oldMemoryBlock = allocatedMemoryBlock; allocatedMemoryBlock = null; + deviceIsAlignedMap = null; + oldMemoryBlock.close(); } } // See if pipe memory is sufficient to be allocated for TsFileSequenceReader. // Only allocate when pipe memory used is less than 50%, because memory here // is hard to shrink and may consume too much memory. - allocatedMemoryBlock = + final PipeMemoryBlock readerMemoryBlock = PipeDataNodeResourceManager.memory() .forceAllocateIfSufficient( PipeConfig.getInstance().getPipeMemoryAllocateForTsFileSequenceReaderInBytes(), MEMORY_SUFFICIENT_THRESHOLD); - if (allocatedMemoryBlock == null) { + if (readerMemoryBlock == null) { LOGGER.info( DataNodePipeMessages.FAILED_TO_CACHEOBJECTSIFABSENT_FOR_TSFILE_BECAUSE_MEMORY, tsFile.getPath()); return false; } + Map> cachedDeviceMeasurementsMap = null; + Map cachedDeviceIsAlignedMap = null; + Map cachedMeasurementDataTypeMap = null; long memoryRequiredInBytes = 0L; - try (TsFileSequenceReader sequenceReader = - new TsFileSequenceReader(tsFile.getPath(), true, true)) { - deviceMeasurementsMap = sequenceReader.getDeviceMeasurementsMap(); - memoryRequiredInBytes += - PipeMemoryWeightUtil.memoryOfIDeviceID2StrList(deviceMeasurementsMap); - - if (Objects.isNull(deviceIsAlignedMap)) { - deviceIsAlignedMap = new HashMap<>(); + try { + try (TsFileSequenceReader sequenceReader = + new TsFileSequenceReader(tsFile.getPath(), true, true)) { + cachedDeviceMeasurementsMap = sequenceReader.getDeviceMeasurementsMap(); + memoryRequiredInBytes += + PipeMemoryWeightUtil.memoryOfIDeviceID2StrList(cachedDeviceMeasurementsMap); + + cachedDeviceIsAlignedMap = new HashMap<>(); final TsFileDeviceIterator deviceIsAlignedIterator = sequenceReader.getAllDevicesIteratorWithIsAligned(); while (deviceIsAlignedIterator.hasNext()) { final Pair deviceIsAlignedPair = deviceIsAlignedIterator.next(); - deviceIsAlignedMap.put(deviceIsAlignedPair.getLeft(), deviceIsAlignedPair.getRight()); + cachedDeviceIsAlignedMap.put( + deviceIsAlignedPair.getLeft(), deviceIsAlignedPair.getRight()); } - } - memoryRequiredInBytes += PipeMemoryWeightUtil.memoryOfIDeviceId2Bool(deviceIsAlignedMap); + memoryRequiredInBytes += + PipeMemoryWeightUtil.memoryOfIDeviceId2Bool(cachedDeviceIsAlignedMap); - measurementDataTypeMap = sequenceReader.getFullPathDataTypeMap(); - memoryRequiredInBytes += PipeMemoryWeightUtil.memoryOfStr2TSDataType(measurementDataTypeMap); + cachedMeasurementDataTypeMap = sequenceReader.getFullPathDataTypeMap(); + memoryRequiredInBytes += + PipeMemoryWeightUtil.memoryOfStr2TSDataType(cachedMeasurementDataTypeMap); + } + } finally { + // The reader block is temporary and must be released even when metadata traversal fails. + readerMemoryBlock.close(); } - // Release memory of TsFileSequenceReader. - allocatedMemoryBlock.close(); - allocatedMemoryBlock = null; // Allocate again for the cached objects. - allocatedMemoryBlock = + final PipeMemoryBlock cachedMemoryBlock = PipeDataNodeResourceManager.memory() .forceAllocateIfSufficient(memoryRequiredInBytes, MEMORY_SUFFICIENT_THRESHOLD); - if (allocatedMemoryBlock == null) { + if (cachedMemoryBlock == null) { LOGGER.info( DataNodePipeMessages.PIPETSFILERESOURCE_FAILED_TO_CACHE_OBJECTS_FOR_TSFILE, tsFile.getPath()); - deviceIsAlignedMap = null; - deviceMeasurementsMap = null; - measurementDataTypeMap = null; return false; } + // Publish all metadata only after the persistent accounting block is ready. + deviceMeasurementsMap = cachedDeviceMeasurementsMap; + deviceIsAlignedMap = cachedDeviceIsAlignedMap; + measurementDataTypeMap = cachedMeasurementDataTypeMap; + allocatedMemoryBlock = cachedMemoryBlock; LOGGER.info( DataNodePipeMessages.PIPETSFILERESOURCE_CACHED_OBJECTS_FOR_TSFILE, tsFile.getPath()); return true; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java index e325e4f0cf858..db98166368128 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java @@ -150,9 +150,9 @@ private File increaseFileReference( } try { increasePublicReference(resultFile, pipeName, isTsFile); - } catch (final IOException e) { + } catch (final IOException | RuntimeException e) { // The private reference must not outlive a failed public reference increase. - decreaseFileReference(resultFile, pipeName, false); + rollbackFileReference(resultFile, pipeName, e); throw e; } return resultFile; @@ -172,10 +172,27 @@ private boolean increaseReferenceIfExists( } finally { segmentLock.unlock(file); } - increasePublicReference(file, pipeName, isTsFile); + try { + increasePublicReference(file, pipeName, isTsFile); + } catch (final IOException | RuntimeException e) { + // The private reference is acquired before the public (assigner) reference. If the latter + // fails, roll back the reference acquired above; otherwise every failed retry permanently + // pins the pipe file in the logical memory/file pool. + rollbackFileReference(file, pipeName, e); + throw e; + } return true; } + private void rollbackFileReference( + final File file, final @Nullable String pipeName, final Exception originalException) { + try { + decreaseFileReference(file, pipeName, false); + } catch (final RuntimeException rollbackException) { + originalException.addSuppressed(rollbackException); + } + } + private void increasePublicReference( final File file, final @Nullable String pipeName, final boolean isTsFile) throws IOException { if (Objects.isNull(pipeName)) { @@ -413,9 +430,20 @@ public Map getMeasurementDataTypeMapFromCache( public void pinTsFileResource( final TsFileResource resource, final boolean withMods, final @Nullable String pipeName) throws IOException { - increaseFileReference(resource.getTsFile(), true, pipeName); - if (withMods && resource.getExclusiveModFile().exists()) { - increaseFileReference(resource.getExclusiveModFile().getFile(), false, pipeName); + final File pinnedTsFile = increaseFileReference(resource.getTsFile(), true, pipeName); + try { + if (withMods && resource.getExclusiveModFile().exists()) { + increaseFileReference(resource.getExclusiveModFile().getFile(), false, pipeName); + } + } catch (final IOException | RuntimeException e) { + // Pinning is a two-file operation. Do not leave the TsFile pinned when the mod file cannot + // be pinned (for example, when the pipe directory is temporarily unavailable). + try { + decreaseFileReference(pinnedTsFile, pipeName); + } catch (final RuntimeException rollbackException) { + e.addSuppressed(rollbackException); + } + throw e; } } @@ -424,13 +452,35 @@ public void unpinTsFileResource( final boolean shouldTransferModFile, final @Nullable String pipeName) throws IOException { - decreaseFileReference( - getHardlinkOrCopiedFileInPipeDir(resource.getTsFile(), pipeName), pipeName); + Exception firstException = null; + try { + decreaseFileReference( + getHardlinkOrCopiedFileInPipeDir(resource.getTsFile(), pipeName), pipeName); + } catch (final IOException | RuntimeException e) { + firstException = e; + } + // Always attempt the mod-file cleanup even when resolving/decreasing the TsFile fails. A + // failed first cleanup must not strand the second reference. if (shouldTransferModFile && resource.exclusiveModFileExists()) { - decreaseFileReference( - getHardlinkOrCopiedFileInPipeDir(resource.getExclusiveModFile().getFile(), pipeName), - pipeName); + try { + decreaseFileReference( + getHardlinkOrCopiedFileInPipeDir(resource.getExclusiveModFile().getFile(), pipeName), + pipeName); + } catch (final IOException | RuntimeException e) { + if (firstException == null) { + firstException = e; + } else { + firstException.addSuppressed(e); + } + } + } + + if (firstException != null) { + if (firstException instanceof IOException) { + throw (IOException) firstException; + } + throw (RuntimeException) firstException; } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java index 39d3bea1dbe15..b84985675facf 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java @@ -19,11 +19,13 @@ package org.apache.iotdb.db.pipe.sink.payload.evolvable.batch; +import org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException; import org.apache.iotdb.commons.pipe.agent.task.progress.CommitterKey; import org.apache.iotdb.commons.pipe.event.EnrichedEvent; import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; -import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager; +import org.apache.iotdb.db.pipe.resource.memory.PipeTabletMemoryBlock; import org.apache.iotdb.db.pipe.sink.protocol.thrift.async.IoTDBDataRegionAsyncSink; import org.apache.iotdb.db.storageengine.dataregion.wal.exception.WALPipeException; import org.apache.iotdb.pipe.api.event.Event; @@ -49,7 +51,8 @@ public abstract class PipeTabletEventBatch implements AutoCloseable { private long firstEventProcessingTime = Long.MIN_VALUE; protected long totalBufferSize = 0; - private final PipeMemoryBlock allocatedMemoryBlock; + private final PipeTabletMemoryBlock allocatedMemoryBlock; + private boolean shouldEmitOnMemoryPressure = false; protected volatile boolean isClosed = false; @@ -61,7 +64,8 @@ protected PipeTabletEventBatch( // limit in buffer size this.maxBatchSizeInBytes = requestMaxBatchSizeInBytes; - this.allocatedMemoryBlock = PipeDataNodeResourceManager.memory().forceAllocate(0); + this.allocatedMemoryBlock = + PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(0); if (recordMetric != null) { this.recordMetric = recordMetric; } else { @@ -91,6 +95,11 @@ public synchronized boolean onEvent(final TabletInsertionEvent event) if (((EnrichedEvent) event) .increaseReferenceCount(PipeTransferBatchReqBuilder.class.getName())) { + final int previousEventsSize = events.size(); + final long previousTotalBufferSize = totalBufferSize; + final boolean previousMemoryPressureState = shouldEmitOnMemoryPressure; + final Object batchState = captureBatchState(); + try { if (constructBatch(event)) { events.add((EnrichedEvent) event); @@ -102,10 +111,28 @@ public synchronized boolean onEvent(final TabletInsertionEvent event) .decreaseReferenceCount(PipeTransferBatchReqBuilder.class.getName(), true); } } catch (final Exception e) { - if (events.isEmpty()) { - clearBatchData(); - resetMemoryUsage(); + try { + rollbackBatchState(batchState); + } catch (final Exception rollbackException) { + e.addSuppressed(rollbackException); + } + + // A failed constructBatch must not retain a partial payload or its memory reservation, + // even when older events are already buffered in this batch. + if (totalBufferSize != previousTotalBufferSize) { + // Shrinking never needs to wait. This path still holds the batch monitor, so a + // blocking resize here would recreate the same lock cycle as a failed append. + PipeDataNodeResourceManager.memory() + .tryResize(allocatedMemoryBlock, previousTotalBufferSize); + } + totalBufferSize = previousTotalBufferSize; + shouldEmitOnMemoryPressure = + previousMemoryPressureState + || (e instanceof PipeRuntimeOutOfMemoryCriticalException && !events.isEmpty()); + if (events.size() > previousEventsSize) { + events.subList(previousEventsSize, events.size()).clear(); } + // If the event is not added to the batch, we need to decrease the reference count. ((EnrichedEvent) event) .decreaseReferenceCount(PipeTransferBatchReqBuilder.class.getName(), false); @@ -131,13 +158,34 @@ public synchronized boolean onEvent(final TabletInsertionEvent event) protected abstract boolean constructBatch(final TabletInsertionEvent event) throws WALPipeException, IOException; + /** Captures subclass payload state before constructing one event. */ + protected Object captureBatchState() { + return null; + } + + /** Restores subclass payload state after a failed event construction. */ + protected void rollbackBatchState(final Object state) {} + protected void increaseTotalBufferSizeAndUpdateMemoryBlock(final long bufferSize) { if (bufferSize <= 0) { return; } final long newTotalBufferSize = Math.min(totalBufferSize + bufferSize, maxBatchSizeInBytes); - PipeDataNodeResourceManager.memory().forceResize(allocatedMemoryBlock, newTotalBufferSize); + final PipeMemoryManager memoryManager = PipeDataNodeResourceManager.memory(); + if (!memoryManager.tryResize(allocatedMemoryBlock, newTotalBufferSize)) { + // Existing events must be emitted before this event is retried. Do not wait here: onEvent() + // is called while both the request builder and this batch are locked. + shouldEmitOnMemoryPressure = !events.isEmpty(); + throw new PipeRuntimeOutOfMemoryCriticalException( + String.format( + DataNodePipeMessages + .PIPE_EXCEPTION_FORCERESIZE_FAILED_TO_ALLOCATE_MEMORY_AFTER_D_RETRIES_TOTAL_8C6948BC, + 0, + memoryManager.getTotalNonFloatingMemorySizeInBytes(), + memoryManager.getUsedMemorySizeInBytes(), + newTotalBufferSize - totalBufferSize)); + } totalBufferSize = newTotalBufferSize; } @@ -147,13 +195,20 @@ protected void releaseAllocatedMemoryBlock() { protected void clearBatchData() {} + /** Close resources owned by a batch that will not be reused. */ + protected void closeBatchData() { + clearBatchData(); + } + public boolean shouldEmit() { if (events.isEmpty()) { return false; } final long diff = System.currentTimeMillis() - firstEventProcessingTime; - if (totalBufferSize >= maxBatchSizeInBytes || diff >= maxDelayInMs) { + if (shouldEmitOnMemoryPressure + || totalBufferSize >= maxBatchSizeInBytes + || diff >= maxDelayInMs) { recordMetric.accept(diff, totalBufferSize, events.size()); return true; } @@ -162,8 +217,20 @@ public boolean shouldEmit() { public synchronized void onSuccess() { events.clear(); + try { + clearBatchData(); + } finally { + resetMemoryUsage(); + } + } - resetMemoryUsage(); + /** + * Close a detached asynchronous batch after its event references have been handed to handlers or + * a retry queue. Unlike {@link #close()}, this method does not release those references. + */ + public synchronized void closeAfterEventTransfer() { + events.clear(); + close(); } @Override @@ -173,11 +240,20 @@ public synchronized void close() { } isClosed = true; - clearEventsReferenceCount(PipeTabletEventBatch.class.getName()); - events.clear(); - clearBatchData(); - resetMemoryUsage(); - allocatedMemoryBlock.close(); + try { + clearEventsReferenceCount(PipeTabletEventBatch.class.getName()); + } finally { + events.clear(); + try { + closeBatchData(); + } finally { + try { + resetMemoryUsage(); + } finally { + allocatedMemoryBlock.close(); + } + } + } } /** @@ -207,6 +283,7 @@ public synchronized void discardEventsOfPipe(final CommitterKey committerKey) { private void resetMemoryUsage() { totalBufferSize = 0; + shouldEmitOnMemoryPressure = false; releaseAllocatedMemoryBlock(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventPlainBatch.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventPlainBatch.java index 3eec34cc94bd8..cb367a1949454 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventPlainBatch.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventPlainBatch.java @@ -83,13 +83,6 @@ protected boolean constructBatch(final TabletInsertionEvent event) throws IOExce return true; } - @Override - public synchronized void onSuccess() { - clearBatchData(); - - super.onSuccess(); - } - @Override protected void clearBatchData() { insertNodeBuffers.clear(); @@ -102,6 +95,85 @@ protected void clearBatchData() { pipe2BytesAccumulated.clear(); } + @Override + protected Object captureBatchState() { + final Map>>> tableModelTabletMapSnapshot = + new HashMap<>(); + tableModelTabletMap.forEach( + (database, tableMap) -> { + final Map>> tableMapSnapshot = new HashMap<>(); + tableMap.forEach( + (table, tablets) -> + tableMapSnapshot.put( + table, new Pair<>(tablets.getLeft(), new ArrayList<>(tablets.getRight())))); + tableModelTabletMapSnapshot.put(database, tableMapSnapshot); + }); + return new BatchState( + insertNodeBuffers.size(), + tabletBuffers.size(), + insertNodeDataBases.size(), + tabletDataBases.size(), + tableModelTabletMapSnapshot, + new HashMap<>(pipe2BytesAccumulated)); + } + + @Override + @SuppressWarnings("unchecked") + protected void rollbackBatchState(final Object state) { + if (!(state instanceof BatchState)) { + return; + } + final BatchState batchState = (BatchState) state; + truncate(insertNodeBuffers, batchState.insertNodeBuffersSize); + truncate(tabletBuffers, batchState.tabletBuffersSize); + truncate(insertNodeDataBases, batchState.insertNodeDataBasesSize); + truncate(tabletDataBases, batchState.tabletDataBasesSize); + + tableModelTabletMap.clear(); + batchState.tableModelTabletMap.forEach( + (database, tableMap) -> { + final Map>> restoredTableMap = new HashMap<>(); + tableMap.forEach( + (table, tablets) -> + restoredTableMap.put( + table, new Pair<>(tablets.getLeft(), new ArrayList<>(tablets.getRight())))); + tableModelTabletMap.put(database, restoredTableMap); + }); + + pipe2BytesAccumulated.clear(); + pipe2BytesAccumulated.putAll(batchState.pipe2BytesAccumulated); + } + + private static void truncate(final List list, final int size) { + if (list.size() > size) { + list.subList(size, list.size()).clear(); + } + } + + private static final class BatchState { + private final int insertNodeBuffersSize; + private final int tabletBuffersSize; + private final int insertNodeDataBasesSize; + private final int tabletDataBasesSize; + private final Map>>> tableModelTabletMap; + private final Map, Long> pipe2BytesAccumulated; + + private BatchState( + final int insertNodeBuffersSize, + final int tabletBuffersSize, + final int insertNodeDataBasesSize, + final int tabletDataBasesSize, + final Map>>> tableModelTabletMap, + final Map, Long> pipe2BytesAccumulated) { + this.insertNodeBuffersSize = insertNodeBuffersSize; + this.tabletBuffersSize = tabletBuffersSize; + this.insertNodeDataBasesSize = insertNodeDataBasesSize; + this.tabletDataBasesSize = tabletDataBasesSize; + this.tableModelTabletMap = tableModelTabletMap; + this.pipe2BytesAccumulated = pipe2BytesAccumulated; + } + } + public PipeTransferTabletBatchReqV2 toTPipeTransferReq() throws IOException { for (final Map.Entry>>> insertTablets : tableModelTabletMap.entrySet()) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatch.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatch.java index a0046e82e3e25..4793f0cbfd1c6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatch.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatch.java @@ -95,72 +95,84 @@ public PipeTabletEventTsFileBatch( @Override protected boolean constructBatch(final TabletInsertionEvent event) { - boolean hasBufferedTablet = false; if (event instanceof PipeInsertNodeTabletInsertionEvent) { final PipeInsertNodeTabletInsertionEvent insertNodeTabletInsertionEvent = (PipeInsertNodeTabletInsertionEvent) event; final boolean isTableModel = insertNodeTabletInsertionEvent.isTableModelEvent(); final List tablets = insertNodeTabletInsertionEvent.convertToTablets(); - increaseTotalBufferSizeAndUpdateMemoryBlock(calculateTabletsSizeInBytes(tablets)); + final List retainedTablets = new ArrayList<>(tablets.size()); + final List retainedAlignedFlags = new ArrayList<>(tablets.size()); for (int i = 0; i < tablets.size(); ++i) { - final Tablet tablet = tablets.get(i); + Tablet tablet = tablets.get(i); if (isTabletEmpty(tablet)) { continue; } if (isTableModel) { - // table Model - final Tablet prunedTablet = + tablet = pruneTableModelTablet( tablet, insertNodeTabletInsertionEvent.getTableModelDatabaseName()); - if (isTabletEmpty(prunedTablet)) { + if (isTabletEmpty(tablet)) { continue; } + } + retainedTablets.add(tablet); + if (!isTableModel) { + retainedAlignedFlags.add(insertNodeTabletInsertionEvent.isAligned(i)); + } + } + + // Pruning can remove all rows/columns from a tablet. Account only for data that is + // actually retained; otherwise a fully (or partially) pruned event permanently inflates the + // batch's memory block and can starve TsFile conversion buffers. + if (retainedTablets.isEmpty()) { + return false; + } + increaseTotalBufferSizeAndUpdateMemoryBlock(calculateTabletsSizeInBytes(retainedTablets)); + for (int i = 0; i < retainedTablets.size(); ++i) { + final Tablet tablet = retainedTablets.get(i); + if (isTableModel) { bufferTableModelTablet( insertNodeTabletInsertionEvent.getPipeName(), insertNodeTabletInsertionEvent.getCreationTime(), - prunedTablet, + tablet, insertNodeTabletInsertionEvent.getTableModelDatabaseName()); - hasBufferedTablet = true; } else { - // tree Model bufferTreeModelTablet( insertNodeTabletInsertionEvent.getPipeName(), insertNodeTabletInsertionEvent.getCreationTime(), tablet, - insertNodeTabletInsertionEvent.isAligned(i)); - hasBufferedTablet = true; + retainedAlignedFlags.get(i)); } } + return true; } else if (event instanceof PipeRawTabletInsertionEvent) { final PipeRawTabletInsertionEvent rawTabletInsertionEvent = (PipeRawTabletInsertionEvent) event; - final Tablet tablet = rawTabletInsertionEvent.convertToTablet(); + Tablet tablet = rawTabletInsertionEvent.convertToTablet(); if (isTabletEmpty(tablet)) { return false; } - increaseTotalBufferSizeAndUpdateMemoryBlock(calculateTabletSizeInBytes(tablet)); if (rawTabletInsertionEvent.isTableModelEvent()) { - // table Model - final Tablet prunedTablet = - pruneTableModelTablet(tablet, rawTabletInsertionEvent.getTableModelDatabaseName()); - if (isTabletEmpty(prunedTablet)) { + tablet = pruneTableModelTablet(tablet, rawTabletInsertionEvent.getTableModelDatabaseName()); + if (isTabletEmpty(tablet)) { return false; } + } + increaseTotalBufferSizeAndUpdateMemoryBlock(calculateTabletSizeInBytes(tablet)); + if (rawTabletInsertionEvent.isTableModelEvent()) { bufferTableModelTablet( rawTabletInsertionEvent.getPipeName(), rawTabletInsertionEvent.getCreationTime(), - prunedTablet, + tablet, rawTabletInsertionEvent.getTableModelDatabaseName()); - hasBufferedTablet = true; } else { - // tree Model bufferTreeModelTablet( rawTabletInsertionEvent.getPipeName(), rawTabletInsertionEvent.getCreationTime(), tablet, rawTabletInsertionEvent.isAligned()); - hasBufferedTablet = true; } + return true; } else { LOGGER.warn( DataNodePipeMessages.BATCH_ID_UNSUPPORTED_EVENT_TYPE_WHEN_CONSTRUCTING, @@ -168,7 +180,7 @@ protected boolean constructBatch(final TabletInsertionEvent event) { event, event.getClass()); } - return hasBufferedTablet; + return false; } private Tablet pruneTableModelTablet(final Tablet tablet, final String databaseName) { @@ -188,6 +200,41 @@ private static long calculateTabletSizeInBytes(final Tablet tablet) { return PipeMemoryWeightUtil.calculateTabletSizeInBytes(tablet) * 2; } + @Override + public Object captureBatchState() { + return new BatchState( + treeModeTsFileBuilder.createCheckpoint(), + tableModeTsFileBuilder.createCheckpoint(), + new HashMap<>(pipeName2WeightMap)); + } + + @Override + public void rollbackBatchState(final Object state) { + if (!(state instanceof BatchState)) { + return; + } + final BatchState batchState = (BatchState) state; + treeModeTsFileBuilder.rollbackToCheckpoint(batchState.treeModeCheckpoint); + tableModeTsFileBuilder.rollbackToCheckpoint(batchState.tableModeCheckpoint); + pipeName2WeightMap.clear(); + pipeName2WeightMap.putAll(batchState.pipeName2WeightMap); + } + + private static final class BatchState { + private final Object treeModeCheckpoint; + private final Object tableModeCheckpoint; + private final Map, Double> pipeName2WeightMap; + + private BatchState( + final Object treeModeCheckpoint, + final Object tableModeCheckpoint, + final Map, Double> pipeName2WeightMap) { + this.treeModeCheckpoint = treeModeCheckpoint; + this.tableModeCheckpoint = tableModeCheckpoint; + this.pipeName2WeightMap = pipeName2WeightMap; + } + } + private void bufferTreeModelTablet( final String pipeName, final long creationTime, @@ -258,13 +305,6 @@ public synchronized List> sealTsFiles() } } - @Override - public synchronized void onSuccess() { - clearBatchData(); - - super.onSuccess(); - } - @Override protected void clearBatchData() { pipeName2WeightMap.clear(); @@ -273,10 +313,12 @@ protected void clearBatchData() { } @Override - public synchronized void close() { - super.close(); - - tableModeTsFileBuilder.close(); - treeModeTsFileBuilder.close(); + protected void closeBatchData() { + pipeName2WeightMap.clear(); + try { + tableModeTsFileBuilder.close(); + } finally { + treeModeTsFileBuilder.close(); + } } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilder.java index ce70cf6f6e33e..45ec6505390fa 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilder.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilder.java @@ -67,6 +67,7 @@ public class PipeTransferBatchReqBuilder implements AutoCloseable { private static final Logger LOGGER = LoggerFactory.getLogger(PipeTransferBatchReqBuilder.class); + private final boolean usingTsFileBatch; private final boolean useLeaderCache; private final int requestMaxDelayInMs; @@ -81,14 +82,14 @@ public class PipeTransferBatchReqBuilder implements AutoCloseable { // If the leader cache is disabled (or unable to find the endpoint of event in the leader cache), // the event will be stored in the default batch. - private final PipeTabletEventBatch defaultBatch; + private PipeTabletEventBatch defaultBatch; // If the leader cache is enabled, the batch will be divided by the leader endpoint, // each endpoint has a batch. // This is only used in plain batch since tsfile does not return redirection info. private final Map endPointToBatch = new HashMap<>(); public PipeTransferBatchReqBuilder(final PipeParameters parameters) { - final boolean usingTsFileBatch = + usingTsFileBatch = parameters .getStringOrDefault( Arrays.asList(CONNECTOR_FORMAT_KEY, SINK_FORMAT_KEY), CONNECTOR_FORMAT_HYBRID_VALUE) @@ -121,12 +122,20 @@ public PipeTransferBatchReqBuilder(final PipeParameters parameters) { usingTsFileBatch ? CONNECTOR_IOTDB_TS_FILE_BATCH_SIZE_DEFAULT_VALUE : CONNECTOR_IOTDB_PLAIN_BATCH_SIZE_DEFAULT_VALUE); - this.defaultBatch = - usingTsFileBatch - ? new PipeTabletEventTsFileBatch( - requestMaxDelayInMs, requestMaxBatchSizeInBytes, this::recordTsFileMetric) - : new PipeTabletEventPlainBatch( - requestMaxDelayInMs, requestMaxBatchSizeInBytes, this::recordTabletMetric); + this.defaultBatch = createDefaultBatch(); + } + + private PipeTabletEventBatch createDefaultBatch() { + return usingTsFileBatch + ? new PipeTabletEventTsFileBatch( + requestMaxDelayInMs, requestMaxBatchSizeInBytes, this::recordTsFileMetric) + : new PipeTabletEventPlainBatch( + requestMaxDelayInMs, requestMaxBatchSizeInBytes, this::recordTabletMetric); + } + + private PipeTabletEventPlainBatch createLeaderCacheBatch() { + return new PipeTabletEventPlainBatch( + requestMaxDelayInMs, requestMaxBatchSizeInBytes, this::recordTabletMetric); } /** @@ -194,6 +203,59 @@ public synchronized void onEvent(final TabletInsertionEvent event) return nonEmptyAndShouldEmitBatches; } + /** + * Atomically detaches every batch that is ready to emit. + * + *

The detached batches are immutable from the builder's point of view: subsequent events are + * appended to fresh batches. This is required for asynchronous sinks, whose completion callback + * may clear a batch after another sink thread has already appended a new event to it. + */ + public synchronized List> + getAllNonEmptyAndShouldEmitBatchesAndDetach() { + final List> batches = + new ArrayList<>(endPointToBatch.size() + 1); + if (!defaultBatch.isEmpty() && defaultBatch.shouldEmit()) { + batches.add(new Pair<>(null, defaultBatch)); + } + + for (final Map.Entry entry : endPointToBatch.entrySet()) { + final PipeTabletEventPlainBatch batch = entry.getValue(); + if (!batch.isEmpty() && batch.shouldEmit()) { + batches.add(new Pair<>(entry.getKey(), batch)); + } + } + + // Construct all replacement batches before changing the builder mappings. If construction of + // one replacement fails (for example while creating a TsFile batch directory), the old + // batches remain owned by this builder and can still be retried or closed by the caller. + final List replacements = new ArrayList<>(batches.size()); + try { + for (final Pair batch : batches) { + replacements.add(batch.getLeft() == null ? createDefaultBatch() : createLeaderCacheBatch()); + } + } catch (final RuntimeException | Error e) { + replacements.forEach( + replacement -> { + try { + replacement.close(); + } catch (final RuntimeException | Error closeException) { + e.addSuppressed(closeException); + } + }); + throw e; + } + + for (int i = 0; i < batches.size(); ++i) { + final Pair batch = batches.get(i); + if (batch.getLeft() == null) { + defaultBatch = replacements.get(i); + } else { + endPointToBatch.put(batch.getLeft(), (PipeTabletEventPlainBatch) replacements.get(i)); + } + } + return batches; + } + public synchronized boolean isEmpty() { if (!defaultBatch.isEmpty()) { return false; @@ -206,6 +268,23 @@ public synchronized boolean isEmpty() { return true; } + /** Returns whether a specific event is still retained by one of the current batches. */ + public synchronized boolean containsEvent(final Event event) { + for (final EnrichedEvent batchedEvent : defaultBatch.events) { + if (batchedEvent == event) { + return true; + } + } + for (final PipeTabletEventPlainBatch batch : endPointToBatch.values()) { + for (final EnrichedEvent batchedEvent : batch.events) { + if (batchedEvent == event) { + return true; + } + } + } + return false; + } + public synchronized void discardEventsOfPipe( final String pipeNameToDrop, final long creationTimeToDrop, final int regionId) { discardEventsOfPipe(new CommitterKey(pipeNameToDrop, creationTimeToDrop, regionId, -1)); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java index ff624204e2478..6ef1c47ad2e11 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java @@ -20,6 +20,7 @@ package org.apache.iotdb.db.pipe.sink.protocol.airgap; import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException; import org.apache.iotdb.commons.pipe.agent.task.progress.CommitterKey; import org.apache.iotdb.commons.pipe.event.EnrichedEvent; import org.apache.iotdb.commons.pipe.sink.limiter.TsFileSendRateLimiter; @@ -129,7 +130,17 @@ public void transfer(final TabletInsertionEvent tabletInsertionEvent) throws Exc // We need to restore the transfer quickly by retry under this circumstance socket.setSoTimeout(PIPE_CONFIG.getPipeAirGapSinkTabletTimeoutMs()); if (isTabletBatchModeEnabled) { - tabletBatchBuilder.onEvent(tabletInsertionEvent); + try { + tabletBatchBuilder.onEvent(tabletInsertionEvent); + } catch (final PipeRuntimeOutOfMemoryCriticalException memoryException) { + try { + doTransferWrapper(socket); + } catch (final Exception transferException) { + transferException.addSuppressed(memoryException); + throw transferException; + } + tabletBatchBuilder.onEvent(tabletInsertionEvent); + } doTransferWrapper(socket); } else if (tabletInsertionEvent instanceof PipeInsertNodeTabletInsertionEvent) { doTransferWrapper(socket, (PipeInsertNodeTabletInsertionEvent) tabletInsertionEvent); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/payload/builder/IoTConsensusV2TransferBatchReqBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/payload/builder/IoTConsensusV2TransferBatchReqBuilder.java index 8c9e0299f31ff..f60ce1de8fe7c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/payload/builder/IoTConsensusV2TransferBatchReqBuilder.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/payload/builder/IoTConsensusV2TransferBatchReqBuilder.java @@ -21,6 +21,7 @@ import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId; import org.apache.iotdb.commons.consensus.index.ProgressIndex; +import org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException; import org.apache.iotdb.commons.pipe.event.EnrichedEvent; import org.apache.iotdb.consensus.iotconsensusv2.thrift.TCommitId; import org.apache.iotdb.consensus.iotconsensusv2.thrift.TIoTConsensusV2TransferReq; @@ -163,7 +164,16 @@ private void increaseTotalBufferSizeAndUpdateMemoryBlock(final long bufferSize) final long newTotalBufferSize = Math.min(totalBufferSize + bufferSize, getMaxBatchSizeInBytes()); - PipeDataNodeResourceManager.memory().forceResize(allocatedMemoryBlock, newTotalBufferSize); + if (!PipeDataNodeResourceManager.memory().tryResize(allocatedMemoryBlock, newTotalBufferSize)) { + throw new PipeRuntimeOutOfMemoryCriticalException( + String.format( + DataNodePipeMessages + .PIPE_EXCEPTION_FORCERESIZE_FAILED_TO_ALLOCATE_MEMORY_AFTER_D_RETRIES_TOTAL_8C6948BC, + 0, + PipeDataNodeResourceManager.memory().getTotalNonFloatingMemorySizeInBytes(), + PipeDataNodeResourceManager.memory().getUsedMemorySizeInBytes(), + newTotalBufferSize - totalBufferSize)); + } totalBufferSize = newTotalBufferSize; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java index 2102f7430596e..332ceeea44543 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java @@ -24,6 +24,7 @@ import org.apache.iotdb.commons.audit.UserEntity; import org.apache.iotdb.commons.client.ThriftClient; import org.apache.iotdb.commons.client.async.AsyncPipeDataTransferServiceClient; +import org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException; import org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkNonReportTimeConfigurableException; import org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkResourceException; import org.apache.iotdb.commons.pipe.agent.task.progress.CommitterKey; @@ -99,6 +100,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; +import java.util.stream.Collectors; import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_ENABLE_SEND_TSFILE_LIMIT; import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_ENABLE_SEND_TSFILE_LIMIT_DEFAULT_VALUE; @@ -136,11 +138,17 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink implements PipeSinkWithS private final BlockingQueue retryTsFileQueue = new LinkedBlockingQueue<>(); private final PipeDataRegionEventCounter retryEventQueueEventCounter = new PipeDataRegionEventCounter(); - // Guarded by this. Events need identity semantics because the same payload may compare equal. + // Guarded by this. The map is also the retry-queue membership index. Events need identity + // semantics because the same payload may compare equal. private final Map retryEvent2ResourceFailureType = new IdentityHashMap<>(); // Keep only the latest text to avoid retaining the complete exception chain for every event. private volatile String lastRetryFailureMessage; + // Events removed from the retry queue remain in this map while their next transfer is being + // started. The set contains handlers created by that attempt. A callback from an older handler + // is ignored instead of creating a second queue entry. + private final Map> retryingEvent2Handlers = + new IdentityHashMap<>(); private IoTDBDataNodeAsyncClientManager clientManager; private IoTDBDataNodeAsyncClientManager transferTsFileClientManager; @@ -266,13 +274,31 @@ public void transfer(final TabletInsertionEvent tabletInsertionEvent) throws Exc } if (isTabletBatchModeEnabled) { - tabletBatchBuilder.onEvent(tabletInsertionEvent); - transferBatchedEventsIfNecessary(); + addTabletEventToBatchAndTransferIfNecessary(tabletInsertionEvent); } else { transferInEventWithoutCheck(tabletInsertionEvent); } } + private void addTabletEventToBatchAndTransferIfNecessary( + final TabletInsertionEvent tabletInsertionEvent) throws Exception { + try { + tabletBatchBuilder.onEvent(tabletInsertionEvent); + } catch (final PipeRuntimeOutOfMemoryCriticalException memoryException) { + // The current event was not retained. Flush the already buffered events, whose batch was + // marked ready by the failed non-blocking resize, then retry the current event once after + // that memory has been released. + try { + transferBatchedEventsIfNecessary(); + } catch (final Exception transferException) { + transferException.addSuppressed(memoryException); + throw transferException; + } + tabletBatchBuilder.onEvent(tabletInsertionEvent); + } + transferBatchedEventsIfNecessary(); + } + private void transferInBatchWithoutCheck( final Pair endPointAndBatch) throws IOException, WriteProcessException { @@ -281,63 +307,119 @@ private void transferInBatchWithoutCheck( } final PipeTabletEventBatch batch = endPointAndBatch.getRight(); - if (batch instanceof PipeTabletEventPlainBatch) { - transfer( - endPointAndBatch.getLeft(), - new PipeTransferTabletBatchEventHandler((PipeTabletEventPlainBatch) batch, this)); + transferInPlainBatchWithoutCheck( + endPointAndBatch.getLeft(), (PipeTabletEventPlainBatch) batch); } else if (batch instanceof PipeTabletEventTsFileBatch) { - final PipeTabletEventTsFileBatch tsFileBatch = (PipeTabletEventTsFileBatch) batch; - final List> dbTsFilePairs = tsFileBatch.sealTsFiles(); - final Map, Double> pipe2WeightMap = tsFileBatch.deepCopyPipe2WeightMap(); - final List events = tsFileBatch.deepCopyEvents(); + transferInTsFileBatchWithoutCheck((PipeTabletEventTsFileBatch) batch); + } else { + final Exception exception = + new PipeException( + String.format( + DataNodePipeMessages + .EXCEPTION_UNSUPPORTED_BATCH_TYPE_ARG_WHEN_TRANSFERRING_TABLET_INSERTION_EVENT_66153E12, + batch.getClass())); + addFailureEventsToRetryQueue(batch.deepCopyEvents(), exception); + batch.closeAfterEventTransfer(); + } + } + + private void transferInPlainBatchWithoutCheck( + final TEndPoint endPoint, final PipeTabletEventPlainBatch batch) { + final List events = batch.deepCopyEvents(); + boolean isBatchClosed = false; + try { + final PipeTransferTabletBatchEventHandler handler = + new PipeTransferTabletBatchEventHandler(batch, this); + trackRetryHandler(handler, events); + // The handler now owns a request snapshot and the event references. Free the detached batch + // before borrowing a client so it cannot occupy memory needed by downstream transfer. + isBatchClosed = true; + batch.closeAfterEventTransfer(); + transfer(endPoint, handler); + } catch (final Exception e) { + addFailureEventsToRetryQueue(events, e); + PipeLogger.log( + LOGGER::warn, + e, + DataNodePipeMessages.FAILED_TO_TRANSFER_TABLETINSERTIONEVENT_BATCH, + events.size(), + events.stream().map(EnrichedEvent::getPipeName).collect(Collectors.toSet())); + } finally { + if (!isBatchClosed) { + batch.closeAfterEventTransfer(); + } + } + } + + private void transferInTsFileBatchWithoutCheck(final PipeTabletEventTsFileBatch batch) { + final List events = batch.deepCopyEvents(); + final AtomicBoolean eventsHadBeenAddedToRetryQueue = new AtomicBoolean(false); + List> dbTsFilePairs = Collections.emptyList(); + int transferredFileCount = 0; + boolean isBatchClosed = false; + try { + dbTsFilePairs = batch.sealTsFiles(); + if (dbTsFilePairs.isEmpty()) { + throw new PipeException( + String.format( + DataNodePipeMessages + .EXCEPTION_FAILED_TO_TRANSFER_TSFILE_BATCH_BECAUSE_NO_TSFILE_WAS_GENERATED_FOR_ARG_CC60CCEB, + batch)); + } + final Map, Double> pipe2WeightMap = batch.deepCopyPipe2WeightMap(); final AtomicInteger eventsReferenceCount = new AtomicInteger(dbTsFilePairs.size()); - final AtomicBoolean eventsHadBeenAddedToRetryQueue = new AtomicBoolean(false); - int transferredFileCount = 0; - try { - for (int outputIndex = 0; outputIndex < dbTsFilePairs.size(); outputIndex++) { - final Pair sealedFile = dbTsFilePairs.get(outputIndex); - transfer( - new PipeTransferTsFileHandler( - this, - pipe2WeightMap, - events, - eventsReferenceCount, - eventsHadBeenAddedToRetryQueue, - sealedFile.right, - null, - false, - sealedFile.left, - outputIndex)); - transferredFileCount++; - } - } catch (final Exception e) { - for (int i = transferredFileCount; i < dbTsFilePairs.size(); i++) { - final Pair untransferredFile = dbTsFilePairs.get(i); - if (!org.apache.iotdb.commons.utils.FileUtils.deleteFileIfExist( - untransferredFile.right)) { - LOGGER.warn( - DataNodePipeMessages.FAILED_TO_DELETE_BATCH_FILE_THIS_FILE, untransferredFile); - } - } - PipeLogger.log( - ignored -> - LOGGER.warn(DataNodePipeMessages.FAILED_TO_TRANSFER_TSFILE_BATCH, dbTsFilePairs, e), - e, - DataNodePipeMessages.FAILED_TO_TRANSFER_TSFILE_BATCH, - dbTsFilePairs); - if (eventsHadBeenAddedToRetryQueue.compareAndSet(false, true)) { - addFailureEventsToRetryQueue(events, e); - } + // Conversion has produced self-contained files. Release all tablet-batch memory before the + // handlers reserve their read buffers. + isBatchClosed = true; + batch.closeAfterEventTransfer(); + + for (int outputIndex = 0; outputIndex < dbTsFilePairs.size(); outputIndex++) { + final Pair sealedFile = dbTsFilePairs.get(outputIndex); + final PipeTransferTsFileHandler handler = + new PipeTransferTsFileHandler( + this, + pipe2WeightMap, + events, + eventsReferenceCount, + eventsHadBeenAddedToRetryQueue, + sealedFile.right, + null, + false, + sealedFile.left, + outputIndex); + trackRetryHandler(handler, events); + transfer(handler); + transferredFileCount++; + } + } catch (final Exception e) { + deleteUntransferredBatchFiles(dbTsFilePairs, transferredFileCount); + if (eventsHadBeenAddedToRetryQueue.compareAndSet(false, true)) { + addFailureEventsToRetryQueue(events, e); + } + final List> filesForLogging = dbTsFilePairs; + PipeLogger.log( + ignored -> + LOGGER.warn(DataNodePipeMessages.FAILED_TO_TRANSFER_TSFILE_BATCH, filesForLogging, e), + e, + DataNodePipeMessages.FAILED_TO_TRANSFER_TSFILE_BATCH, + filesForLogging); + } finally { + if (!isBatchClosed) { + batch.closeAfterEventTransfer(); } - } else { - LOGGER.warn( - DataNodePipeMessages.UNSUPPORTED_BATCH_TYPE_WHEN_TRANSFERRING_TABLET_INSERTION, - batch.getClass()); } + } - endPointAndBatch.getRight().onSuccess(); + private void deleteUntransferredBatchFiles( + final List> dbTsFilePairs, final int transferredFileCount) { + for (int i = transferredFileCount; i < dbTsFilePairs.size(); i++) { + final Pair untransferredFile = dbTsFilePairs.get(i); + if (!org.apache.iotdb.commons.utils.FileUtils.deleteFileIfExist(untransferredFile.right)) { + LOGGER.warn(DataNodePipeMessages.FAILED_TO_DELETE_BATCH_FILE_THIS_FILE, untransferredFile); + } + } } private boolean transferInEventWithoutCheck(final TabletInsertionEvent tabletInsertionEvent) @@ -350,22 +432,33 @@ private boolean transferInEventWithoutCheck(final TabletInsertionEvent tabletIns IoTDBDataRegionAsyncSink.class.getName())) { return false; } - - final InsertNode insertNode = pipeInsertNodeTabletInsertionEvent.getInsertNode(); - final String databaseName = - pipeInsertNodeTabletInsertionEvent.isTableModelEvent() - ? pipeInsertNodeTabletInsertionEvent.getTableModelDatabaseName() - : pipeInsertNodeTabletInsertionEvent.getTreeModelDatabaseName(); - final TPipeTransferReq pipeTransferReq = - compressIfNeeded( - PipeTransferTabletInsertNodeReqV2.toTPipeTransferReq(insertNode, databaseName)); - final PipeTransferTabletInsertNodeEventHandler pipeTransferInsertNodeReqHandler = - new PipeTransferTabletInsertNodeEventHandler( - pipeInsertNodeTabletInsertionEvent, pipeTransferReq, this); - - transfer( - // getDeviceId() may return null for InsertRowsNode - pipeInsertNodeTabletInsertionEvent.getDeviceId(), pipeTransferInsertNodeReqHandler); + boolean handedToHandler = false; + try { + final InsertNode insertNode = pipeInsertNodeTabletInsertionEvent.getInsertNode(); + final String databaseName = + pipeInsertNodeTabletInsertionEvent.isTableModelEvent() + ? pipeInsertNodeTabletInsertionEvent.getTableModelDatabaseName() + : pipeInsertNodeTabletInsertionEvent.getTreeModelDatabaseName(); + final TPipeTransferReq pipeTransferReq = + compressIfNeeded( + PipeTransferTabletInsertNodeReqV2.toTPipeTransferReq(insertNode, databaseName)); + final PipeTransferTabletInsertNodeEventHandler pipeTransferInsertNodeReqHandler = + new PipeTransferTabletInsertNodeEventHandler( + pipeInsertNodeTabletInsertionEvent, pipeTransferReq, this); + trackRetryHandler( + pipeTransferInsertNodeReqHandler, + Collections.singletonList(pipeInsertNodeTabletInsertionEvent)); + handedToHandler = true; + + transfer( + // getDeviceId() may return null for InsertRowsNode + pipeInsertNodeTabletInsertionEvent.getDeviceId(), pipeTransferInsertNodeReqHandler); + } finally { + if (!handedToHandler) { + pipeInsertNodeTabletInsertionEvent.decreaseReferenceCount( + IoTDBDataRegionAsyncSink.class.getName(), false); + } + } } else { // tabletInsertionEvent instanceof PipeRawTabletInsertionEvent final PipeRawTabletInsertionEvent pipeRawTabletInsertionEvent = (PipeRawTabletInsertionEvent) tabletInsertionEvent; @@ -374,20 +467,30 @@ private boolean transferInEventWithoutCheck(final TabletInsertionEvent tabletIns IoTDBDataRegionAsyncSink.class.getName())) { return false; } - - final TPipeTransferReq pipeTransferTabletRawReq = - compressIfNeeded( - PipeTransferTabletRawReqV2.toTPipeTransferReq( - pipeRawTabletInsertionEvent.convertToTablet(), - pipeRawTabletInsertionEvent.isAligned(), - pipeRawTabletInsertionEvent.isTableModelEvent() - ? pipeRawTabletInsertionEvent.getTableModelDatabaseName() - : pipeRawTabletInsertionEvent.getTreeModelDatabaseName())); - final PipeTransferTabletRawEventHandler pipeTransferTabletReqHandler = - new PipeTransferTabletRawEventHandler( - pipeRawTabletInsertionEvent, pipeTransferTabletRawReq, this); - - transfer(pipeRawTabletInsertionEvent.getDeviceId(), pipeTransferTabletReqHandler); + boolean handedToHandler = false; + try { + final TPipeTransferReq pipeTransferTabletRawReq = + compressIfNeeded( + PipeTransferTabletRawReqV2.toTPipeTransferReq( + pipeRawTabletInsertionEvent.convertToTablet(), + pipeRawTabletInsertionEvent.isAligned(), + pipeRawTabletInsertionEvent.isTableModelEvent() + ? pipeRawTabletInsertionEvent.getTableModelDatabaseName() + : pipeRawTabletInsertionEvent.getTreeModelDatabaseName())); + final PipeTransferTabletRawEventHandler pipeTransferTabletReqHandler = + new PipeTransferTabletRawEventHandler( + pipeRawTabletInsertionEvent, pipeTransferTabletRawReq, this); + trackRetryHandler( + pipeTransferTabletReqHandler, Collections.singletonList(pipeRawTabletInsertionEvent)); + handedToHandler = true; + + transfer(pipeRawTabletInsertionEvent.getDeviceId(), pipeTransferTabletReqHandler); + } finally { + if (!handedToHandler) { + pipeRawTabletInsertionEvent.decreaseReferenceCount( + IoTDBDataRegionAsyncSink.class.getName(), false); + } + } } return true; @@ -463,14 +566,14 @@ private boolean transferWithoutCheck(final TsFileInsertionEvent tsFileInsertionE return false; } - // We assume that no exceptions will be thrown after reference count is increased. + PipeTransferTsFileHandler pipeTransferTsFileHandler = null; try { // Just in case. To avoid the case that exception occurred when constructing the handler. if (!pipeTsFileInsertionEvent.getTsFile().exists()) { throw new FileNotFoundException(pipeTsFileInsertionEvent.getTsFile().getAbsolutePath()); } - final PipeTransferTsFileHandler pipeTransferTsFileHandler = + pipeTransferTsFileHandler = new PipeTransferTsFileHandler( this, Collections.singletonMap( @@ -488,10 +591,19 @@ private boolean transferWithoutCheck(final TsFileInsertionEvent tsFileInsertionE pipeTsFileInsertionEvent.isTableModelEvent() ? pipeTsFileInsertionEvent.getTableModelDatabaseName() : pipeTsFileInsertionEvent.getTreeModelDatabaseName()); + trackRetryHandler( + pipeTransferTsFileHandler, Collections.singletonList(pipeTsFileInsertionEvent)); transfer(pipeTransferTsFileHandler); return true; } catch (final Exception e) { + if (pipeTransferTsFileHandler != null) { + try { + pipeTransferTsFileHandler.close(); + } catch (final RuntimeException closeException) { + e.addSuppressed(closeException); + } + } // Just in case. To avoid the case that exception occurred when constructing the handler. pipeTsFileInsertionEvent.decreaseReferenceCount( IoTDBDataRegionAsyncSink.class.getName(), false); @@ -523,7 +635,10 @@ private void transfer(final PipeTransferTsFileHandler pipeTransferTsFileHandler) transferTsFileClientManager.getExecutor()); } catch (final RuntimeException e) { transferTsFileCounter.decrementAndGet(); - throw e; + markSchedulingDelayIfHandshakeFailed(null); + logOnClientException(null, e); + pipeTransferTsFileHandler.onError(e); + return; } if (PipeConfig.getInstance().isTransferTsFileSync()) { @@ -582,7 +697,7 @@ private void transferBatchedEventsIfNecessary() throws IOException, WriteProcess } for (final Pair endPointAndBatch : - tabletBatchBuilder.getAllNonEmptyAndShouldEmitBatches()) { + tabletBatchBuilder.getAllNonEmptyAndShouldEmitBatchesAndDetach()) { transferInBatchWithoutCheck(endPointAndBatch); } } @@ -672,6 +787,7 @@ private void transferQueuedEventsIfNecessary(final boolean forced) { final long retryStartTime = System.currentTimeMillis(); final int remainingEvents = retryEventQueue.size() + retryTsFileQueue.size(); while (!retryEventQueue.isEmpty() || !retryTsFileQueue.isEmpty()) { + final Event retryEvent; synchronized (this) { if (isClosed.get()) { return; @@ -680,42 +796,41 @@ private void transferQueuedEventsIfNecessary(final boolean forced) { break; } - final Event peekedEvent; - final Event polledEvent; if (!retryEventQueue.isEmpty()) { - peekedEvent = retryEventQueue.peek(); - retryEvent2ResourceFailureType.remove(peekedEvent); - - if (peekedEvent instanceof PipeInsertNodeTabletInsertionEvent) { - retryTransfer((PipeInsertNodeTabletInsertionEvent) peekedEvent); - } else if (peekedEvent instanceof PipeRawTabletInsertionEvent) { - retryTransfer((PipeRawTabletInsertionEvent) peekedEvent); - } else { - LOGGER.warn( - DataNodePipeMessages - .IOTDBTHRIFTASYNCCONNECTOR_DOES_NOT_SUPPORT_TRANSFER_GENERIC_EVENT, - peekedEvent); - } - - polledEvent = retryEventQueue.poll(); + retryEvent = retryEventQueue.poll(); } else { if (transferTsFileCounter.get() >= PipeConfig.getInstance().getPipeRealTimeQueueMaxWaitingTsFileSize()) { return; } - peekedEvent = retryTsFileQueue.peek(); - retryEvent2ResourceFailureType.remove(peekedEvent); - retryTransfer((PipeTsFileInsertionEvent) peekedEvent); - polledEvent = retryTsFileQueue.poll(); + retryEvent = retryTsFileQueue.poll(); } - retryEventQueueEventCounter.decreaseEventCount(polledEvent); - if (polledEvent != peekedEvent) { - LOGGER.error( - DataNodePipeMessages.THE_EVENT_POLLED_FROM_THE_QUEUE_IS, peekedEvent, polledEvent); + if (retryEvent == null) { + break; } - if (polledEvent != null && LOGGER.isDebugEnabled()) { - LOGGER.debug(DataNodePipeMessages.POLLED_EVENT_FROM_RETRY_QUEUE, polledEvent); + // Remove membership before retrying. A failed retry can then re-enqueue the event without + // being mistaken for a duplicate. The queue's reference ownership stays with the event + // throughout the retry call, and retryingEvent2Handlers records the in-flight state. + retryEvent2ResourceFailureType.remove(retryEvent); + retryEventQueueEventCounter.decreaseEventCount(retryEvent); + retryingEvent2Handlers.put(retryEvent, Collections.newSetFromMap(new IdentityHashMap<>())); + } + + if (retryEvent instanceof PipeInsertNodeTabletInsertionEvent) { + retryTransfer((PipeInsertNodeTabletInsertionEvent) retryEvent); + } else if (retryEvent instanceof PipeRawTabletInsertionEvent) { + retryTransfer((PipeRawTabletInsertionEvent) retryEvent); + } else if (retryEvent instanceof PipeTsFileInsertionEvent) { + retryTransfer((PipeTsFileInsertionEvent) retryEvent); + } else { + LOGGER.warn( + DataNodePipeMessages.IOTDBTHRIFTASYNCCONNECTOR_DOES_NOT_SUPPORT_TRANSFER_GENERIC_EVENT, + retryEvent); + retryingEvent2Handlers.remove(retryEvent); + if (retryEvent instanceof EnrichedEvent) { + ((EnrichedEvent) retryEvent) + .clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName()); } } @@ -762,12 +877,24 @@ private void transferQueuedEventsIfNecessary(final boolean forced) { private void retryTransfer(final TabletInsertionEvent tabletInsertionEvent) { if (isTabletBatchModeEnabled) { + // A failed batch detachment may leave the event in the builder while the event is also + // present in the retry queue. Do not append it again (or rely on the last-element-only batch + // deduplication); relinquish the retry-queue ownership and let the existing batch transfer + // it once. + if (tabletBatchBuilder != null && tabletBatchBuilder.containsEvent(tabletInsertionEvent)) { + if (tabletInsertionEvent instanceof EnrichedEvent) { + ((EnrichedEvent) tabletInsertionEvent) + .decreaseReferenceCount(IoTDBDataRegionAsyncSink.class.getName(), false); + clearRetryingEventIfNoHandler((EnrichedEvent) tabletInsertionEvent); + } + return; + } try { - tabletBatchBuilder.onEvent(tabletInsertionEvent); - transferBatchedEventsIfNecessary(); + addTabletEventToBatchAndTransferIfNecessary(tabletInsertionEvent); if (tabletInsertionEvent instanceof EnrichedEvent) { ((EnrichedEvent) tabletInsertionEvent) .decreaseReferenceCount(IoTDBDataRegionAsyncSink.class.getName(), false); + clearRetryingEventIfNoHandler((EnrichedEvent) tabletInsertionEvent); } } catch (final Exception e) { addFailureEventToRetryQueue(tabletInsertionEvent, e); @@ -786,10 +913,6 @@ private void retryTransfer(final TabletInsertionEvent tabletInsertionEvent) { addFailureEventToRetryQueue(tabletInsertionEvent, null); } } catch (final Exception e) { - if (tabletInsertionEvent instanceof EnrichedEvent) { - ((EnrichedEvent) tabletInsertionEvent) - .decreaseReferenceCount(IoTDBDataRegionAsyncSink.class.getName(), false); - } addFailureEventToRetryQueue(tabletInsertionEvent, e); } } @@ -807,6 +930,16 @@ private void retryTransfer(final PipeTsFileInsertionEvent tsFileInsertionEvent) } } + private synchronized void clearRetryingEventIfNoHandler(final EnrichedEvent event) { + if (tabletBatchBuilder != null && tabletBatchBuilder.containsEvent(event)) { + return; + } + final Set handlers = retryingEvent2Handlers.get(event); + if (handlers == null || handlers.isEmpty()) { + retryingEvent2Handlers.remove(event); + } + } + /** * Add failure {@link Event} to retry queue. * @@ -814,11 +947,32 @@ private void retryTransfer(final PipeTsFileInsertionEvent tsFileInsertionEvent) */ @SuppressWarnings("java:S899") public void addFailureEventToRetryQueue(final Event event, final Exception e) { - addFailureEventToRetryQueue(event, e, null); + addFailureEventToRetryQueue(event, e, null, null); + } + + /** + * Reports a failure from a specific transfer handler. The handler identity is used to reject a + * late callback from an older retry attempt after the event has been polled again. + */ + public void addFailureEventToRetryQueue( + final Event event, final Exception e, final PipeTransferTrackableHandler sourceHandler) { + addFailureEventToRetryQueue(event, e, null, sourceHandler); } private synchronized void addFailureEventToRetryQueue( final Event event, final Exception e, final Set> failureRecordedPipes) { + addFailureEventToRetryQueue(event, e, failureRecordedPipes, null); + } + + private synchronized void addFailureEventToRetryQueue( + final Event event, + final Exception e, + final Set> failureRecordedPipes, + final PipeTransferTrackableHandler sourceHandler) { + if (event == null) { + return; + } + final PipeResourceFailureType resourceFailureType = PipeStopStrategy.getResourceFailureType(e, null); isConnectionException = @@ -826,15 +980,18 @@ private synchronized void addFailureEventToRetryQueue( if (event instanceof EnrichedEvent) { final EnrichedEvent enrichedEvent = (EnrichedEvent) event; if (enrichedEvent.isReleased()) { + retryingEvent2Handlers.remove(event); return; } if (isDroppedPipe(enrichedEvent)) { + retryingEvent2Handlers.remove(event); enrichedEvent.clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName()); return; } } if (isClosed.get()) { + retryingEvent2Handlers.remove(event); if (event instanceof EnrichedEvent) { ((EnrichedEvent) event).clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName()); } @@ -849,7 +1006,29 @@ private synchronized void addFailureEventToRetryQueue( lastRetryFailureMessage = getRetryFailureMessage(e); } - if (resourceFailureType != null && event instanceof EnrichedEvent) { + final Set retryingHandlers = retryingEvent2Handlers.get(event); + if (retryingHandlers != null) { + if (sourceHandler == null) { + // A synchronous failure before a retry handler was created. The queue still owns the + // original reference, so this attempt may be re-enqueued. + retryingEvent2Handlers.remove(event); + } else if (!retryingHandlers.remove(sourceHandler)) { + // A callback from a handler that belongs to an older attempt is stale. + return; + } else { + // The first current handler failure transfers its reference to the retry queue. Any later + // handler callback for the same event is deduplicated by retryEvent2ResourceFailureType. + retryingEvent2Handlers.remove(event); + } + } + + final boolean alreadyInRetryQueue = retryEvent2ResourceFailureType.containsKey(event); + final PipeResourceFailureType previousResourceFailureType = + retryEvent2ResourceFailureType.get(event); + + if (resourceFailureType != null + && event instanceof EnrichedEvent + && (!alreadyInRetryQueue || previousResourceFailureType != resourceFailureType)) { final EnrichedEvent enrichedEvent = (EnrichedEvent) event; final Pair pipeKey = new Pair<>(enrichedEvent.getPipeName(), enrichedEvent.getCreationTime()); @@ -860,9 +1039,17 @@ private synchronized void addFailureEventToRetryQueue( } } - if (resourceFailureType == null) { - retryEvent2ResourceFailureType.remove(event); - } else { + // A handler and the outer transfer wrapper can both report the same failure. Only the first + // report may transfer ownership to the retry queue; otherwise the queue counter and eventual + // reference release drift apart. + if (alreadyInRetryQueue) { + if (resourceFailureType != null) { + retryEvent2ResourceFailureType.put(event, resourceFailureType); + } + return; + } + + { retryEvent2ResourceFailureType.put(event, resourceFailureType); } @@ -877,12 +1064,6 @@ private synchronized void addFailureEventToRetryQueue( if (LOGGER.isDebugEnabled()) { LOGGER.debug(DataNodePipeMessages.ADDED_EVENT_TO_RETRY_QUEUE, event); } - - if (isClosed.get()) { - if (event instanceof EnrichedEvent) { - ((EnrichedEvent) event).clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName()); - } - } } /** @@ -936,6 +1117,15 @@ synchronized String getLastRetryFailureMessage() { return lastRetryFailureMessage; } + public void addFailureEventsToRetryQueue( + final Iterable events, + final Exception e, + final PipeTransferTrackableHandler sourceHandler) { + final Set> failureRecordedPipes = new HashSet<>(); + events.forEach( + event -> addFailureEventToRetryQueue(event, e, failureRecordedPipes, sourceHandler)); + } + private synchronized PipeResourceFailureType getRetryQueueResourceFailureType() { for (final PipeResourceFailureType failureType : PipeResourceFailureType.values()) { if (retryEvent2ResourceFailureType.containsValue(failureType)) { @@ -1095,6 +1285,7 @@ && isDroppedPipe((EnrichedEvent) event, committerKey)) { ((EnrichedEvent) event).clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName()); retryEventQueueEventCounter.decreaseEventCount(event); retryEvent2ResourceFailureType.remove(event); + retryingEvent2Handlers.remove(event); return true; } return false; @@ -1107,20 +1298,33 @@ && isDroppedPipe((EnrichedEvent) event, committerKey)) { ((EnrichedEvent) event).clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName()); retryEventQueueEventCounter.decreaseEventCount(event); retryEvent2ResourceFailureType.remove(event); + retryingEvent2Handlers.remove(event); return true; } return false; }); - + retryingEvent2Handlers + .keySet() + .removeIf( + event -> + event instanceof EnrichedEvent + && isDroppedPipe((EnrichedEvent) event, committerKey)); if (retryEventQueue.isEmpty() && retryTsFileQueue.isEmpty()) { lastRetryFailureMessage = null; } } @Override - // synchronized to avoid close connector when transfer event - public synchronized void close() { - isClosed.set(true); + public void close() { + final Set handlersToClose; + synchronized (this) { + if (!isClosed.compareAndSet(false, true)) { + return; + } + // Do not invoke handler callbacks while holding the sink monitor. A callback may already + // hold its handler monitor and call back into this sink during elimination. + handlersToClose = ImmutableSet.copyOf(pendingHandlers.keySet()); + } syncSink.close(); @@ -1128,15 +1332,13 @@ public synchronized void close() { tabletBatchBuilder.close(); } - // ensure all on-the-fly handlers have been cleared - if (hasPendingHandlers()) { - ImmutableSet.copyOf(pendingHandlers.keySet()) - .forEach( - handler -> { - handler.clearEventsReferenceCount(); - eliminateHandler(handler, true); - }); - } + // Ensure all on-the-fly handlers have been cleared outside the sink monitor. This avoids the + // sink -> handler -> sink lock cycle between close and an asynchronous callback. + handlersToClose.forEach( + handler -> { + handler.clearEventsReferenceCount(); + eliminateHandler(handler, true); + }); try { if (clientManager != null) { @@ -1164,12 +1366,14 @@ public synchronized void clearRetryEventsReferenceCount() { retryTsFileQueue.isEmpty() ? retryEventQueue.poll() : retryTsFileQueue.poll(); retryEventQueueEventCounter.decreaseEventCount(event); retryEvent2ResourceFailureType.remove(event); + retryingEvent2Handlers.remove(event); if (event instanceof EnrichedEvent) { ((EnrichedEvent) event).clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName()); } } retryEvent2ResourceFailureType.clear(); lastRetryFailureMessage = null; + retryingEvent2Handlers.clear(); } //////////////////////// APIs provided for metric framework //////////////////////// @@ -1193,7 +1397,42 @@ public boolean isClosed() { } public void trackHandler(final PipeTransferTrackableHandler handler) { - pendingHandlers.put(handler, handler); + boolean closeImmediately = false; + synchronized (this) { + if (isClosed.get()) { + closeImmediately = true; + } else { + pendingHandlers.put(handler, handler); + } + } + + if (closeImmediately) { + // The close snapshot has already been taken. Complete this late registration outside the + // sink monitor so a handler callback cannot deadlock against close(). + handler.clearEventsReferenceCount(); + eliminateHandler(handler, true); + } + } + + /** Registers a newly created handler with any retry attempt currently owning the events. */ + public synchronized void trackRetryHandler( + final PipeTransferTrackableHandler handler, final Iterable events) { + for (final Event event : events) { + final Set handlers = retryingEvent2Handlers.get(event); + if (handlers != null) { + handlers.add(handler); + } + } + } + + private synchronized void untrackRetryHandler(final PipeTransferTrackableHandler handler) { + retryingEvent2Handlers + .entrySet() + .removeIf( + entry -> { + entry.getValue().remove(handler); + return entry.getValue().isEmpty(); + }); } public void eliminateHandler( @@ -1203,6 +1442,7 @@ public void eliminateHandler( } handler.close(); pendingHandlers.remove(handler); + untrackRetryHandler(handler); } public boolean hasPendingHandlers() { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletBatchEventHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletBatchEventHandler.java index d0907c45abe89..6297bafc15e31 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletBatchEventHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletBatchEventHandler.java @@ -122,7 +122,7 @@ protected void onErrorInternal(final Exception exception) { events.size(), events.stream().map(EnrichedEvent::getPipeName).collect(Collectors.toSet())); } finally { - sink.addFailureEventsToRetryQueue(events, exception); + sink.addFailureEventsToRetryQueue(events, exception, this); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletInsertionEventHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletInsertionEventHandler.java index 94ca8ee5fe6c9..445e014ceec94 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletInsertionEventHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletInsertionEventHandler.java @@ -98,7 +98,7 @@ protected void onErrorInternal(final Exception exception) { event.getCommitterKey(), event.getCommitId()); } finally { - sink.addFailureEventToRetryQueue(event, exception); + sink.addFailureEventToRetryQueue(event, exception, this); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandler.java index e52c2479c0426..bdfdfe0ad968f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandler.java @@ -38,6 +38,7 @@ import org.slf4j.LoggerFactory; import java.util.Objects; +import java.util.concurrent.atomic.AtomicBoolean; public abstract class PipeTransferTrackableHandler implements AsyncMethodCallback, AutoCloseable { @@ -45,24 +46,45 @@ public abstract class PipeTransferTrackableHandler protected final IoTDBDataRegionAsyncSink sink; protected volatile AsyncPipeDataTransferServiceClient client; + // A client may report a failure through more than one path (callback, synchronous throw, or + // connection close). Only one path may release references, enqueue a retry, and eliminate this + // handler. + private final AtomicBoolean terminal = new AtomicBoolean(false); public PipeTransferTrackableHandler(final IoTDBDataRegionAsyncSink sink) { this.sink = sink; } @Override - public void onComplete(final TPipeTransferResp response) { + public synchronized void onComplete(final TPipeTransferResp response) { + if (terminal.get()) { + return; + } + if (Objects.nonNull(client) && Objects.nonNull(response)) { sink.recordReceiverStatus(client.getEndPoint(), response.getStatus()); } if (sink.isClosed()) { + if (!terminal.compareAndSet(false, true)) { + return; + } clearEventsReferenceCount(); sink.eliminateHandler(this, true); return; } - if (onCompleteInternal(response)) { + final boolean completed; + try { + completed = onCompleteInternal(response); + } catch (final Exception e) { + onError(e); + return; + } + if (completed) { + if (!terminal.compareAndSet(false, true)) { + return; + } // eliminate handler only when all transmissions corresponding to the handler have been // completed // NOTE: We should not clear the reference count of events, as this would cause the @@ -72,10 +94,22 @@ public void onComplete(final TPipeTransferResp response) { } @Override - public void onError(final Exception exception) { + public synchronized void onError(final Exception exception) { + if (!terminal.compareAndSet(false, true)) { + return; + } + if (client != null) { - ThriftClient.resolveException(exception, client); - client.setPrintLogWhenEncounterException(false); + try { + ThriftClient.resolveException(exception, client); + } catch (final Exception resolveException) { + exception.addSuppressed(resolveException); + LOGGER.warn( + DataNodePipeMessages.LOG_FAILED_TO_RESOLVE_TRANSFER_EXCEPTION_A4F5397A, + resolveException); + } finally { + client.setPrintLogWhenEncounterException(false); + } } if (sink.isClosed()) { @@ -84,8 +118,11 @@ public void onError(final Exception exception) { return; } - onErrorInternal(exception); - sink.eliminateHandler(this, false); + try { + onErrorInternal(exception); + } finally { + sink.eliminateHandler(this, false); + } } /** @@ -97,9 +134,12 @@ public void onError(final Exception exception) { * is closed or the receiver probe is delayed * @throws TException if an error occurs during the transfer */ - protected boolean tryTransfer( + protected synchronized boolean tryTransfer( final AsyncPipeDataTransferServiceClient client, final TPipeTransferReq req) throws TException { + if (terminal.get()) { + return false; + } if (Objects.isNull(this.client)) { this.client = client; } @@ -122,25 +162,31 @@ protected boolean tryTransfer( return true; } - private boolean returnFalseIfSinkIsClosed(final AsyncPipeDataTransferServiceClient client) { + private synchronized boolean returnFalseIfSinkIsClosed( + final AsyncPipeDataTransferServiceClient client) { if (!sink.isClosed()) { return false; } + if (!terminal.compareAndSet(false, true)) { + return true; + } clearEventsReferenceCount(); sink.eliminateHandler(this, true); - client.setShouldReturnSelf(true); - client.returnSelf( - (e) -> { - if (e instanceof IllegalStateException) { - PipeLogger.log( - ignored -> - LOGGER.info(DataNodePipeMessages.ILLEGAL_STATE_WHEN_RETURN_THE_CLIENT_TO), - DataNodePipeMessages.ILLEGAL_STATE_WHEN_RETURN_THE_CLIENT_TO); - return true; - } - return false; - }); + if (client != null) { + client.setShouldReturnSelf(true); + client.returnSelf( + (e) -> { + if (e instanceof IllegalStateException) { + PipeLogger.log( + ignored -> + LOGGER.info(DataNodePipeMessages.ILLEGAL_STATE_WHEN_RETURN_THE_CLIENT_TO), + DataNodePipeMessages.ILLEGAL_STATE_WHEN_RETURN_THE_CLIENT_TO); + return true; + } + return false; + }); + } this.client = null; return true; } @@ -178,7 +224,7 @@ protected final void transferWithOptionalRequestSlicing( throws TException { final int bodySizeLimit = PipeTransferSliceReqBuilder.getBodySizeLimit(); if (!PipeTransferSliceReqBuilder.shouldSlice(req, bodySizeLimit)) { - client.pipeTransfer(req, this); + transferWithExactlyOnceCallback(client, req, this); return; } @@ -218,7 +264,8 @@ private void transferSlicedRequest( final int bodySizeLimit) throws Exception { client.setShouldReturnSelf(shouldReturnSelf && sliceIndex == sliceCount - 1); - client.pipeTransfer( + transferWithExactlyOnceCallback( + client, PipeTransferSliceReqBuilder.buildSliceReq( originalReq, sliceOrderId, sliceIndex, sliceCount, bodySizeLimit), new AsyncMethodCallback() { @@ -303,7 +350,7 @@ private void fallbackToWholeRequest( if (returnFalseIfSinkIsClosed(client)) { return; } - client.pipeTransfer(originalReq, this); + transferWithExactlyOnceCallback(client, originalReq, this); } catch (final PipeRuntimeSinkNonReportTimeConfigurableException e) { returnClientToPool(client); PipeTransferTrackableHandler.this.onError(e); @@ -328,8 +375,34 @@ public void closeClient() { } } + private void transferWithExactlyOnceCallback( + final AsyncPipeDataTransferServiceClient client, + final TPipeTransferReq req, + final AsyncMethodCallback callback) + throws TException { + client.pipeTransfer( + req, + new AsyncMethodCallback() { + private final AtomicBoolean callbackHandled = new AtomicBoolean(false); + + @Override + public void onComplete(final TPipeTransferResp response) { + if (callbackHandled.compareAndSet(false, true)) { + callback.onComplete(response); + } + } + + @Override + public void onError(final Exception exception) { + if (callbackHandled.compareAndSet(false, true)) { + callback.onError(exception); + } + } + }); + } + @Override public void close() { - // Do nothing + terminal.set(true); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java index b79e7d6cb2bcf..408aaeefec9ae 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java @@ -200,7 +200,16 @@ public void transfer( if (readBuffer == null) { memoryBlock = PipeDataNodeResourceManager.memory().forceAllocateForTsFileWithRetry(readFileBufferSize); - readBuffer = new byte[readFileBufferSize]; + try { + readBuffer = new byte[readFileBufferSize]; + } catch (final RuntimeException | Error e) { + try { + releaseReadBufferMemoryBlock(); + } catch (final RuntimeException releaseException) { + e.addSuppressed(releaseException); + } + throw e; + } } if (reader == null) { @@ -216,11 +225,8 @@ public void transfer( if (currentFile == modFile) { currentFile = tsFile; position = 0; - try { - reader.close(); - } catch (final IOException e) { - LOGGER.warn(DataNodePipeMessages.FAILED_TO_CLOSE_FILE_READER_WHEN_SUCCESSFULLY, e); - } + reader.close(); + reader = null; reader = new RandomAccessFile(tsFile, "r"); transfer(clientManager, client); } else if (currentFile == tsFile) { @@ -305,13 +311,12 @@ protected void mayLimitRateAndRecordIO(final long requiredBytes) { } @Override - public void onComplete(final TPipeTransferResp response) { + public synchronized void onComplete(final TPipeTransferResp response) { try { super.onComplete(response); } finally { if (sink.isClosed()) { - releaseReadBufferMemoryBlock(); - returnClientIfNecessary(); + releaseReadBufferAndReturnClient(); } } } @@ -337,45 +342,33 @@ protected boolean onCompleteInternal(final TPipeTransferResp response) { } try { - if (reader != null) { - reader.close(); - } - - // Delete current file when using tsFile as batch - if (events.stream().anyMatch(event -> !(event instanceof PipeTsFileInsertionEvent))) { - RetryUtils.retryOnException( - () -> { - FileUtils.delete(currentFile); - return null; - }); - } - } catch (final IOException e) { - LOGGER.warn(DataNodePipeMessages.FAILED_TO_CLOSE_FILE_READER_OR_DELETE_1, e); + closeReaderAndDeleteBatchFile(true); } finally { - final int referenceCount = eventsReferenceCount.decrementAndGet(); - if (referenceCount <= 0) { - events.forEach( - event -> - event.decreaseReferenceCount(PipeTransferTsFileHandler.class.getName(), true)); - } + try { + final int referenceCount = eventsReferenceCount.decrementAndGet(); + if (referenceCount <= 0) { + events.forEach( + event -> + event.decreaseReferenceCount(PipeTransferTsFileHandler.class.getName(), true)); + } - if (events.size() <= 1 || LOGGER.isDebugEnabled()) { - LOGGER.info( - DataNodePipeMessages.SUCCESSFULLY_TRANSFERRED_FILE_COMMITTER_KEY_COMMIT_ID, - tsFile, - events.stream().map(EnrichedEvent::getCommitterKey).collect(Collectors.toList()), - events.stream().map(EnrichedEvent::getCommitIds).collect(Collectors.toList()), - referenceCount); - } else { - LOGGER.info( - DataNodePipeMessages - .SUCCESSFULLY_TRANSFERRED_FILE_BATCHED_TABLEINSERTIONEVENTS_REFERENCE_COUNT, - tsFile, - referenceCount); + if (events.size() <= 1 || LOGGER.isDebugEnabled()) { + LOGGER.info( + DataNodePipeMessages.SUCCESSFULLY_TRANSFERRED_FILE_COMMITTER_KEY_COMMIT_ID, + tsFile, + events.stream().map(EnrichedEvent::getCommitterKey).collect(Collectors.toList()), + events.stream().map(EnrichedEvent::getCommitIds).collect(Collectors.toList()), + referenceCount); + } else { + LOGGER.info( + DataNodePipeMessages + .SUCCESSFULLY_TRANSFERRED_FILE_BATCHED_TABLEINSERTIONEVENTS_REFERENCE_COUNT, + tsFile, + referenceCount); + } + } finally { + releaseReadBufferAndReturnClient(); } - - releaseReadBufferMemoryBlock(); - returnClientIfNecessary(); } return true; @@ -413,12 +406,11 @@ protected boolean onCompleteInternal(final TPipeTransferResp response) { } @Override - public void onError(final Exception exception) { + public synchronized void onError(final Exception exception) { try { super.onError(exception); } finally { - releaseReadBufferMemoryBlock(); - returnClientIfNecessary(); + releaseReadBufferAndReturnClient(); } } @@ -453,27 +445,13 @@ protected void onErrorInternal(final Exception exception) { } try { - if (reader != null) { - reader.close(); - } - - // Delete current file when using tsFile as batch - if (events.stream().anyMatch(event -> !(event instanceof PipeTsFileInsertionEvent))) { - RetryUtils.retryOnException( - () -> { - FileUtils.delete(currentFile); - return null; - }); - } - } catch (final IOException e) { - LOGGER.warn(DataNodePipeMessages.FAILED_TO_CLOSE_FILE_READER_OR_DELETE, e); + closeReaderAndDeleteBatchFile(false); } finally { try { - releaseReadBufferMemoryBlock(); - returnClientIfNecessary(); + releaseReadBufferAndReturnClient(); } finally { if (eventsHadBeenAddedToRetryQueue.compareAndSet(false, true)) { - sink.addFailureEventsToRetryQueue(events, exception); + sink.addFailureEventsToRetryQueue(events, exception, this); } } } @@ -529,34 +507,70 @@ public void clearEventsReferenceCount() { } @Override - public void close() { + public synchronized void close() { try { - if (reader != null) { - reader.close(); - reader = null; + closeReaderAndDeleteBatchFile(false); + } finally { + try { + super.close(); + } finally { + releaseReadBufferMemoryBlock(); } + } + } + + private void closeReaderAndDeleteBatchFile(final boolean transferSucceeded) { + final String errorMessage = + transferSucceeded + ? DataNodePipeMessages.FAILED_TO_CLOSE_FILE_READER_OR_DELETE_1 + : DataNodePipeMessages.FAILED_TO_CLOSE_FILE_READER_OR_DELETE; - if (currentFile.exists() - && events.stream().anyMatch(event -> !(event instanceof PipeTsFileInsertionEvent))) { + final RandomAccessFile readerToClose = reader; + if (readerToClose != null) { + try { + RetryUtils.retryOnException( + () -> { + readerToClose.close(); + return null; + }); + if (reader == readerToClose) { + reader = null; + } + } catch (final IOException e) { + LOGGER.warn(errorMessage, e); + } + } + + // Reader cleanup and generated-file cleanup are intentionally independent. A failed close + // must not leave a tablet-batch TsFile behind. + if (currentFile.exists() + && events.stream().anyMatch(event -> !(event instanceof PipeTsFileInsertionEvent))) { + try { RetryUtils.retryOnException( () -> { FileUtils.delete(currentFile); return null; }); + } catch (final IOException e) { + LOGGER.warn(errorMessage, e); } - } catch (final IOException e) { - LOGGER.warn(DataNodePipeMessages.FAILED_TO_CLOSE_FILE_READER_OR_DELETE, e); - } finally { - super.close(); + } + } + + private void releaseReadBufferAndReturnClient() { + try { releaseReadBufferMemoryBlock(); + } finally { + returnClientIfNecessary(); } } private void releaseReadBufferMemoryBlock() { - if (memoryBlock != null) { - memoryBlock.close(); - memoryBlock = null; - readBuffer = null; + final PipeTsFileMemoryBlock block = memoryBlock; + memoryBlock = null; + readBuffer = null; + if (block != null) { + block.close(); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java index 16c044b491d4a..49d9df02ab99d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java @@ -21,6 +21,7 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException; import org.apache.iotdb.commons.pipe.agent.task.progress.CommitterKey; import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.commons.pipe.event.EnrichedEvent; @@ -147,7 +148,17 @@ public void transfer(final TabletInsertionEvent tabletInsertionEvent) throws Exc try { if (isTabletBatchModeEnabled) { - tabletBatchBuilder.onEvent(tabletInsertionEvent); + try { + tabletBatchBuilder.onEvent(tabletInsertionEvent); + } catch (final PipeRuntimeOutOfMemoryCriticalException memoryException) { + try { + doTransferWrapper(); + } catch (final Exception transferException) { + transferException.addSuppressed(memoryException); + throw transferException; + } + tabletBatchBuilder.onEvent(tabletInsertionEvent); + } doTransferWrapper(); } else { if (tabletInsertionEvent instanceof PipeInsertNodeTabletInsertionEvent) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilder.java index 09499155a25ab..28c2a63f66cba 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilder.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilder.java @@ -99,6 +99,33 @@ public boolean isEmpty() { return dataBase2TabletList.isEmpty(); } + @Override + public Object createCheckpoint() { + final Map tabletListSizes = new HashMap<>(); + dataBase2TabletList.forEach( + (database, tablets) -> tabletListSizes.put(database, tablets.size())); + return new BatchState(tabletListSizes); + } + + @Override + public void rollbackToCheckpoint(final Object checkpoint) { + if (!(checkpoint instanceof BatchState)) { + return; + } + final BatchState batchState = (BatchState) checkpoint; + dataBase2TabletList + .entrySet() + .removeIf( + entry -> { + final Integer size = batchState.tabletListSizes.get(entry.getKey()); + if (size == null) { + return true; + } + truncate(entry.getValue(), size); + return false; + }); + } + @Override public synchronized void onSuccess() { super.onSuccess(); @@ -111,6 +138,20 @@ public synchronized void close() { dataBase2TabletList.clear(); } + private static void truncate(final List list, final int size) { + if (list.size() > size) { + list.subList(size, list.size()).clear(); + } + } + + private static final class BatchState { + private final Map tabletListSizes; + + private BatchState(final Map tabletListSizes) { + this.tabletListSizes = tabletListSizes; + } + } + private >>> List> writeTableModelTabletsToTsFiles( final List tabletList, final String dataBase) throws IOException { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilderV2.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilderV2.java index 8163d79ad51bb..ca4efdadb9802 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilderV2.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilderV2.java @@ -116,6 +116,34 @@ public boolean isEmpty() { return dataBase2TabletList.isEmpty(); } + @Override + public Object createCheckpoint() { + final Map tabletListSizes = new HashMap<>(); + dataBase2TabletList.forEach( + (database, tablets) -> tabletListSizes.put(database, tablets.size())); + return new BatchState(tabletListSizes, fallbackBuilder.createCheckpoint()); + } + + @Override + public void rollbackToCheckpoint(final Object checkpoint) { + if (!(checkpoint instanceof BatchState)) { + return; + } + final BatchState batchState = (BatchState) checkpoint; + dataBase2TabletList + .entrySet() + .removeIf( + entry -> { + final Integer size = batchState.tabletListSizes.get(entry.getKey()); + if (size == null) { + return true; + } + truncate(entry.getValue(), size); + return false; + }); + fallbackBuilder.rollbackToCheckpoint(batchState.fallbackCheckpoint); + } + @Override public synchronized void onSuccess() { super.onSuccess(); @@ -130,6 +158,23 @@ public synchronized void close() { fallbackBuilder.close(); } + private static void truncate(final List list, final int size) { + if (list.size() > size) { + list.subList(size, list.size()).clear(); + } + } + + private static final class BatchState { + private final Map tabletListSizes; + private final Object fallbackCheckpoint; + + private BatchState( + final Map tabletListSizes, final Object fallbackCheckpoint) { + this.tabletListSizes = tabletListSizes; + this.fallbackCheckpoint = fallbackCheckpoint; + } + } + private List> writeTabletsToTsFiles(final String dataBase) throws WriteProcessException { final IMemTable memTable = new PrimitiveMemTable(null, null); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTreeModelTsFileBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTreeModelTsFileBuilder.java index b68dc4315f6b2..0ac0c884ec132 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTreeModelTsFileBuilder.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTreeModelTsFileBuilder.java @@ -86,6 +86,21 @@ public boolean isEmpty() { return tabletList.isEmpty(); } + @Override + public Object createCheckpoint() { + return new BatchState(tabletList.size(), isTabletAlignedList.size()); + } + + @Override + public void rollbackToCheckpoint(final Object checkpoint) { + if (!(checkpoint instanceof BatchState)) { + return; + } + final BatchState batchState = (BatchState) checkpoint; + truncate(tabletList, batchState.tabletListSize); + truncate(isTabletAlignedList, batchState.isTabletAlignedListSize); + } + @Override public void onSuccess() { super.onSuccess(); @@ -100,6 +115,22 @@ public synchronized void close() { isTabletAlignedList.clear(); } + private static void truncate(final List list, final int size) { + if (list.size() > size) { + list.subList(size, list.size()).clear(); + } + } + + private static final class BatchState { + private final int tabletListSize; + private final int isTabletAlignedListSize; + + private BatchState(final int tabletListSize, final int isTabletAlignedListSize) { + this.tabletListSize = tabletListSize; + this.isTabletAlignedListSize = isTabletAlignedListSize; + } + } + private List> writeTabletsToTsFiles() throws IOException, WriteProcessException { final Map> device2Tablets = new HashMap<>(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTreeModelTsFileBuilderV2.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTreeModelTsFileBuilderV2.java index a325f2cbdfb13..9d9d8b0fb0a83 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTreeModelTsFileBuilderV2.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTreeModelTsFileBuilderV2.java @@ -101,6 +101,23 @@ public boolean isEmpty() { return tabletList.isEmpty(); } + @Override + public Object createCheckpoint() { + return new BatchState( + tabletList.size(), isTabletAlignedList.size(), fallbackBuilder.createCheckpoint()); + } + + @Override + public void rollbackToCheckpoint(final Object checkpoint) { + if (!(checkpoint instanceof BatchState)) { + return; + } + final BatchState batchState = (BatchState) checkpoint; + truncate(tabletList, batchState.tabletListSize); + truncate(isTabletAlignedList, batchState.isTabletAlignedListSize); + fallbackBuilder.rollbackToCheckpoint(batchState.fallbackCheckpoint); + } + @Override public void onSuccess() { super.onSuccess(); @@ -117,6 +134,27 @@ public synchronized void close() { fallbackBuilder.close(); } + private static void truncate(final List list, final int size) { + if (list.size() > size) { + list.subList(size, list.size()).clear(); + } + } + + private static final class BatchState { + private final int tabletListSize; + private final int isTabletAlignedListSize; + private final Object fallbackCheckpoint; + + private BatchState( + final int tabletListSize, + final int isTabletAlignedListSize, + final Object fallbackCheckpoint) { + this.tabletListSize = tabletListSize; + this.isTabletAlignedListSize = isTabletAlignedListSize; + this.fallbackCheckpoint = fallbackCheckpoint; + } + } + private List> writeTabletsToTsFiles() throws WriteProcessException { final IMemTable memTable = new PrimitiveMemTable(null, null); final List> sealedFiles = new ArrayList<>(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTsFileBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTsFileBuilder.java index 414d2fad2315e..9af6872279287 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTsFileBuilder.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTsFileBuilder.java @@ -129,6 +129,16 @@ public abstract List> convertTabletToTsFileWithDBInfo() public abstract boolean isEmpty(); + /** Captures the in-memory tablet state before appending one event. */ + public Object createCheckpoint() { + return null; + } + + /** Restores the in-memory tablet state after a failed append. */ + public void rollbackToCheckpoint(final Object checkpoint) { + // Builders that keep append-only state override this method. + } + public synchronized void onSuccess() { fileWriter = null; } 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 42db57eef8dd2..41cec382b4fd2 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 @@ -96,6 +96,8 @@ import java.util.Map; import java.util.Objects; import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; @@ -465,6 +467,66 @@ public void testConsumeTabletInsertionEventsWithRetryKeepsParserForTransientOutO event.close(); } + @Test(timeout = 60000) + public void testCloseDefersPendingTabletReleaseUntilConsumerReturns() throws Exception { + final PipeTsFileInsertionEvent event = + createPipeTsFileInsertionEventForRetryTest("nonaligned-consume-close-race.tsfile"); + final CountDownLatch consumerEntered = new CountDownLatch(1); + final CountDownLatch allowConsumerReturn = new CountDownLatch(1); + final AtomicReference parsedEventReference = + new AtomicReference<>(); + final AtomicReference consumerFailure = new AtomicReference<>(); + + final Thread consumerThread = + new Thread( + () -> { + try { + event.consumeTabletInsertionEventsWithRetry( + parsedEvent -> { + parsedEventReference.set(parsedEvent); + consumerEntered.countDown(); + try { + allowConsumerReturn.await(); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + throw new RuntimeException("stop after close race"); + }, + "test"); + } catch (final Throwable t) { + consumerFailure.set(t); + } + }, + "pipe-tsfile-close-race-test"); + consumerThread.start(); + + try { + Assert.assertTrue(consumerEntered.await(10, TimeUnit.SECONDS)); + final PipeRawTabletInsertionEvent parsedEvent = parsedEventReference.get(); + Assert.assertNotNull(parsedEvent); + Assert.assertFalse(parsedEvent.isReleased()); + Assert.assertNotNull(getEventParser(event).get()); + + event.close(); + + // close() detaches the parser immediately, but the tablet remains valid for the callback + // that was already handed it. + Assert.assertFalse(parsedEvent.isReleased()); + Assert.assertNull(getEventParser(event).get()); + + allowConsumerReturn.countDown(); + consumerThread.join(10_000); + Assert.assertFalse(consumerThread.isAlive()); + Assert.assertTrue(parsedEvent.isReleased()); + Assert.assertNotNull(consumerFailure.get()); + } finally { + allowConsumerReturn.countDown(); + consumerThread.join(10_000); + event.close(); + } + } + private PipeTsFileInsertionEvent createPipeTsFileInsertionEventForRetryTest(final String fileName) throws Exception { nonalignedTsFile = diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerResizeTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerResizeTest.java index d63fe1c6e8309..ddaf043b969d9 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerResizeTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerResizeTest.java @@ -139,6 +139,60 @@ public void testTabletResizeLeavesMemoryForSinkForwardProgress() { Assert.assertEquals(0, manager.getUsedMemorySizeInBytes()); } + @Test + public void testTryResizeRejectsImmediatelyWithoutChangingAccounting() { + final PipeMemoryManager manager = + new PipeMemoryManager( + new AtomicLongMemoryBlock( + "PipeMemoryManagerResizeTest", + null, + TOTAL_MEMORY_SIZE_IN_BYTES, + MemoryBlockType.DYNAMIC)); + final PipeTabletMemoryBlock retainedTablet = + manager.forceAllocateForTabletWithRetry(TABLET_MEMORY_SIZE_IN_BYTES); + final PipeTabletMemoryBlock pendingTablet = manager.forceAllocateForTabletWithRetry(0); + + try { + Assert.assertFalse(manager.tryResize(pendingTablet, 1)); + Assert.assertEquals(0, pendingTablet.getMemoryUsageInBytes()); + Assert.assertEquals(TABLET_MEMORY_SIZE_IN_BYTES, manager.getUsedMemorySizeInBytes()); + Assert.assertEquals(TABLET_MEMORY_SIZE_IN_BYTES, manager.getUsedMemorySizeInBytesOfTablets()); + + manager.release(retainedTablet); + Assert.assertTrue(manager.tryResize(pendingTablet, 1)); + Assert.assertEquals(1, pendingTablet.getMemoryUsageInBytes()); + Assert.assertEquals(1, manager.getUsedMemorySizeInBytesOfTablets()); + } finally { + manager.release(retainedTablet); + manager.release(pendingTablet); + } + + Assert.assertEquals(0, manager.getUsedMemorySizeInBytes()); + } + + @Test + public void testTryResizeRejectsNegativeTargetWithoutChangingAccounting() { + final PipeMemoryManager manager = + new PipeMemoryManager( + new AtomicLongMemoryBlock( + "PipeMemoryManagerResizeTest", + null, + TOTAL_MEMORY_SIZE_IN_BYTES, + MemoryBlockType.DYNAMIC)); + final PipeTabletMemoryBlock tablet = manager.forceAllocateForTabletWithRetry(10); + + try { + Assert.assertFalse(manager.tryResize(tablet, -1)); + Assert.assertEquals(10, tablet.getMemoryUsageInBytes()); + Assert.assertEquals(10, manager.getUsedMemorySizeInBytes()); + Assert.assertEquals(10, manager.getUsedMemorySizeInBytesOfTablets()); + } finally { + manager.release(tablet); + } + + Assert.assertEquals(0, manager.getUsedMemorySizeInBytes()); + } + @Test public void testFloatingAndNonFloatingMemoryShareTheSamePool() { final AtomicLong floatingMemoryUsageInBytes = new AtomicLong(0); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java index 11c433ac1b5eb..b7732d85d2f86 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java @@ -342,8 +342,11 @@ public void testAsyncSinkDropDoesNotRequeueDroppedPipeEvents() throws Exception recreatedPipeEvent.setCommitterKeyAndCommitId(new CommitterKey("pipe", 2L, 1, -1), 1L); connector.addFailureEventToRetryQueue(recreatedPipeEvent, new PipeException("test")); + connector.addFailureEventToRetryQueue(recreatedPipeEvent, new PipeException("test-again")); Assert.assertEquals(1, connector.getRetryEventQueueSize()); + connector.clearRetryEventsReferenceCount(); + Assert.assertTrue(recreatedPipeEvent.isReleased()); } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatchTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatchTest.java index de0d85d5a4ced..3f70f55d1e704 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatchTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatchTest.java @@ -41,6 +41,7 @@ public void testFullyPrunedTableModelTabletIsReleasedAndNotRetained() throws Exc Assert.assertFalse(batch.onEvent(event)); Assert.assertTrue(batch.isEmpty()); + Assert.assertEquals(0, batch.totalBufferSize); Assert.assertTrue(event.isReleased()); Assert.assertEquals(0, event.getReferenceCount()); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilderTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilderTest.java new file mode 100644 index 0000000000000..b366c37ecb42b --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilderTest.java @@ -0,0 +1,110 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.pipe.sink.payload.evolvable.batch; + +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException; +import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent; +import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters; +import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent; + +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.utils.Pair; +import org.apache.tsfile.write.record.Tablet; +import org.junit.Assert; +import org.junit.Test; + +import java.util.Collections; +import java.util.List; + +import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_BATCH_DELAY_MS_KEY; + +public class PipeTransferBatchReqBuilderTest { + + @Test + public void testDetachedBatchCannotClearSubsequentEvent() throws Exception { + final PipeTransferBatchReqBuilder builder = + new PipeTransferBatchReqBuilder( + new PipeParameters(Collections.singletonMap(CONNECTOR_IOTDB_BATCH_DELAY_MS_KEY, "0"))); + final PipeRawTabletInsertionEvent firstEvent = createEvent(1); + final PipeRawTabletInsertionEvent secondEvent = createEvent(2); + + try { + builder.onEvent(firstEvent); + final List> detachedBatches = + builder.getAllNonEmptyAndShouldEmitBatchesAndDetach(); + Assert.assertEquals(1, detachedBatches.size()); + + builder.onEvent(secondEvent); + detachedBatches.get(0).getRight().closeAfterEventTransfer(); + + Assert.assertEquals(1, builder.size()); + Assert.assertEquals(1, secondEvent.getReferenceCount()); + Assert.assertFalse(secondEvent.isReleased()); + + // Simulate completion by the handler that owns the detached event reference. + firstEvent.decreaseReferenceCount(getClass().getName(), false); + Assert.assertTrue(firstEvent.isReleased()); + } finally { + builder.close(); + } + + Assert.assertTrue(secondEvent.isReleased()); + } + + @Test + public void testMemoryPressureKeepsExistingBatchEmittable() throws Exception { + final PipeTabletEventBatch batch = + new PipeTabletEventBatch(Integer.MAX_VALUE, Long.MAX_VALUE, null) { + private int constructCount; + + @Override + protected boolean constructBatch(final TabletInsertionEvent event) { + increaseTotalBufferSizeAndUpdateMemoryBlock( + constructCount++ == 0 ? 1 : Long.MAX_VALUE / 2); + return true; + } + }; + + try { + Assert.assertFalse(batch.onEvent(createEvent(1))); + Assert.assertThrows( + PipeRuntimeOutOfMemoryCriticalException.class, () -> batch.onEvent(createEvent(2))); + Assert.assertTrue(batch.shouldEmit()); + } finally { + batch.close(); + } + } + + private static PipeRawTabletInsertionEvent createEvent(final int value) { + final Tablet tablet = + new Tablet( + IDeviceID.Factory.DEFAULT_FACTORY.create("root.test.device"), + Collections.singletonList("s1"), + Collections.singletonList(TSDataType.INT32), + 1); + tablet.addTimestamp(0, value); + tablet.addValue("s1", 0, value); + tablet.setRowSize(1); + return new PipeRawTabletInsertionEvent( + false, "root.test", null, "root.test", tablet, false, null, 0, null, null, false); + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandlerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandlerTest.java index 28f4f910b0f9e..1b9e3af41e8ad 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandlerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandlerTest.java @@ -29,6 +29,7 @@ import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeRequestType; import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeTransferSliceReq; import org.apache.iotdb.db.pipe.sink.protocol.thrift.async.IoTDBDataRegionAsyncSink; +import org.apache.iotdb.pipe.api.exception.PipeException; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq; import org.apache.iotdb.service.rpc.thrift.TPipeTransferResp; @@ -47,6 +48,8 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; public class PipeTransferTrackableHandlerTest { @@ -220,6 +223,113 @@ public void testClientIsReturnedWhenReceiverProbeIsDelayed() throws Exception { .pipeTransfer(Mockito.any(TPipeTransferReq.class), Mockito.any()); } + @Test + public void testTerminalCallbacksAreIdempotent() { + final IoTDBDataRegionAsyncSink sink = Mockito.mock(IoTDBDataRegionAsyncSink.class); + final TestPipeTransferTrackableHandler completeHandler = + new TestPipeTransferTrackableHandler(sink); + + completeHandler.onComplete(successResp()); + completeHandler.onComplete(successResp()); + + Assert.assertEquals(1, completeHandler.completeCount); + Mockito.verify(sink, Mockito.times(1)).eliminateHandler(completeHandler, false); + + final TestPipeTransferTrackableHandler errorHandler = + new TestPipeTransferTrackableHandler(sink); + errorHandler.onError(new PipeException("first")); + errorHandler.onError(new PipeException("second")); + + Assert.assertEquals(1, errorHandler.errorCount); + Mockito.verify(sink, Mockito.times(1)).eliminateHandler(errorHandler, false); + } + + @Test + public void testWholeRequestCallbackIsHandledOnlyOnce() throws Exception { + commonConfig.setPipeSinkRequestSliceThresholdBytes(1024); + final IoTDBDataRegionAsyncSink sink = Mockito.mock(IoTDBDataRegionAsyncSink.class); + final AsyncPipeDataTransferServiceClient client = + Mockito.mock(AsyncPipeDataTransferServiceClient.class); + Mockito.doAnswer( + invocation -> { + final AsyncMethodCallback callback = invocation.getArgument(1); + callback.onComplete(successResp()); + callback.onComplete(successResp()); + return null; + }) + .when(client) + .pipeTransfer(Mockito.any(TPipeTransferReq.class), Mockito.any()); + + final TestPipeTransferTrackableHandler handler = + new TestPipeTransferTrackableHandler(sink, false); + handler.transfer(client, createReq(1)); + + Assert.assertEquals(1, handler.completeCount); + Mockito.verify(sink, Mockito.never()).eliminateHandler(handler, false); + } + + @Test + public void testSinkCloseDoesNotDeadlockWithHandlerCallback() throws Exception { + final CloseAwareAsyncSink sink = new CloseAwareAsyncSink(); + final CountDownLatch callbackEntered = new CountDownLatch(1); + final CountDownLatch allowCallbackToFinish = new CountDownLatch(1); + final PipeTransferTrackableHandler handler = + new PipeTransferTrackableHandler(sink) { + @Override + protected boolean onCompleteInternal(final TPipeTransferResp response) { + callbackEntered.countDown(); + try { + allowCallbackToFinish.await(); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + return false; + } + return true; + } + + @Override + protected void onErrorInternal(final Exception exception) { + // No-op. + } + + @Override + protected void doTransfer( + final AsyncPipeDataTransferServiceClient client, final TPipeTransferReq req) { + // No-op. + } + + @Override + public void clearEventsReferenceCount() { + // No-op. + } + }; + sink.trackHandler(handler); + + final Thread callbackThread = + new Thread(() -> handler.onComplete(successResp()), "pipe-handler-callback"); + callbackThread.setDaemon(true); + final Thread closeThread = new Thread(sink::close, "pipe-sink-close"); + closeThread.setDaemon(true); + + callbackThread.start(); + Assert.assertTrue(callbackEntered.await(5, TimeUnit.SECONDS)); + closeThread.start(); + Assert.assertTrue(sink.closeEliminationStarted.await(5, TimeUnit.SECONDS)); + + try { + allowCallbackToFinish.countDown(); + callbackThread.join(TimeUnit.SECONDS.toMillis(5)); + closeThread.join(TimeUnit.SECONDS.toMillis(5)); + Assert.assertFalse(callbackThread.isAlive()); + Assert.assertFalse(closeThread.isAlive()); + Assert.assertTrue(sink.isClosed()); + } finally { + allowCallbackToFinish.countDown(); + callbackThread.interrupt(); + closeThread.interrupt(); + } + } + @Test public void testReceiverRetriesAreSerializedForAnyFailureStatus() { commonConfig.setPipeSinkSubtaskSleepIntervalInitMs(40); @@ -297,9 +407,16 @@ private static class TestPipeTransferTrackableHandler extends PipeTransferTracka private int completeCount; private int errorCount; + private final boolean completeOnResponse; private TestPipeTransferTrackableHandler(final IoTDBDataRegionAsyncSink sink) { + this(sink, true); + } + + private TestPipeTransferTrackableHandler( + final IoTDBDataRegionAsyncSink sink, final boolean completeOnResponse) { super(sink); + this.completeOnResponse = completeOnResponse; } private void transfer( @@ -311,7 +428,7 @@ private void transfer( @Override protected boolean onCompleteInternal(final TPipeTransferResp response) { completeCount++; - return true; + return completeOnResponse; } @Override @@ -331,4 +448,24 @@ public void clearEventsReferenceCount() { // Do nothing } } + + private static class CloseAwareAsyncSink extends IoTDBDataRegionAsyncSink { + private final CountDownLatch closeEliminationStarted = new CountDownLatch(1); + private volatile Thread closeThread; + + @Override + public void close() { + closeThread = Thread.currentThread(); + super.close(); + } + + @Override + public void eliminateHandler( + final PipeTransferTrackableHandler handler, final boolean closeClient) { + if (Thread.currentThread() == closeThread) { + closeEliminationStarted.countDown(); + } + super.eliminateHandler(handler, closeClient); + } + } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandlerCleanupTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandlerCleanupTest.java index 892295b34b347..3cca3285cca15 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandlerCleanupTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandlerCleanupTest.java @@ -33,6 +33,8 @@ import org.mockito.Mockito; import java.io.File; +import java.io.IOException; +import java.io.RandomAccessFile; import java.lang.reflect.Field; import java.nio.file.Files; import java.util.Collections; @@ -51,6 +53,23 @@ public void testCloseDeletesBatchFile() throws Exception { Assert.assertFalse(file.exists()); } + @Test + public void testCloseDeletesBatchFileWhenReaderCloseFails() throws Exception { + final File file = Files.createTempFile("pipe-transfer-close-failure", ".tsfile").toFile(); + final EnrichedEvent event = Mockito.mock(EnrichedEvent.class); + final PipeTransferTsFileHandler handler = createHandler(file, event); + final RandomAccessFile reader = Mockito.mock(RandomAccessFile.class); + Mockito.doThrow(new IOException("close failed")).when(reader).close(); + final Field readerField = PipeTransferTsFileHandler.class.getDeclaredField("reader"); + readerField.setAccessible(true); + readerField.set(handler, reader); + + handler.close(); + + Mockito.verify(reader, Mockito.atLeastOnce()).close(); + Assert.assertFalse(file.exists()); + } + @Test public void testNullClientDeletesBatchFile() throws Exception { final File file = Files.createTempFile("pipe-transfer-null-client", ".tsfile").toFile(); @@ -111,7 +130,9 @@ public void testSealFailurePassesNestedReceiverMessageToRetryQueue() throws Exce final ArgumentCaptor exceptionCaptor = ArgumentCaptor.forClass(Exception.class); Mockito.verify(sink) .addFailureEventsToRetryQueue( - Mockito.eq(Collections.singletonList(event)), exceptionCaptor.capture()); + Mockito.eq(Collections.singletonList(event)), + exceptionCaptor.capture(), + Mockito.eq(handler)); Assert.assertEquals("receiver disk is full", exceptionCaptor.getValue().getMessage()); } finally { if (file.exists()) {