From e96bc0c88f06974a60ddde4362ac37764ea2d71c Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 17 Aug 2026 11:20:35 +0800 Subject: [PATCH 1/2] [Pipe] Support pooled TsFile parsing --- .../task/connection/PipeEventCollector.java | 151 +++++-- .../task/stage/PipeTaskProcessorStage.java | 11 +- .../processor/PipeProcessorSubtask.java | 416 +++++++++++++++++- .../tablet/PipeRawTabletInsertionEvent.java | 101 ++++- .../tsfile/PipeTsFileInsertionEvent.java | 33 +- .../resource/memory/PipeMemoryManager.java | 45 +- .../evolvable/batch/PipeTabletEventBatch.java | 7 +- .../apache/iotdb/db/conf/PropertiesTest.java | 4 + .../PipeProcessorSubtaskExecutorTest.java | 390 ++++++++++++++++ .../PipeRawTabletInsertionEventTest.java | 113 +++++ .../memory/PipeMemoryManagerTest.java | 11 + .../conf/iotdb-system.properties.template | 6 +- .../iotdb/commons/concurrent/ThreadName.java | 2 + .../iotdb/commons/conf/CommonConfig.java | 14 +- .../donothing/DoNothingProcessor.java | 18 +- .../task/connection/BlockingPendingQueue.java | 174 +++++++- .../UnboundedBlockingPendingQueue.java | 4 +- .../constant/PipeProcessorConstant.java | 4 + .../connection/BlockingPendingQueueTest.java | 119 +++++ 19 files changed, 1545 insertions(+), 78 deletions(-) create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEventTest.java create mode 100644 iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/agent/task/connection/BlockingPendingQueueTest.java diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java index df72ccb830dbc..50a0b07e93ed4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java @@ -19,8 +19,10 @@ package org.apache.iotdb.db.pipe.agent.task.connection; +import org.apache.iotdb.commons.pipe.agent.task.connection.BlockingPendingQueue.PendingEventMemoryReservation; import org.apache.iotdb.commons.pipe.agent.task.connection.UnboundedBlockingPendingQueue; import org.apache.iotdb.commons.pipe.agent.task.progress.PipeEventCommitManager; +import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePatternOperations; import org.apache.iotdb.commons.pipe.event.EnrichedEvent; import org.apache.iotdb.commons.pipe.event.ProgressReportEvent; @@ -56,6 +58,8 @@ public class PipeEventCollector implements EventCollector { private final boolean skipParsing; + private final boolean isTsFileParserCollector; + private final AtomicInteger collectInvocationCount = new AtomicInteger(0); private boolean hasNoGeneratedEvent = true; private boolean isFailedToIncreaseReferenceCount = false; @@ -66,11 +70,22 @@ public PipeEventCollector( final int regionId, final boolean forceTabletFormat, final boolean skipParsing) { + this(pendingQueue, creationTime, regionId, forceTabletFormat, skipParsing, false); + } + + private PipeEventCollector( + final UnboundedBlockingPendingQueue pendingQueue, + final long creationTime, + final int regionId, + final boolean forceTabletFormat, + final boolean skipParsing, + final boolean isTsFileParserCollector) { this.pendingQueue = pendingQueue; this.creationTime = creationTime; this.regionId = regionId; this.forceTabletFormat = forceTabletFormat; this.skipParsing = skipParsing; + this.isTsFileParserCollector = isTsFileParserCollector; } @Override @@ -118,7 +133,7 @@ private void parseAndCollectEvent(final PipeRawTabletInsertionEvent sourceEvent) if (sourceEvent.shouldParseTimeOrPattern()) { collectParsedRawTableEvent(sourceEvent.parseEventWithPatternOrTime()); } else { - collectEvent(sourceEvent); + collectEvent(sourceEvent, isTsFileParserCollector); } } @@ -152,10 +167,32 @@ public static boolean canSkipParsing4TsFileEvent(final PipeTsFileInsertionEvent return !sourceEvent.shouldParseTimeOrPattern(); } + public boolean shouldParseTsFileEvent(final PipeTsFileInsertionEvent sourceEvent) { + return !skipParsing && (forceTabletFormat || !canSkipParsing4TsFileEvent(sourceEvent)); + } + + public void prepareTsFileEventForParallelParsing(final PipeTsFileInsertionEvent sourceEvent) { + if (sourceEvent.isProgressReportManagedByTsFileParser()) { + return; + } + if (sourceEvent.getCommitId() <= EnrichedEvent.NO_COMMIT_ID) { + PipeEventCommitManager.getInstance() + .enrichWithCommitterKeyAndCommitId(sourceEvent, creationTime, regionId); + } + if (sourceEvent.getCommitId() > EnrichedEvent.NO_COMMIT_ID) { + sourceEvent.markProgressReportManagedByTsFileParser(); + } + } + + public PipeEventCollector forkForTsFileParser() { + return new PipeEventCollector( + pendingQueue, creationTime, regionId, forceTabletFormat, skipParsing, true); + } + private void collectParsedRawTableEvent(final PipeRawTabletInsertionEvent parsedEvent) { if (!parsedEvent.hasNoNeedParsingAndIsEmpty()) { hasNoGeneratedEvent = false; - collectEvent(parsedEvent); + collectEvent(parsedEvent, isTsFileParserCollector); } } @@ -183,37 +220,95 @@ private void parseAndCollectEvent(final PipeSchemaRegionWritePlanEvent deleteDat } private void collectEvent(final Event event) { - if (event instanceof EnrichedEvent) { - final EnrichedEvent enrichedEvent = (EnrichedEvent) event; - if (!enrichedEvent.increaseReferenceCount(PipeEventCollector.class.getName())) { - LOGGER.warn("PipeEventCollector: The event {} is already released, skipping it.", event); - isFailedToIncreaseReferenceCount = true; - return; - } + collectEvent(event, false); + } - // Assign a commit id for this event in order to report progress in order. - PipeEventCommitManager.getInstance() - .enrichWithCommitterKeyAndCommitId(enrichedEvent, creationTime, regionId); - - // Assign a rebootTime for iotConsensusV2 - enrichedEvent.setRebootTimes(PipeDataNodeAgent.runtime().getRebootTimes()); - - if (enrichedEvent.getPipeName() != null - && (pendingQueue.isEventFromDroppedPipe(enrichedEvent) - || (enrichedEvent.getCommitterKey() == null - && pendingQueue.isPipeDropped( - enrichedEvent.getPipeName(), creationTime, regionId)))) { - enrichedEvent.clearReferenceCount(PipeEventCollector.class.getName()); - return; + private void collectEvent(final Event event, final boolean useParserQueueMemoryBackpressure) { + PendingEventMemoryReservation memoryReservation = null; + long tabletSizeInBytes = 0; + if (useParserQueueMemoryBackpressure && event instanceof PipeRawTabletInsertionEvent) { + tabletSizeInBytes = ((PipeRawTabletInsertionEvent) event).getTabletSizeInBytes(); + memoryReservation = + pendingQueue.waitForMemoryReservation( + tabletSizeInBytes, Math.max(1, PipeConfig.getInstance().getTsFileParserMemory())); + if (memoryReservation == null) { + throw new PipeException("Interrupted while waiting for parser output queue memory."); } } - if (event instanceof PipeHeartbeatEvent) { - ((PipeHeartbeatEvent) event).recordConnectorQueueSize(pendingQueue); - } + boolean isReferenceIncreased = false; + boolean isOffered = false; + try { + if (event instanceof EnrichedEvent) { + final EnrichedEvent enrichedEvent = (EnrichedEvent) event; + final boolean increased = + useParserQueueMemoryBackpressure && event instanceof PipeRawTabletInsertionEvent + ? ((PipeRawTabletInsertionEvent) event) + .increaseReferenceCountWithReservedMemory( + PipeEventCollector.class.getName(), + Math.max( + tabletSizeInBytes > Long.MAX_VALUE / 2 + ? Long.MAX_VALUE + : tabletSizeInBytes * 2, + PipeConfig.getInstance().getPipeDataStructureTabletSizeInBytes())) + : enrichedEvent.increaseReferenceCount(PipeEventCollector.class.getName()); + if (!increased) { + LOGGER.warn("PipeEventCollector: The event {} is already released, skipping it.", event); + isFailedToIncreaseReferenceCount = true; + return; + } + isReferenceIncreased = true; + + final PipeTsFileInsertionEvent progressReportSourceTsFile = + event instanceof PipeRawTabletInsertionEvent + ? ((PipeRawTabletInsertionEvent) event).getProgressReportSourceTsFile() + : null; + if (progressReportSourceTsFile == null) { + // Assign a commit id for this event in order to report progress in order. + PipeEventCommitManager.getInstance() + .enrichWithCommitterKeyAndCommitId(enrichedEvent, creationTime, regionId); + } else { + // The source TsFile owns the ordered commit id. Raw tablets retain its committer key but + // do not create independent commits. + enrichedEvent.setCommitterKeyAndCommitId( + progressReportSourceTsFile.getCommitterKey(), EnrichedEvent.NO_COMMIT_ID); + } + + // Assign a rebootTime for iotConsensusV2 + enrichedEvent.setRebootTimes(PipeDataNodeAgent.runtime().getRebootTimes()); + + if (enrichedEvent.getPipeName() != null + && (pendingQueue.isEventFromDroppedPipe(enrichedEvent) + || (enrichedEvent.getCommitterKey() == null + && pendingQueue.isPipeDropped( + enrichedEvent.getPipeName(), creationTime, regionId)))) { + enrichedEvent.clearReferenceCount(PipeEventCollector.class.getName()); + return; + } + } + + if (event instanceof PipeHeartbeatEvent) { + ((PipeHeartbeatEvent) event).recordConnectorQueueSize(pendingQueue); + } - if (pendingQueue.offer(event)) { - collectInvocationCount.incrementAndGet(); + isOffered = + memoryReservation == null + ? pendingQueue.offer(event) + : pendingQueue.offer(event, memoryReservation); + if (isOffered) { + memoryReservation = null; + collectInvocationCount.incrementAndGet(); + } + } finally { + if (memoryReservation != null) { + memoryReservation.close(); + } + if (!isOffered + && isReferenceIncreased + && event instanceof EnrichedEvent + && !((EnrichedEvent) event).isReleased()) { + ((EnrichedEvent) event).decreaseReferenceCount(PipeEventCollector.class.getName(), false); + } } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/stage/PipeTaskProcessorStage.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/stage/PipeTaskProcessorStage.java index e2607267fbdd9..6b98d220957cc 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/stage/PipeTaskProcessorStage.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/stage/PipeTaskProcessorStage.java @@ -22,6 +22,7 @@ import org.apache.iotdb.commons.audit.UserEntity; import org.apache.iotdb.commons.consensus.DataRegionId; import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin; +import org.apache.iotdb.commons.pipe.agent.plugin.builtin.processor.donothing.DoNothingProcessor; import org.apache.iotdb.commons.pipe.agent.task.connection.EventSupplier; import org.apache.iotdb.commons.pipe.agent.task.connection.UnboundedBlockingPendingQueue; import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta; @@ -44,6 +45,9 @@ import org.apache.iotdb.pipe.api.event.Event; import org.apache.iotdb.pipe.api.exception.PipeException; +import static org.apache.iotdb.commons.pipe.config.constant.PipeProcessorConstant.PROCESSOR_TSFILE_PARSER_PARALLELISM_DEFAULT_VALUE; +import static org.apache.iotdb.commons.pipe.config.constant.PipeProcessorConstant.PROCESSOR_TSFILE_PARSER_PARALLELISM_KEY; + public class PipeTaskProcessorStage extends PipeTaskStage { private final PipeProcessorSubtaskExecutor executor; @@ -113,7 +117,12 @@ public PipeTaskProcessorStage( regionId, pipeSourceInputEventSupplier, pipeProcessor, - pipeSinkOutputEventCollector); + pipeSinkOutputEventCollector, + pipeProcessor.getClass() == DoNothingProcessor.class + ? pipeProcessorParameters.getIntOrDefault( + PROCESSOR_TSFILE_PARSER_PARALLELISM_KEY, + PROCESSOR_TSFILE_PARSER_PARALLELISM_DEFAULT_VALUE) + : PROCESSOR_TSFILE_PARSER_PARALLELISM_DEFAULT_VALUE); this.executor = executor; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java index 3c4b3d55019d0..4d80ba794a6dc 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java @@ -19,14 +19,18 @@ package org.apache.iotdb.db.pipe.agent.task.subtask.processor; +import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory; +import org.apache.iotdb.commons.concurrent.ThreadName; import org.apache.iotdb.commons.consensus.DataRegionId; import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException; import org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException; +import org.apache.iotdb.commons.pipe.agent.plugin.builtin.processor.donothing.DoNothingProcessor; import org.apache.iotdb.commons.pipe.agent.task.connection.EventSupplier; import org.apache.iotdb.commons.pipe.agent.task.execution.PipeSubtaskScheduler; import org.apache.iotdb.commons.pipe.agent.task.progress.PipeEventCommitManager; import org.apache.iotdb.commons.pipe.agent.task.subtask.PipeReportableSubtask; import org.apache.iotdb.commons.pipe.event.EnrichedEvent; +import org.apache.iotdb.commons.pipe.event.ProgressReportEvent; import org.apache.iotdb.commons.pipe.resource.PipeResourceFailureType; import org.apache.iotdb.commons.pipe.resource.log.PipeLogger; import org.apache.iotdb.commons.utils.ErrorHandlingCommonUtils; @@ -34,6 +38,7 @@ import org.apache.iotdb.db.pipe.agent.task.connection.PipeEventCollector; import org.apache.iotdb.db.pipe.event.UserDefinedEnrichedEvent; import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent; +import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; import org.apache.iotdb.db.pipe.metric.overview.PipeDataNodeSinglePipeMetrics; import org.apache.iotdb.db.pipe.metric.processor.PipeProcessorMetrics; import org.apache.iotdb.db.pipe.processor.pipeconsensus.PipeConsensusProcessor; @@ -49,14 +54,24 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.ArrayDeque; +import java.util.Deque; +import java.util.Iterator; import java.util.Objects; +import java.util.concurrent.CancellationException; +import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; import java.util.concurrent.atomic.AtomicReference; public class PipeProcessorSubtask extends PipeReportableSubtask { private static final Logger LOGGER = LoggerFactory.getLogger(PipeProcessorSubtask.class); + private static final ExecutorService TS_FILE_PARSER_EXECUTOR = + IoTDBThreadPoolFactory.newCachedThreadPool( + ThreadName.PIPE_TSFILE_PARSER_EXECUTOR_POOL.getName()); + private static final AtomicReference subtaskWorkerManager = new AtomicReference<>(); @@ -69,6 +84,12 @@ public class PipeProcessorSubtask extends PipeReportableSubtask { private final PipeProcessor pipeProcessor; private final PipeEventCollector outputEventCollector; + private final int tsFileParserParallelism; + private final Object tsFileParserTaskLock = new Object(); + private final Deque inFlightTsFileParserTasks = new ArrayDeque<>(); + private Event pendingEventAfterTsFileParserBarrier; + private volatile PipeTsFileInsertionEvent retryingFailedTsFileParserEvent; + // This variable is used to distinguish between old and new subtasks before and after stuck // restart. private final long subtaskCreationTime; @@ -81,6 +102,26 @@ public PipeProcessorSubtask( final EventSupplier inputEventSupplier, final PipeProcessor pipeProcessor, final PipeEventCollector outputEventCollector) { + this( + taskID, + pipeName, + creationTime, + regionId, + inputEventSupplier, + pipeProcessor, + outputEventCollector, + 1); + } + + public PipeProcessorSubtask( + final String taskID, + final String pipeName, + final long creationTime, + final int regionId, + final EventSupplier inputEventSupplier, + final PipeProcessor pipeProcessor, + final PipeEventCollector outputEventCollector, + final int tsFileParserParallelism) { super(taskID, creationTime); this.pipeName = pipeName; this.pipeNameWithCreationTime = pipeName + "_" + creationTime; @@ -88,6 +129,10 @@ public PipeProcessorSubtask( this.inputEventSupplier = inputEventSupplier; this.pipeProcessor = pipeProcessor; this.outputEventCollector = outputEventCollector; + this.tsFileParserParallelism = + pipeProcessor.getClass() == DoNothingProcessor.class + ? Math.max(1, tsFileParserParallelism) + : 1; this.subtaskCreationTime = System.currentTimeMillis(); // Only register dataRegions @@ -122,17 +167,60 @@ protected boolean executeOnce() throws Exception { return false; } - final Event event = - lastEvent != null - ? lastEvent - : UserDefinedEnrichedEvent.maybeOf(inputEventSupplier.supply()); - // Record the last event for retry when exception occurs - setLastEvent(event); + // Preserve the event currently being retried. Other parser failures are reaped after this + // event has been resubmitted, otherwise a later failure could overwrite lastEvent. + final TsFileParserTaskResult failedResult = + lastEvent == null ? reapCompletedTsFileParserTasks() : null; + if (failedResult != null) { + if (!retainFailedTsFileParserEvent(failedResult.event)) { + return false; + } + if (ExceptionUtils.getRootCause(failedResult.exception) + instanceof PipeRuntimeOutOfMemoryCriticalException) { + PipeLogger.log( + LOGGER::info, + "Temporarily out of memory in parallel TsFile parsing, will wait for memory to release. Message: %s", + failedResult.exception.getMessage()); + return false; + } + retryingFailedTsFileParserEvent = failedResult.event; + throw new PipeException( + String.format( + "Exception in parallel TsFile parsing, subtask: %s, event: %s, root cause: %s", + taskID, + failedResult.event.coreReportMessage(), + ErrorHandlingCommonUtils.getRootCause(failedResult.exception).getMessage()), + failedResult.exception); + } + + final Event event = getNextEvent(); if (Objects.isNull(event)) { return false; } + if (shouldParseTsFileEventInPool(event)) { + final PipeTsFileInsertionEvent tsFileInsertionEvent = (PipeTsFileInsertionEvent) event; + if (!tsFileInsertionEvent.tryReserveTsFileParserMemory()) { + return false; + } + + try { + outputEventCollector.prepareTsFileEventForParallelParsing(tsFileInsertionEvent); + submitTsFileParserTask(tsFileInsertionEvent); + setLastEvent(null); + return true; + } catch (final Exception e) { + tsFileInsertionEvent.close(); + throw e; + } + } + + if (event != retryingFailedTsFileParserEvent && deferEventUntilTsFileParserBarrier(event)) { + setLastEvent(null); + return false; + } + outputEventCollector.resetFlags(); try { // event can be supplied after the subtask is closed, so we need to check isClosed here @@ -170,15 +258,24 @@ protected boolean executeOnce() throws Exception { // 2. If the event is not collected (not passed to the connector), the reference count // of the event must be zero in the processor stage, at this time, the progress of the // event needs to be reported. - && outputEventCollector.hasNoGeneratedEvent() + && (outputEventCollector.hasNoGeneratedEvent() + || event instanceof PipeTsFileInsertionEvent + && ((PipeTsFileInsertionEvent) event).isProgressReportManagedByTsFileParser()) // If the event's reference count cannot be increased, it means that the event has // been released, and the progress of the event can not be reported. && !outputEventCollector.isFailedToIncreaseReferenceCount() // Events generated from consensusPipe's transferred data should never be reported. && !(pipeProcessor instanceof PipeConsensusProcessor); + if (!shouldReport + && event instanceof PipeTsFileInsertionEvent + && ((PipeTsFileInsertionEvent) event).isProgressReportManagedByTsFileParser()) { + ((PipeTsFileInsertionEvent) event).abortProgressReportManagedByTsFileParser(); + ((PipeTsFileInsertionEvent) event).skipReportOnCommit(); + } if (shouldReport && event instanceof EnrichedEvent - && outputEventCollector.hasNoCollectInvocationAfterReset()) { + && outputEventCollector.hasNoCollectInvocationAfterReset() + && ((EnrichedEvent) event).getCommitId() <= EnrichedEvent.NO_COMMIT_ID) { // An event should be reported here when it is not passed to the connector stage, and it // does not generate any new events to be passed to the connector. In our system, before // reporting an event, we need to enrich a commitKey and commitId, which is done in the @@ -189,6 +286,9 @@ protected boolean executeOnce() throws Exception { .enrichWithCommitterKeyAndCommitId((EnrichedEvent) event, creationTime, regionId); } decreaseReferenceCountAndReleaseLastEvent(event, shouldReport); + if (event == retryingFailedTsFileParserEvent) { + retryingFailedTsFileParserEvent = null; + } } catch (final PipeRuntimeOutOfMemoryCriticalException e) { recordResourceFailure(event, PipeResourceFailureType.MEMORY_TIMEOUT); PipeLogger.log( @@ -226,6 +326,286 @@ protected boolean executeOnce() throws Exception { return true; } + private Event getNextEvent() throws Exception { + if (lastEvent != null) { + return lastEvent; + } + + synchronized (tsFileParserTaskLock) { + if (pendingEventAfterTsFileParserBarrier != null) { + if (!inFlightTsFileParserTasks.isEmpty()) { + return null; + } + final Event event = pendingEventAfterTsFileParserBarrier; + pendingEventAfterTsFileParserBarrier = null; + setLastEvent(event); + return event; + } + + if (inFlightTsFileParserTasks.size() >= tsFileParserParallelism) { + return null; + } + } + + final Event event = UserDefinedEnrichedEvent.maybeOf(inputEventSupplier.supply()); + setLastEvent(event); + return event; + } + + private synchronized boolean retainFailedTsFileParserEvent(final PipeTsFileInsertionEvent event) { + if (isClosed.get()) { + if (!event.isReleased()) { + event.clearReferenceCount(PipeProcessorSubtask.class.getName()); + } + return false; + } + lastEvent = event; + return true; + } + + private boolean shouldParseTsFileEventInPool(final Event event) { + return tsFileParserParallelism > 1 + && event instanceof PipeTsFileInsertionEvent + && event != retryingFailedTsFileParserEvent + && outputEventCollector.shouldParseTsFileEvent((PipeTsFileInsertionEvent) event); + } + + private boolean deferEventUntilTsFileParserBarrier(final Event event) { + // These control events do not depend on parser completion. ProgressReportEvent is committed + // after preceding parser tasks, while PipeHeartbeatEvent does not need ordered commits. Let + // them pass so they do not split a run of parseable TsFiles into small parser batches. + if (event instanceof ProgressReportEvent || event instanceof PipeHeartbeatEvent) { + return false; + } + + synchronized (tsFileParserTaskLock) { + if (isClosed.get() || inFlightTsFileParserTasks.isEmpty()) { + return false; + } + pendingEventAfterTsFileParserBarrier = event; + return true; + } + } + + private void submitTsFileParserTask(final PipeTsFileInsertionEvent event) { + final TsFileParserTask task = + new TsFileParserTask(event, outputEventCollector.forkForTsFileParser()); + synchronized (tsFileParserTaskLock) { + if (isClosed.get()) { + task.cancel(); + return; + } + inFlightTsFileParserTasks.addLast(task); + } + + try { + task.setFuture(TS_FILE_PARSER_EXECUTOR.submit(task::execute)); + } catch (final RuntimeException e) { + synchronized (tsFileParserTaskLock) { + inFlightTsFileParserTasks.remove(task); + } + task.cancel(); + throw e; + } + } + + private TsFileParserTaskResult reapCompletedTsFileParserTasks() + throws InterruptedException, ExecutionException { + while (true) { + final TsFileParserTask task; + synchronized (tsFileParserTaskLock) { + TsFileParserTask completedTask = null; + final Iterator iterator = inFlightTsFileParserTasks.iterator(); + while (iterator.hasNext()) { + final TsFileParserTask candidate = iterator.next(); + if (candidate.isDone()) { + completedTask = candidate; + iterator.remove(); + break; + } + } + task = completedTask; + if (task == null) { + return null; + } + } + + final TsFileParserTaskResult result; + try { + result = task.getResult(); + } catch (final CancellationException e) { + if (isClosed.get()) { + return null; + } + throw e; + } + if (result.exception != null) { + return result; + } + } + } + + private void completeTsFileParserTask( + final PipeTsFileInsertionEvent event, final PipeEventCollector eventCollector) { + final boolean shouldReport = + !isClosed.get() + && (event.isProgressReportManagedByTsFileParser() + || eventCollector.hasNoGeneratedEvent()) + && !eventCollector.isFailedToIncreaseReferenceCount(); + if (!shouldReport && event.isProgressReportManagedByTsFileParser()) { + event.abortProgressReportManagedByTsFileParser(); + event.skipReportOnCommit(); + } + if (shouldReport + && eventCollector.hasNoCollectInvocationAfterReset() + && event.getCommitId() <= EnrichedEvent.NO_COMMIT_ID) { + PipeEventCommitManager.getInstance() + .enrichWithCommitterKeyAndCommitId(event, creationTime, regionId); + } + + if (!event.isReleased()) { + event.decreaseReferenceCount(PipeProcessorSubtask.class.getName(), shouldReport); + } + } + + private class TsFileParserTask { + + private final PipeTsFileInsertionEvent event; + private final PipeEventCollector eventCollector; + + private Future future; + private boolean isStarted; + private boolean isFinished; + private boolean isCancelled; + private boolean ownsEvent = true; + + private TsFileParserTask( + final PipeTsFileInsertionEvent event, final PipeEventCollector eventCollector) { + this.event = event; + this.eventCollector = eventCollector; + } + + private TsFileParserTaskResult execute() { + synchronized (this) { + if (isCancelled) { + return TsFileParserTaskResult.cancelled(event); + } + isStarted = true; + } + + try { + eventCollector.resetFlags(); + eventCollector.collect(event); + event.close(); + + synchronized (this) { + if (isCancelled || isClosed.get()) { + releaseOwnedEvent(); + isFinished = true; + return TsFileParserTaskResult.cancelled(event); + } + } + + PipeProcessorMetrics.getInstance().markTsFileEvent(taskID); + PipeDataNodeSinglePipeMetrics.getInstance() + .markTsFileCollectInvocationCount( + pipeNameWithCreationTime, eventCollector.getCollectInvocationCount()); + completeTsFileParserTask(event, eventCollector); + synchronized (this) { + ownsEvent = false; + isFinished = true; + } + return TsFileParserTaskResult.success(event); + } catch (final Exception e) { + event.releaseTsFileParserMemoryIfReserved(); + synchronized (this) { + if (isCancelled || isClosed.get()) { + releaseOwnedEvent(); + isFinished = true; + return TsFileParserTaskResult.cancelled(event); + } + isFinished = true; + } + return TsFileParserTaskResult.failure(event, e); + } + } + + private synchronized void setFuture(final Future future) { + this.future = future; + if (isCancelled) { + future.cancel(true); + } + } + + private synchronized boolean isDone() { + return future != null && future.isDone(); + } + + private TsFileParserTaskResult getResult() throws InterruptedException, ExecutionException { + final Future currentFuture; + synchronized (this) { + currentFuture = future; + } + final TsFileParserTaskResult result = currentFuture.get(); + if (result.exception != null) { + synchronized (this) { + ownsEvent = false; + } + } + return result; + } + + private void cancel() { + final Future currentFuture; + synchronized (this) { + isCancelled = true; + if ((!isStarted || isFinished) && ownsEvent) { + releaseOwnedEvent(); + } + currentFuture = future; + } + if (currentFuture != null) { + currentFuture.cancel(true); + } + } + + private void releaseOwnedEvent() { + if (!ownsEvent) { + return; + } + event.close(); + if (!event.isReleased()) { + event.clearReferenceCount(PipeProcessorSubtask.class.getName()); + } + ownsEvent = false; + } + } + + private static class TsFileParserTaskResult { + + private final PipeTsFileInsertionEvent event; + private final Exception exception; + + private TsFileParserTaskResult( + final PipeTsFileInsertionEvent event, final Exception exception) { + this.event = event; + this.exception = exception; + } + + private static TsFileParserTaskResult success(final PipeTsFileInsertionEvent event) { + return new TsFileParserTaskResult(event, null); + } + + private static TsFileParserTaskResult failure( + final PipeTsFileInsertionEvent event, final Exception exception) { + return new TsFileParserTaskResult(event, exception); + } + + private static TsFileParserTaskResult cancelled(final PipeTsFileInsertionEvent event) { + return new TsFileParserTaskResult(event, null); + } + } + @Override public void submitSelf() { // this subtask won't be submitted to the executor directly @@ -233,6 +613,15 @@ public void submitSelf() { // and the worker will be submitted to the executor } + @Override + public void onSuccess(final Boolean hasAtLeastOneEventProcessed) { + if (retryingFailedTsFileParserEvent != null) { + submitSelf(); + return; + } + super.onSuccess(hasAtLeastOneEventProcessed); + } + public boolean isStoppedByException() { return lastEvent instanceof EnrichedEvent && retryCount.get() > MAX_RETRY_TIMES; } @@ -243,6 +632,17 @@ public void close() { PipeProcessorMetrics.getInstance().deregister(taskID); try { isClosed.set(true); + retryingFailedTsFileParserEvent = null; + final Event pendingEvent; + synchronized (tsFileParserTaskLock) { + inFlightTsFileParserTasks.forEach(TsFileParserTask::cancel); + inFlightTsFileParserTasks.clear(); + pendingEvent = pendingEventAfterTsFileParserBarrier; + pendingEventAfterTsFileParserBarrier = null; + } + if (pendingEvent instanceof EnrichedEvent && !((EnrichedEvent) pendingEvent).isReleased()) { + ((EnrichedEvent) pendingEvent).clearReferenceCount(PipeProcessorSubtask.class.getName()); + } pipeProcessor.close(); // It is important to note that even if the subtask and its corresponding processor are // closed, the execution thread may still deliver events downstream. 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 f47544ab64f6c..bf803cd8f5056 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 @@ -48,6 +48,9 @@ public class PipeRawTabletInsertionEvent extends EnrichedEvent implements TabletInsertionEvent, ReferenceTrackableEvent, AutoCloseable { + private static final String SOURCE_TSFILE_PROGRESS_HOLDER = + PipeRawTabletInsertionEvent.class.getName() + "#source-tsfile-progress"; + // For better calculation private static final long INSTANCE_SIZE = RamUsageEstimator.shallowSizeOfInstance(PipeRawTabletInsertionEvent.class); @@ -62,7 +65,10 @@ public class PipeRawTabletInsertionEvent extends EnrichedEvent private final EnrichedEvent sourceEvent; private boolean needToReport; + private boolean isSourceTsFileProgressReferenceIncreased; + private final PipeTabletMemoryBlock allocatedMemoryBlock; + private long memoryReservedForNextReferenceIncrease; private TabletInsertionDataContainer dataContainer; @@ -210,15 +216,52 @@ public PipeRawTabletInsertionEvent( @Override public boolean internallyIncreaseResourceReferenceCount(final String holderMessage) { - PipeDataNodeResourceManager.memory() - .forceResize( - allocatedMemoryBlock, - PipeMemoryWeightUtil.calculateTabletSizeInBytes(tablet) + INSTANCE_SIZE); - if (Objects.nonNull(pipeName)) { - PipeDataNodeSinglePipeMetrics.getInstance() - .increaseRawTabletEventCount(pipeName, creationTime); + final PipeTsFileInsertionEvent progressReportSourceTsFile = getProgressReportSourceTsFile(); + if (progressReportSourceTsFile != null) { + if (!progressReportSourceTsFile.increaseReferenceCount(SOURCE_TSFILE_PROGRESS_HOLDER)) { + return false; + } + isSourceTsFileProgressReferenceIncreased = true; + } + + try { + final long targetMemoryInBytes = getTabletSizeInBytes() + INSTANCE_SIZE; + if (memoryReservedForNextReferenceIncrease > 0) { + PipeDataNodeResourceManager.memory() + .forceResizeWithReservedMemory( + allocatedMemoryBlock, targetMemoryInBytes, memoryReservedForNextReferenceIncrease); + } else { + PipeDataNodeResourceManager.memory().forceResize(allocatedMemoryBlock, targetMemoryInBytes); + } + if (Objects.nonNull(pipeName)) { + PipeDataNodeSinglePipeMetrics.getInstance() + .increaseRawTabletEventCount(pipeName, creationTime); + } + return true; + } catch (final RuntimeException e) { + // The tablet has not been published yet. Roll back its source reference without aborting a + // retry of the same TsFile. + releaseSourceTsFileProgressReference(true, false); + throw e; } - return true; + } + + public synchronized boolean increaseReferenceCountWithReservedMemory( + final String holderMessage, final long reservedMemoryInBytes) { + if (referenceCount.get() > 0) { + return super.increaseReferenceCount(holderMessage); + } + + memoryReservedForNextReferenceIncrease = Math.max(0, reservedMemoryInBytes); + try { + return super.increaseReferenceCount(holderMessage); + } finally { + memoryReservedForNextReferenceIncrease = 0; + } + } + + public long getTabletSizeInBytes() { + return PipeMemoryWeightUtil.calculateTabletSizeInBytes(tablet); } @Override @@ -255,9 +298,28 @@ public boolean internallyDecreaseResourceReferenceCount(final String holderMessa } } + releaseSourceTsFileProgressReference(shouldReportOnCommit, !shouldReportOnCommit); + return true; } + private void releaseSourceTsFileProgressReference( + final boolean shouldReport, final boolean shouldAbortSourceProgressReport) { + if (!isSourceTsFileProgressReferenceIncreased) { + return; + } + isSourceTsFileProgressReferenceIncreased = false; + + final PipeTsFileInsertionEvent progressReportSourceTsFile = getProgressReportSourceTsFile(); + if (progressReportSourceTsFile != null && !progressReportSourceTsFile.isReleased()) { + if (shouldAbortSourceProgressReport) { + progressReportSourceTsFile.abortProgressReportManagedByTsFileParser(); + } + progressReportSourceTsFile.decreaseReferenceCount( + SOURCE_TSFILE_PROGRESS_HOLDER, shouldReport); + } + } + protected void eliminateProgressIndex() { if (sourceEvent instanceof PipeTsFileInsertionEvent) { ((PipeTsFileInsertionEvent) sourceEvent).eliminateProgressIndex(); @@ -388,6 +450,22 @@ public String getTableModelDatabaseName() { : sourceDatabaseNameFromDataRegion; } + public PipeTsFileInsertionEvent getProgressReportSourceTsFile() { + if (sourceEvent instanceof PipeTsFileInsertionEvent + && ((PipeTsFileInsertionEvent) sourceEvent).isProgressReportManagedByTsFileParser()) { + return (PipeTsFileInsertionEvent) sourceEvent; + } + if (sourceEvent instanceof PipeRawTabletInsertionEvent) { + return ((PipeRawTabletInsertionEvent) sourceEvent).getProgressReportSourceTsFile(); + } + return null; + } + + @Override + public boolean needToCommit() { + return getProgressReportSourceTsFile() == null; + } + @Override public boolean isShouldReportOnCommit() { return shouldReportOnCommit && needToReport; @@ -460,11 +538,14 @@ public PipeRawTabletInsertionEvent parseEventWithPatternOrTime() { treeModelDatabaseName, convertToTablet(), isAligned, + this, + needToReport, pipeName, creationTime, pipeTaskMeta, - this, - needToReport); + null, + Long.MIN_VALUE, + Long.MAX_VALUE); } public boolean hasNoNeedParsingAndIsEmpty() { 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 b6938890cf0cd..9fd759beb1d41 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 @@ -96,6 +96,8 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent new AtomicReference<>(); private final AtomicInteger parsedTabletInsertionEventCount = new AtomicInteger(0); private final AtomicBoolean isTsFileParsingCompleted = new AtomicBoolean(false); + private final AtomicBoolean isProgressReportManagedByTsFileParser = new AtomicBoolean(false); + private final AtomicBoolean isTsFileParserProgressReportAborted = new AtomicBoolean(false); private final AtomicLong parsedPointCountForCount = new AtomicLong(0); // The point count of the TsFile. Used for metrics on PipeConsensus' receiver side. @@ -760,7 +762,15 @@ private boolean tryReserveTsFileParserMemory(final PipeMemoryManager memoryManag } } - private void releaseTsFileParserMemoryIfReserved() { + /** + * Tries to reserve a parser slot without blocking. This is used by the processor-side parser pool + * so that only admitted TsFiles occupy parser worker threads. + */ + public boolean tryReserveTsFileParserMemory() { + return tryReserveTsFileParserMemory(PipeDataNodeResourceManager.memory()); + } + + public void releaseTsFileParserMemoryIfReserved() { synchronized (isTsFileParserMemoryReserved) { if (isTsFileParserMemoryReserved.compareAndSet(true, false)) { PipeDataNodeResourceManager.memory() @@ -786,6 +796,27 @@ public boolean isGeneratedByHistoricalExtractor() { return isGeneratedByHistoricalExtractor; } + public void markProgressReportManagedByTsFileParser() { + isProgressReportManagedByTsFileParser.set(true); + } + + public boolean isProgressReportManagedByTsFileParser() { + return isProgressReportManagedByTsFileParser.get(); + } + + public void abortProgressReportManagedByTsFileParser() { + if (isProgressReportManagedByTsFileParser.get() + && isTsFileParserProgressReportAborted.compareAndSet(false, true)) { + LOGGER.warn("Abort progress report for partially transferred parsed TsFile: {}", tsFile); + } + } + + @Override + public boolean needToCommit() { + return !isProgressReportManagedByTsFileParser.get() + || !isTsFileParserProgressReportAborted.get(); + } + private TsFileInsertionDataContainer initDataContainer() { try { dataContainer.compareAndSet( 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 5dfff56c52e1c..2aa2366ef58c6 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 @@ -182,9 +182,7 @@ public synchronized boolean tryReserveTsFileParserMemory( enqueueTsFileParserReservationRequest(pipeRegionIdentity, reservationKey); final int globalLimit = Math.max(1, PIPE_CONFIG.getPipeTsFileParserInFlightMaxNum()); - final int perPipeRegionLimit = - Math.max( - 1, Math.min(globalLimit, PIPE_CONFIG.getPipeTsFileParserInFlightMaxNumPerPipeRegion())); + final int perPipeRegionLimit = getTsFileParserInFlightMaxNumPerPipeRegion(globalLimit); final int reservedCountOfPipeRegion = reservedTsFileParserCountByPipeRegion.getOrDefault(pipeRegionIdentity, 0); if (reservedTsFileParserCount >= globalLimit @@ -310,9 +308,7 @@ private void notifyNextTsFileParserMemoryReservationInternal() { return; } - final int perPipeRegionLimit = - Math.max( - 1, Math.min(globalLimit, PIPE_CONFIG.getPipeTsFileParserInFlightMaxNumPerPipeRegion())); + final int perPipeRegionLimit = getTsFileParserInFlightMaxNumPerPipeRegion(globalLimit); final PipeRegionIdentity nextPipeRegion = getNextEligibleTsFileParserPipeRegion(perPipeRegionLimit, !isSoftMemoryEnough); if (nextPipeRegion == null) { @@ -370,6 +366,11 @@ private PipeRegionIdentity getNextEligibleTsFileParserPipeRegion( return firstEligiblePipeRegion; } + private static int getTsFileParserInFlightMaxNumPerPipeRegion(final int globalLimit) { + final int configuredLimit = PIPE_CONFIG.getPipeTsFileParserInFlightMaxNumPerPipeRegion(); + return configuredLimit <= 0 ? globalLimit : Math.min(globalLimit, configuredLimit); + } + private void clearTsFileParserAdmissionCursorIfIdle() { if (reservedTsFileParserCount == 0 && waitingTsFileParserPipeOrder.isEmpty()) { lastAdmittedWaitingTsFileParserPipe = null; @@ -677,8 +678,20 @@ public void forceResize(final PipeMemoryBlock block, final long targetSize) { resize(block, targetSize, true); } - public synchronized void resize( - final PipeMemoryBlock block, final long targetSize, final boolean force) { + public void forceResizeWithReservedMemory( + final PipeMemoryBlock block, final long targetSize, final long reservedMemoryInBytes) { + resize(block, targetSize, true, Math.max(0, reservedMemoryInBytes)); + } + + public void resize(final PipeMemoryBlock block, final long targetSize, final boolean force) { + resize(block, targetSize, force, 0); + } + + private synchronized void resize( + final PipeMemoryBlock block, + final long targetSize, + final boolean force, + final long reservedMemoryInBytes) { if (block == null || block.isReleased()) { LOGGER.warn("forceResize: cannot resize a null or released memory block"); return; @@ -710,14 +723,19 @@ public synchronized void resize( return; } - long sizeInBytes = targetSize - oldSize; + final long sizeInBytes = targetSize - oldSize; + final long requiredFreeMemoryInBytes = + sizeInBytes > Long.MAX_VALUE - reservedMemoryInBytes + ? Long.MAX_VALUE + : sizeInBytes + reservedMemoryInBytes; final int memoryAllocateMaxRetries = PipeConfig.getInstance().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() - usedMemorySizeInBytes >= sizeInBytes) { + && getTotalNonFloatingMemorySizeInBytes() - usedMemorySizeInBytes + >= requiredFreeMemoryInBytes) { usedMemorySizeInBytes += sizeInBytes; if (oldSize == 0) { // If the memory block is not registered, we need to register it first. @@ -736,7 +754,7 @@ && getTotalNonFloatingMemorySizeInBytes() - usedMemorySizeInBytes >= sizeInBytes } try { - tryShrinkUntilFreeMemorySatisfy(sizeInBytes); + tryShrinkUntilFreeMemorySatisfy(requiredFreeMemoryInBytes); this.wait(PipeConfig.getInstance().getPipeMemoryAllocateRetryIntervalInMs()); } catch (InterruptedException e) { Thread.currentThread().interrupt(); @@ -749,11 +767,12 @@ && getTotalNonFloatingMemorySizeInBytes() - usedMemorySizeInBytes >= sizeInBytes String.format( "forceResize: failed to allocate memory after %d retries, " + "total memory size %d bytes, used memory size %d bytes, " - + "requested memory size %d bytes", + + "requested memory size %d bytes, reserved memory size %d bytes", memoryAllocateMaxRetries, getTotalNonFloatingMemorySizeInBytes(), usedMemorySizeInBytes, - sizeInBytes)); + sizeInBytes, + reservedMemoryInBytes)); } } 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 3e0ec9f779efd..5036cb986e9c0 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 @@ -22,7 +22,7 @@ import org.apache.iotdb.commons.pipe.agent.task.progress.CommitterKey; import org.apache.iotdb.commons.pipe.event.EnrichedEvent; 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.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; @@ -48,7 +48,7 @@ public abstract class PipeTabletEventBatch implements AutoCloseable { private long firstEventProcessingTime = Long.MIN_VALUE; protected long totalBufferSize = 0; - private final PipeMemoryBlock allocatedMemoryBlock; + private final PipeTabletMemoryBlock allocatedMemoryBlock; protected volatile boolean isClosed = false; @@ -60,7 +60,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 { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/PropertiesTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/PropertiesTest.java index 458871ec4f889..af1465b659657 100755 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/PropertiesTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/PropertiesTest.java @@ -53,6 +53,10 @@ public void testHotReloadTsFileParserInFlightLimits() throws Exception { Assert.assertEquals(3, commonConfig.getPipeTsFileParserInFlightMaxNum()); Assert.assertEquals(2, commonConfig.getPipeTsFileParserInFlightMaxNumPerPipeRegion()); + + properties.setProperty("pipe_tsfile_parser_in_flight_max_num_per_pipe_region", "0"); + descriptor.loadHotModifiedProps(properties); + Assert.assertEquals(0, commonConfig.getPipeTsFileParserInFlightMaxNumPerPipeRegion()); } finally { final TrimProperties properties = new TrimProperties(); properties.setProperty( diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeProcessorSubtaskExecutorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeProcessorSubtaskExecutorTest.java index a403e83329ef1..38e373f0e0c53 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeProcessorSubtaskExecutorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeProcessorSubtaskExecutorTest.java @@ -20,16 +20,20 @@ package org.apache.iotdb.db.pipe.agent.task; import org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException; +import org.apache.iotdb.commons.pipe.agent.plugin.builtin.processor.donothing.DoNothingProcessor; import org.apache.iotdb.commons.pipe.agent.task.connection.EventSupplier; +import org.apache.iotdb.commons.pipe.event.ProgressReportEvent; import org.apache.iotdb.db.pipe.agent.task.connection.PipeEventCollector; import org.apache.iotdb.db.pipe.agent.task.execution.PipeProcessorSubtaskExecutor; import org.apache.iotdb.db.pipe.agent.task.subtask.processor.PipeProcessorSubtask; +import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent; import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; 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.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.pipe.api.PipeProcessor; +import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent; import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent; import org.apache.iotdb.pipe.api.exception.PipeException; @@ -39,7 +43,12 @@ import org.mockito.Mockito; import java.io.File; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -122,6 +131,359 @@ public void testProcessorSubtaskTreatsOutOfMemoryCauseAsTemporaryFailure() throw Assert.assertFalse(pipeProcessorSubtask.executeOnceForTest()); } + @Test + public void testTsFilesCanBeParsedInParallelInOneProcessorSubtask() throws Exception { + final EventSupplier eventSupplier = mock(EventSupplier.class); + final PipeEventCollector pipeEventCollector = mock(PipeEventCollector.class); + final PipeEventCollector firstParserCollector = mock(PipeEventCollector.class); + final PipeEventCollector secondParserCollector = mock(PipeEventCollector.class); + final PipeTsFileInsertionEvent firstEvent = mock(PipeTsFileInsertionEvent.class); + final PipeTsFileInsertionEvent secondEvent = mock(PipeTsFileInsertionEvent.class); + + when(eventSupplier.supply()).thenReturn(firstEvent, secondEvent, null); + when(pipeEventCollector.shouldParseTsFileEvent(any(PipeTsFileInsertionEvent.class))) + .thenReturn(true); + when(pipeEventCollector.forkForTsFileParser()) + .thenReturn(firstParserCollector, secondParserCollector); + when(firstEvent.tryReserveTsFileParserMemory()).thenReturn(true); + when(secondEvent.tryReserveTsFileParserMemory()).thenReturn(true); + + final CountDownLatch parsersStarted = new CountDownLatch(2); + final CountDownLatch releaseParsers = new CountDownLatch(1); + doAnswer( + invocation -> { + parsersStarted.countDown(); + releaseParsers.await(5, TimeUnit.SECONDS); + return null; + }) + .when(firstParserCollector) + .collect(firstEvent); + doAnswer( + invocation -> { + parsersStarted.countDown(); + releaseParsers.await(5, TimeUnit.SECONDS); + return null; + }) + .when(secondParserCollector) + .collect(secondEvent); + + final TestablePipeProcessorSubtask pipeProcessorSubtask = + new TestablePipeProcessorSubtask( + "parallel-test", + "pipe", + System.currentTimeMillis(), + 0, + eventSupplier, + new DoNothingProcessor(), + pipeEventCollector, + 2); + try { + Assert.assertTrue(pipeProcessorSubtask.executeOnceForTest()); + Assert.assertTrue(pipeProcessorSubtask.executeOnceForTest()); + Assert.assertTrue(parsersStarted.await(5, TimeUnit.SECONDS)); + } finally { + releaseParsers.countDown(); + pipeProcessorSubtask.close(); + } + } + + @Test + public void testNonTsFileEventWaitsForInFlightTsFileParser() throws Exception { + final EventSupplier eventSupplier = mock(EventSupplier.class); + final PipeEventCollector pipeEventCollector = mock(PipeEventCollector.class); + final PipeEventCollector parserCollector = mock(PipeEventCollector.class); + final PipeTsFileInsertionEvent tsFileEvent = mock(PipeTsFileInsertionEvent.class); + final TabletInsertionEvent barrierEvent = mock(TabletInsertionEvent.class); + + when(eventSupplier.supply()).thenReturn(tsFileEvent, barrierEvent, null); + when(pipeEventCollector.shouldParseTsFileEvent(tsFileEvent)).thenReturn(true); + when(pipeEventCollector.forkForTsFileParser()).thenReturn(parserCollector); + when(tsFileEvent.tryReserveTsFileParserMemory()).thenReturn(true); + + final CountDownLatch parserStarted = new CountDownLatch(1); + final CountDownLatch releaseParser = new CountDownLatch(1); + doAnswer( + invocation -> { + parserStarted.countDown(); + releaseParser.await(5, TimeUnit.SECONDS); + return null; + }) + .when(parserCollector) + .collect(tsFileEvent); + + final TestablePipeProcessorSubtask pipeProcessorSubtask = + new TestablePipeProcessorSubtask( + "barrier-test", + "pipe", + System.currentTimeMillis(), + 0, + eventSupplier, + new DoNothingProcessor(), + pipeEventCollector, + 2); + try { + Assert.assertTrue(pipeProcessorSubtask.executeOnceForTest()); + Assert.assertTrue(parserStarted.await(5, TimeUnit.SECONDS)); + + Assert.assertFalse(pipeProcessorSubtask.executeOnceForTest()); + Mockito.verify(pipeEventCollector, Mockito.never()).collect(barrierEvent); + + releaseParser.countDown(); + final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + boolean barrierProcessed = false; + while (!barrierProcessed && System.nanoTime() < deadline) { + barrierProcessed = pipeProcessorSubtask.executeOnceForTest(); + if (!barrierProcessed) { + Thread.sleep(10); + } + } + Assert.assertTrue(barrierProcessed); + Mockito.verify(pipeEventCollector).collect(barrierEvent); + } finally { + releaseParser.countDown(); + pipeProcessorSubtask.close(); + } + } + + @Test + public void testProgressReportEventDoesNotWaitForInFlightTsFileParser() throws Exception { + final EventSupplier eventSupplier = mock(EventSupplier.class); + final PipeEventCollector pipeEventCollector = mock(PipeEventCollector.class); + final PipeEventCollector firstParserCollector = mock(PipeEventCollector.class); + final PipeEventCollector secondParserCollector = mock(PipeEventCollector.class); + final PipeTsFileInsertionEvent firstTsFileEvent = mock(PipeTsFileInsertionEvent.class); + final ProgressReportEvent progressReportEvent = mock(ProgressReportEvent.class); + final PipeTsFileInsertionEvent secondTsFileEvent = mock(PipeTsFileInsertionEvent.class); + + when(eventSupplier.supply()) + .thenReturn(firstTsFileEvent, progressReportEvent, secondTsFileEvent, null); + when(pipeEventCollector.shouldParseTsFileEvent(any(PipeTsFileInsertionEvent.class))) + .thenReturn(true); + when(pipeEventCollector.forkForTsFileParser()) + .thenReturn(firstParserCollector, secondParserCollector); + when(firstTsFileEvent.tryReserveTsFileParserMemory()).thenReturn(true); + when(secondTsFileEvent.tryReserveTsFileParserMemory()).thenReturn(true); + + final CountDownLatch parsersStarted = new CountDownLatch(2); + final CountDownLatch releaseParsers = new CountDownLatch(1); + doAnswer( + invocation -> { + parsersStarted.countDown(); + releaseParsers.await(5, TimeUnit.SECONDS); + return null; + }) + .when(firstParserCollector) + .collect(firstTsFileEvent); + doAnswer( + invocation -> { + parsersStarted.countDown(); + releaseParsers.await(5, TimeUnit.SECONDS); + return null; + }) + .when(secondParserCollector) + .collect(secondTsFileEvent); + + final TestablePipeProcessorSubtask pipeProcessorSubtask = + new TestablePipeProcessorSubtask( + "progress-report-barrier-test", + "pipe", + System.currentTimeMillis(), + 0, + eventSupplier, + new DoNothingProcessor(), + pipeEventCollector, + 2); + try { + Assert.assertTrue(pipeProcessorSubtask.executeOnceForTest()); + Assert.assertTrue(pipeProcessorSubtask.executeOnceForTest()); + Mockito.verify(pipeEventCollector).collect(progressReportEvent); + Assert.assertTrue(pipeProcessorSubtask.executeOnceForTest()); + Assert.assertTrue(parsersStarted.await(5, TimeUnit.SECONDS)); + } finally { + releaseParsers.countDown(); + pipeProcessorSubtask.close(); + } + } + + @Test + public void testHeartbeatEventDoesNotWaitForInFlightTsFileParser() throws Exception { + final EventSupplier eventSupplier = mock(EventSupplier.class); + final PipeEventCollector pipeEventCollector = mock(PipeEventCollector.class); + final PipeEventCollector firstParserCollector = mock(PipeEventCollector.class); + final PipeEventCollector secondParserCollector = mock(PipeEventCollector.class); + final PipeTsFileInsertionEvent firstTsFileEvent = mock(PipeTsFileInsertionEvent.class); + final PipeHeartbeatEvent heartbeatEvent = mock(PipeHeartbeatEvent.class); + final PipeTsFileInsertionEvent secondTsFileEvent = mock(PipeTsFileInsertionEvent.class); + + when(eventSupplier.supply()) + .thenReturn(firstTsFileEvent, heartbeatEvent, secondTsFileEvent, null); + when(pipeEventCollector.shouldParseTsFileEvent(any(PipeTsFileInsertionEvent.class))) + .thenReturn(true); + when(pipeEventCollector.forkForTsFileParser()) + .thenReturn(firstParserCollector, secondParserCollector); + when(firstTsFileEvent.tryReserveTsFileParserMemory()).thenReturn(true); + when(secondTsFileEvent.tryReserveTsFileParserMemory()).thenReturn(true); + + final CountDownLatch parsersStarted = new CountDownLatch(2); + final CountDownLatch releaseParsers = new CountDownLatch(1); + doAnswer( + invocation -> { + parsersStarted.countDown(); + releaseParsers.await(5, TimeUnit.SECONDS); + return null; + }) + .when(firstParserCollector) + .collect(firstTsFileEvent); + doAnswer( + invocation -> { + parsersStarted.countDown(); + releaseParsers.await(5, TimeUnit.SECONDS); + return null; + }) + .when(secondParserCollector) + .collect(secondTsFileEvent); + + final TestablePipeProcessorSubtask pipeProcessorSubtask = + new TestablePipeProcessorSubtask( + "heartbeat-barrier-test", + "pipe", + System.currentTimeMillis(), + 0, + eventSupplier, + new DoNothingProcessor(), + pipeEventCollector, + 2); + try { + Assert.assertTrue(pipeProcessorSubtask.executeOnceForTest()); + Assert.assertTrue(pipeProcessorSubtask.executeOnceForTest()); + Mockito.verify(pipeEventCollector).collect(heartbeatEvent); + Mockito.verify(heartbeatEvent).onProcessed(); + Assert.assertTrue(pipeProcessorSubtask.executeOnceForTest()); + Assert.assertTrue(parsersStarted.await(5, TimeUnit.SECONDS)); + } finally { + releaseParsers.countDown(); + pipeProcessorSubtask.close(); + } + } + + @Test + public void testPermanentParallelParserFailureReachesRetryLimit() throws Exception { + final EventSupplier eventSupplier = mock(EventSupplier.class); + final PipeEventCollector pipeEventCollector = mock(PipeEventCollector.class); + final PipeEventCollector parserCollector = mock(PipeEventCollector.class); + final PipeTsFileInsertionEvent tsFileEvent = mock(PipeTsFileInsertionEvent.class); + final PipeException parserFailure = new PipeException("broken TsFile"); + + when(eventSupplier.supply()).thenReturn(tsFileEvent, null); + when(pipeEventCollector.shouldParseTsFileEvent(tsFileEvent)).thenReturn(true); + when(pipeEventCollector.forkForTsFileParser()).thenReturn(parserCollector); + when(tsFileEvent.tryReserveTsFileParserMemory()).thenReturn(true); + doThrow(parserFailure).when(parserCollector).collect(tsFileEvent); + doThrow(parserFailure).when(pipeEventCollector).collect(tsFileEvent); + + final TestablePipeProcessorSubtask pipeProcessorSubtask = + new TestablePipeProcessorSubtask( + "parser-retry-limit-test", + "pipe", + System.currentTimeMillis(), + 0, + eventSupplier, + new DoNothingProcessor(), + pipeEventCollector, + 2); + try { + Assert.assertTrue(pipeProcessorSubtask.executeOnceForTest()); + awaitParserFailure(pipeProcessorSubtask); + pipeProcessorSubtask.recordFailureForTest(); + + for (int retry = 0; retry < PipeProcessorSubtask.MAX_RETRY_TIMES; retry++) { + Assert.assertThrows(PipeException.class, pipeProcessorSubtask::executeOnceForTest); + pipeProcessorSubtask.recordFailureForTest(); + } + + Assert.assertEquals( + PipeProcessorSubtask.MAX_RETRY_TIMES + 1, pipeProcessorSubtask.getRetryCountForTest()); + Assert.assertTrue(pipeProcessorSubtask.isStoppedByException()); + Mockito.verify(pipeEventCollector, Mockito.times(1)).forkForTsFileParser(); + } finally { + pipeProcessorSubtask.close(); + } + } + + @Test + public void testParallelParserRetryCountClearsOnlyAfterSuccessfulRetry() throws Exception { + final EventSupplier eventSupplier = mock(EventSupplier.class); + final PipeEventCollector pipeEventCollector = mock(PipeEventCollector.class); + final PipeEventCollector parserCollector = mock(PipeEventCollector.class); + final long creationTime = System.currentTimeMillis(); + final File tsFile = new File("target/testParallelParserRetry.tsfile"); + final TsFileResource resource = mock(TsFileResource.class); + when(resource.getTsFilePath()).thenReturn(tsFile.getPath()); + final PipeTsFileInsertionEvent tsFileEvent = + mock( + PipeTsFileInsertionEvent.class, + Mockito.withSettings() + .useConstructor( + resource, tsFile, false, false, false, "pipe", creationTime, null, null, 0L, 1L) + .defaultAnswer(Mockito.RETURNS_DEFAULTS)); + + when(eventSupplier.supply()).thenReturn(tsFileEvent, null); + when(pipeEventCollector.shouldParseTsFileEvent(tsFileEvent)).thenReturn(true); + when(pipeEventCollector.forkForTsFileParser()).thenReturn(parserCollector); + doReturn(true).when(tsFileEvent).tryReserveTsFileParserMemory(); + doThrow(new PipeException("transient parser failure")) + .when(parserCollector) + .collect(tsFileEvent); + doThrow( + new PipeException( + "temporary memory pressure", + new PipeRuntimeOutOfMemoryCriticalException("parser memory unavailable"))) + .doNothing() + .when(pipeEventCollector) + .collect(tsFileEvent); + + final TestablePipeProcessorSubtask pipeProcessorSubtask = + new TestablePipeProcessorSubtask( + "parser-retry-success-test", + "pipe", + creationTime, + 0, + eventSupplier, + new DoNothingProcessor(), + pipeEventCollector, + 2); + try { + Assert.assertTrue(pipeProcessorSubtask.executeOnceForTest()); + awaitParserFailure(pipeProcessorSubtask); + pipeProcessorSubtask.recordFailureForTest(); + + Assert.assertFalse(pipeProcessorSubtask.executeOnceForTest()); + pipeProcessorSubtask.onSuccess(false); + Assert.assertEquals(1, pipeProcessorSubtask.getRetryCountForTest()); + + Assert.assertTrue(pipeProcessorSubtask.executeOnceForTest()); + pipeProcessorSubtask.onSuccess(true); + Assert.assertEquals(0, pipeProcessorSubtask.getRetryCountForTest()); + Mockito.verify(pipeEventCollector, Mockito.times(1)).forkForTsFileParser(); + } finally { + pipeProcessorSubtask.close(); + } + } + + private static PipeException awaitParserFailure( + final TestablePipeProcessorSubtask pipeProcessorSubtask) throws Exception { + final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + while (System.nanoTime() < deadline) { + try { + pipeProcessorSubtask.executeOnceForTest(); + } catch (final PipeException e) { + return e; + } + Thread.sleep(10); + } + throw new AssertionError("Timed out waiting for parallel parser failure"); + } + private static class TestablePipeProcessorSubtask extends PipeProcessorSubtask { private TestablePipeProcessorSubtask( @@ -142,8 +504,36 @@ private TestablePipeProcessorSubtask( outputEventCollector); } + private TestablePipeProcessorSubtask( + final String taskID, + final String pipeName, + final long creationTime, + final int regionId, + final EventSupplier inputEventSupplier, + final PipeProcessor pipeProcessor, + final PipeEventCollector outputEventCollector, + final int tsFileParserParallelism) { + super( + taskID, + pipeName, + creationTime, + regionId, + inputEventSupplier, + pipeProcessor, + outputEventCollector, + tsFileParserParallelism); + } + private boolean executeOnceForTest() throws Exception { return executeOnce(); } + + private void recordFailureForTest() { + retryCount.incrementAndGet(); + } + + private int getRetryCountForTest() { + return retryCount.get(); + } } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEventTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEventTest.java new file mode 100644 index 0000000000000..1567527acbd0a --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEventTest.java @@ -0,0 +1,113 @@ +/* + * 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.event.common.tablet; + +import org.apache.iotdb.commons.pipe.event.EnrichedEvent; +import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; +import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; + +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.write.record.Tablet; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.junit.Assert; +import org.junit.Test; + +import java.io.File; +import java.util.Collections; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +public class PipeRawTabletInsertionEventTest { + + @Test + public void testFailedParsedTabletAbortsSourceProgressRegardlessOfReleaseOrder() { + assertFailedParsedTabletAbortsSourceProgress(true); + assertFailedParsedTabletAbortsSourceProgress(false); + } + + private static void assertFailedParsedTabletAbortsSourceProgress( + final boolean releaseFailedTabletFirst) { + final TestPipeTsFileInsertionEvent sourceEvent = createProgressManagedSourceEvent(); + final PipeRawTabletInsertionEvent failedTablet = createEvent(sourceEvent, false); + final PipeRawTabletInsertionEvent successfulTablet = createEvent(sourceEvent, true); + + Assert.assertFalse(failedTablet.needToCommit()); + Assert.assertFalse(successfulTablet.needToCommit()); + Assert.assertTrue(sourceEvent.increaseReferenceCount("processor")); + Assert.assertTrue(failedTablet.increaseReferenceCount("collector")); + Assert.assertTrue(successfulTablet.increaseReferenceCount("collector")); + Assert.assertEquals(3, sourceEvent.getReferenceCount()); + + sourceEvent.decreaseReferenceCount("processor", true); + if (releaseFailedTabletFirst) { + failedTablet.clearReferenceCount("discarded"); + successfulTablet.decreaseReferenceCount("transferred", true); + } else { + successfulTablet.decreaseReferenceCount("transferred", true); + failedTablet.clearReferenceCount("discarded"); + } + + Assert.assertTrue(sourceEvent.isReleased()); + Assert.assertFalse(sourceEvent.needToCommit()); + } + + private static TestPipeTsFileInsertionEvent createProgressManagedSourceEvent() { + final File tsFile = new File("target/source-progress.tsfile"); + final TsFileResource resource = mock(TsFileResource.class); + when(resource.getTsFile()).thenReturn(tsFile); + when(resource.getModFile()) + .thenReturn(new ModificationFile(tsFile.getPath() + ModificationFile.FILE_SUFFIX)); + when(resource.isClosed()).thenReturn(true); + + final TestPipeTsFileInsertionEvent sourceEvent = + new TestPipeTsFileInsertionEvent(resource, tsFile); + sourceEvent.markProgressReportManagedByTsFileParser(); + return sourceEvent; + } + + private static PipeRawTabletInsertionEvent createEvent( + final EnrichedEvent sourceEvent, final boolean needToReport) { + final MeasurementSchema schema = new MeasurementSchema("s", TSDataType.INT64); + final Tablet tablet = new Tablet("root.sg.d", Collections.singletonList(schema), 1); + tablet.addTimestamp(0, 1); + tablet.addValue("s", 0, 1L); + return new PipeRawTabletInsertionEvent( + tablet, false, "pipe", 1, null, sourceEvent, needToReport); + } + + private static class TestPipeTsFileInsertionEvent extends PipeTsFileInsertionEvent { + + private TestPipeTsFileInsertionEvent(final TsFileResource resource, final File tsFile) { + super(resource, tsFile, false, false, false, "pipe", 1, null, null, 0, 1); + } + + @Override + public boolean internallyIncreaseResourceReferenceCount(final String holderMessage) { + return true; + } + + @Override + public boolean internallyDecreaseResourceReferenceCount(final String holderMessage) { + return true; + } + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java index 3a50002667e37..256a346189d9a 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java @@ -124,6 +124,17 @@ public void testGlobalAndPerPipeRegionLimitsAreBothEnforced() { Assert.assertTrue(tryAcquire(pipeC)); } + @Test + public void testNonPositivePerPipeRegionLimitFollowsGlobalLimit() { + commonConfig.setPipeTsFileParserInFlightMaxNum(2); + commonConfig.setPipeTsFileParserInFlightMaxNumPerPipeRegion(0); + + final Reservation first = new Reservation("pipe", 1, "1"); + final Reservation second = new Reservation("pipe", 1, "1"); + Assert.assertTrue(tryAcquire(first)); + Assert.assertTrue(tryAcquire(second)); + } + @Test public void testDifferentRegionsOfSamePipeCanRunConcurrently() { commonConfig.setPipeTsFileParserInFlightMaxNum(2); diff --git a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template index 2bc03179bd14d..0491c9232a4e6 100644 --- a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template +++ b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template @@ -1819,10 +1819,12 @@ pipe_tsfile_parser_in_flight_max_num=0 # The maximum number of TsFile parsers that can run concurrently for one DataRegion of one Pipe. # Different DataRegions of the same Pipe have independent limits. -# When <= 0, use 1. +# When <= 0, follow pipe_tsfile_parser_in_flight_max_num. +# Per-Pipe parallel parsing remains disabled unless processor.tsfile-parser.parallelism is greater +# than 1 for that Pipe. # effectiveMode: hot_reload # Datatype: int -pipe_tsfile_parser_in_flight_max_num_per_pipe_region=1 +pipe_tsfile_parser_in_flight_max_num_per_pipe_region=0 # The connection timeout (in milliseconds) for the thrift client. # effectiveMode: restart diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java index a902893094056..7dfffc5f98fb9 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java @@ -130,6 +130,7 @@ public enum ThreadName { // -------------------------- Compute -------------------------- PIPE_SOURCE_DISRUPTOR("Pipe-Source-Disruptor"), PIPE_PROCESSOR_EXECUTOR_POOL("Pipe-Processor-Executor-Pool"), + PIPE_TSFILE_PARSER_EXECUTOR_POOL("Pipe-TsFile-Parser-Executor-Pool"), PIPE_CONSENSUS_EXECUTOR_POOL("Pipe-Consensus-Executor-Pool"), PIPE_SINK_EXECUTOR_POOL("Pipe-Sink-Executor-Pool"), PIPE_CONFIGNODE_EXECUTOR_POOL("Pipe-ConfigNode-Executor-Pool"), @@ -289,6 +290,7 @@ public enum ThreadName { Arrays.asList( PIPE_SOURCE_DISRUPTOR, PIPE_PROCESSOR_EXECUTOR_POOL, + PIPE_TSFILE_PARSER_EXECUTOR_POOL, PIPE_SINK_EXECUTOR_POOL, PIPE_CONSENSUS_EXECUTOR_POOL, PIPE_CONFIGNODE_EXECUTOR_POOL, diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java index 6bc02fbef2c42..274d51909c991 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java @@ -242,7 +242,9 @@ public class CommonConfig { // parser reserves pipeTsFileParserMemory bytes. private int pipeTsFileParserInFlightMaxNum = Math.max(1, Runtime.getRuntime().availableProcessors() / 2); - private int pipeTsFileParserInFlightMaxNumPerPipeRegion = 1; + // A non-positive value means following the global parser limit. Parallel parsing is still + // disabled by default for every Pipe and must be explicitly enabled in processor attributes. + private int pipeTsFileParserInFlightMaxNumPerPipeRegion = 0; // Memory for Sink batch sending (InsertNode/TsFile, choose one) // 1. InsertNode: 15MB, used for batch sending data to the downstream system @@ -924,12 +926,14 @@ public int getPipeTsFileParserInFlightMaxNumPerPipeRegion() { public void setPipeTsFileParserInFlightMaxNumPerPipeRegion( final int pipeTsFileParserInFlightMaxNumPerPipeRegion) { - final int validatedValue = Math.max(1, pipeTsFileParserInFlightMaxNumPerPipeRegion); - if (this.pipeTsFileParserInFlightMaxNumPerPipeRegion == validatedValue) { + if (this.pipeTsFileParserInFlightMaxNumPerPipeRegion + == pipeTsFileParserInFlightMaxNumPerPipeRegion) { return; } - this.pipeTsFileParserInFlightMaxNumPerPipeRegion = validatedValue; - logger.info("pipeTsFileParserInFlightMaxNumPerPipeRegion is set to {}.", validatedValue); + this.pipeTsFileParserInFlightMaxNumPerPipeRegion = pipeTsFileParserInFlightMaxNumPerPipeRegion; + logger.info( + "pipeTsFileParserInFlightMaxNumPerPipeRegion is set to {}.", + pipeTsFileParserInFlightMaxNumPerPipeRegion); } public long getPipeSinkBatchMemoryInsertNode() { diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/processor/donothing/DoNothingProcessor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/processor/donothing/DoNothingProcessor.java index 0a2dc4895bd72..0b3cacc0c6c5d 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/processor/donothing/DoNothingProcessor.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/processor/donothing/DoNothingProcessor.java @@ -30,11 +30,25 @@ import java.io.IOException; +import static org.apache.iotdb.commons.pipe.config.constant.PipeProcessorConstant.PROCESSOR_TSFILE_PARSER_PARALLELISM_DEFAULT_VALUE; +import static org.apache.iotdb.commons.pipe.config.constant.PipeProcessorConstant.PROCESSOR_TSFILE_PARSER_PARALLELISM_KEY; + public class DoNothingProcessor implements PipeProcessor { @Override - public void validate(PipeParameterValidator validator) { - // do nothing + public void validate(PipeParameterValidator validator) throws Exception { + final int parallelism = + validator + .getParameters() + .getIntOrDefault( + PROCESSOR_TSFILE_PARSER_PARALLELISM_KEY, + PROCESSOR_TSFILE_PARSER_PARALLELISM_DEFAULT_VALUE); + validator.validate( + value -> (Integer) value >= 1, + String.format( + "%s must be greater than or equal to 1, but got %s", + PROCESSOR_TSFILE_PARSER_PARALLELISM_KEY, parallelism), + parallelism); } @Override diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/connection/BlockingPendingQueue.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/connection/BlockingPendingQueue.java index c7b91f36d222b..3506c0387bda0 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/connection/BlockingPendingQueue.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/connection/BlockingPendingQueue.java @@ -28,6 +28,9 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.ArrayList; +import java.util.IdentityHashMap; +import java.util.Map; import java.util.Set; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentHashMap; @@ -47,6 +50,13 @@ public abstract class BlockingPendingQueue { protected final AtomicBoolean isClosed = new AtomicBoolean(false); + private final Object pendingEventMemoryLock = new Object(); + private final Map eventToMemoryReservation = + new IdentityHashMap<>(); + private final Map activeMemoryReservations = + new IdentityHashMap<>(); + private long pendingEventMemoryUsageInBytes; + protected final Set droppedPipeTaskKeys = ConcurrentHashMap.newKeySet(); protected BlockingPendingQueue( @@ -82,9 +92,78 @@ public boolean put(final E event) { } } + /** + * Waits until the caller can reserve bytes for one event. An event larger than the limit is + * admitted only when no other byte-accounted event is pending, which guarantees progress without + * turning the limit into an unbounded overshoot. + */ + public PendingEventMemoryReservation waitForMemoryReservation( + final long eventMemoryInBytes, final long maxPendingEventMemoryInBytes) { + final long normalizedEventMemoryInBytes = Math.max(0, eventMemoryInBytes); + final long normalizedMaxPendingEventMemoryInBytes = Math.max(1, maxPendingEventMemoryInBytes); + + synchronized (pendingEventMemoryLock) { + while (!isClosed.get() + && pendingEventMemoryUsageInBytes > 0 + && normalizedEventMemoryInBytes + > normalizedMaxPendingEventMemoryInBytes - pendingEventMemoryUsageInBytes) { + try { + pendingEventMemoryLock.wait(); + } catch (final InterruptedException e) { + LOGGER.info("Pending queue memory reservation is interrupted.", e); + Thread.currentThread().interrupt(); + return null; + } + } + + if (isClosed.get()) { + return null; + } + + final PendingEventMemoryReservation reservation = + new PendingEventMemoryReservation(this, normalizedEventMemoryInBytes); + activeMemoryReservations.put(reservation, Boolean.TRUE); + pendingEventMemoryUsageInBytes += normalizedEventMemoryInBytes; + return reservation; + } + } + + /** Publishes an event using bytes previously reserved by {@link #waitForMemoryReservation}. */ + public boolean offer(final E event, final PendingEventMemoryReservation reservation) { + if (reservation == null || reservation.owner != this) { + throw new IllegalArgumentException("The memory reservation does not belong to this queue."); + } + + synchronized (pendingEventMemoryLock) { + if (!checkBeforeOffer(event)) { + releaseMemoryReservationInternal(reservation); + return false; + } + if (reservation.released || reservation.published) { + throw new IllegalStateException("The memory reservation is no longer publishable."); + } + if (eventToMemoryReservation.containsKey(event)) { + releaseMemoryReservationInternal(reservation); + throw new IllegalStateException("The same event is already byte-accounted in the queue."); + } + + final boolean offered = pendingQueue.offer(event); + if (!offered) { + releaseMemoryReservationInternal(reservation); + return false; + } + + reservation.published = true; + reservation.event = event; + eventToMemoryReservation.put(event, reservation); + eventCounter.increaseEventCount(event); + return true; + } + } + public E directPoll() { final E event = pendingQueue.poll(); - eventCounter.decreaseEventCount(event); + onEventPolled(event); return event; } @@ -95,7 +174,7 @@ public E waitedPoll() { pendingQueue.poll( PIPE_CONFIG.getPipeSubtaskExecutorPendingQueueMaxBlockingTimeMs(), TimeUnit.MILLISECONDS); - eventCounter.decreaseEventCount(event); + onEventPolled(event); } catch (final InterruptedException e) { LOGGER.info("pending queue poll is interrupted.", e); Thread.currentThread().interrupt(); @@ -108,9 +187,10 @@ public E peek() { } public void clear() { - isClosed.set(true); + closeOffers(); pendingQueue.clear(); eventCounter.reset(); + releaseAllMemoryReservations(); droppedPipeTaskKeys.clear(); } @@ -120,7 +200,8 @@ public void forEach(final Consumer action) { } public void discardAllEvents() { - isClosed.set(true); + closeOffers(); + final ArrayList discardedEvents = new ArrayList<>(); pendingQueue.removeIf( event -> { if (event instanceof EnrichedEvent) { @@ -128,9 +209,12 @@ public void discardAllEvents() { eventCounter.decreaseEventCount(event); } } + discardedEvents.add(event); return true; }); + discardedEvents.forEach(this::releasePendingEventMemory); eventCounter.reset(); + releaseAllMemoryReservations(); droppedPipeTaskKeys.clear(); } @@ -141,6 +225,7 @@ public void discardEventsOfPipe( public void discardEventsOfPipe(final CommitterKey committerKey) { droppedPipeTaskKeys.add(committerKey); + final ArrayList discardedEvents = new ArrayList<>(); pendingQueue.removeIf( event -> { if (event instanceof EnrichedEvent @@ -148,10 +233,12 @@ && isEventFromPipe((EnrichedEvent) event, committerKey)) { if (((EnrichedEvent) event).clearReferenceCount(BlockingPendingQueue.class.getName())) { eventCounter.decreaseEventCount(event); } + discardedEvents.add(event); return true; } return false; }); + discardedEvents.forEach(this::releasePendingEventMemory); } public boolean isEmpty() { @@ -174,6 +261,65 @@ public int getPipeHeartbeatEventCount() { return eventCounter.getPipeHeartbeatEventCount(); } + public long getPendingEventMemoryUsageInBytes() { + synchronized (pendingEventMemoryLock) { + return pendingEventMemoryUsageInBytes; + } + } + + protected void onEventPolled(final E event) { + eventCounter.decreaseEventCount(event); + releasePendingEventMemory(event); + } + + private void releasePendingEventMemory(final E event) { + if (event == null) { + return; + } + synchronized (pendingEventMemoryLock) { + final PendingEventMemoryReservation reservation = eventToMemoryReservation.remove(event); + if (reservation != null) { + releaseMemoryReservationInternal(reservation); + } + } + } + + private void releaseAllMemoryReservations() { + synchronized (pendingEventMemoryLock) { + for (final PendingEventMemoryReservation reservation : + new ArrayList<>(activeMemoryReservations.keySet())) { + releaseMemoryReservationInternal(reservation); + } + eventToMemoryReservation.clear(); + pendingEventMemoryLock.notifyAll(); + } + } + + private void releaseMemoryReservation(final PendingEventMemoryReservation reservation) { + synchronized (pendingEventMemoryLock) { + releaseMemoryReservationInternal(reservation); + } + } + + private void releaseMemoryReservationInternal(final PendingEventMemoryReservation reservation) { + if (reservation.released) { + return; + } + reservation.released = true; + activeMemoryReservations.remove(reservation); + if (reservation.event != null) { + eventToMemoryReservation.remove(reservation.event); + } + pendingEventMemoryUsageInBytes -= reservation.memoryInBytes; + pendingEventMemoryLock.notifyAll(); + } + + private void closeOffers() { + synchronized (pendingEventMemoryLock) { + isClosed.set(true); + } + } + protected boolean checkBeforeOffer(final E event) { final boolean shouldReject = isClosed.get() || isEventFromDroppedPipe(event); if (shouldReject && event instanceof EnrichedEvent) { @@ -218,4 +364,24 @@ public boolean isPipeDropped(final String pipeName, final long creationTime, fin && key.getCreationTime() == creationTime && key.getRegionId() == regionId); } + + public static final class PendingEventMemoryReservation implements AutoCloseable { + + private final BlockingPendingQueue owner; + private final long memoryInBytes; + private Object event; + private boolean published; + private boolean released; + + private PendingEventMemoryReservation( + final BlockingPendingQueue owner, final long memoryInBytes) { + this.owner = owner; + this.memoryInBytes = memoryInBytes; + } + + @Override + public void close() { + owner.releaseMemoryReservation(this); + } + } } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/connection/UnboundedBlockingPendingQueue.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/connection/UnboundedBlockingPendingQueue.java index 43fa64c158ea4..262c76fda0430 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/connection/UnboundedBlockingPendingQueue.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/connection/UnboundedBlockingPendingQueue.java @@ -39,6 +39,8 @@ public E peekLast() { } public E pollLast() { - return pendingDeque.pollLast(); + final E event = pendingDeque.pollLast(); + onEventPolled(event); + return event; } } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeProcessorConstant.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeProcessorConstant.java index 22bc87b2917bf..8d62b286d2965 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeProcessorConstant.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeProcessorConstant.java @@ -25,6 +25,10 @@ public class PipeProcessorConstant { public static final String PROCESSOR_KEY = "processor"; + public static final String PROCESSOR_TSFILE_PARSER_PARALLELISM_KEY = + "processor.tsfile-parser.parallelism"; + public static final int PROCESSOR_TSFILE_PARSER_PARALLELISM_DEFAULT_VALUE = 1; + public static final String PROCESSOR_DOWN_SAMPLING_SPLIT_FILE_KEY = "processor.down-sampling.split-file"; public static final boolean PROCESSOR_DOWN_SAMPLING_SPLIT_FILE_DEFAULT_VALUE = false; diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/agent/task/connection/BlockingPendingQueueTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/agent/task/connection/BlockingPendingQueueTest.java new file mode 100644 index 0000000000000..2cc43311adafc --- /dev/null +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/agent/task/connection/BlockingPendingQueueTest.java @@ -0,0 +1,119 @@ +/* + * 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.commons.pipe.agent.task.connection; + +import org.apache.iotdb.commons.pipe.agent.task.connection.BlockingPendingQueue.PendingEventMemoryReservation; +import org.apache.iotdb.commons.pipe.metric.PipeEventCounter; +import org.apache.iotdb.pipe.api.event.Event; + +import org.junit.Assert; +import org.junit.Test; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import static org.mockito.Mockito.mock; + +public class BlockingPendingQueueTest { + + @Test + public void testMemoryReservationReleasedAfterPolling() throws Exception { + final UnboundedBlockingPendingQueue queue = + new UnboundedBlockingPendingQueue<>(mock(PipeEventCounter.class)); + final Event firstEvent = mock(Event.class); + final Event secondEvent = mock(Event.class); + + final PendingEventMemoryReservation firstReservation = queue.waitForMemoryReservation(6, 10); + Assert.assertNotNull(firstReservation); + Assert.assertTrue(queue.offer(firstEvent, firstReservation)); + Assert.assertEquals(6, queue.getPendingEventMemoryUsageInBytes()); + + final ExecutorService executor = Executors.newSingleThreadExecutor(); + final CountDownLatch waiterStarted = new CountDownLatch(1); + final Future blockedReservation = + executor.submit( + () -> { + waiterStarted.countDown(); + return queue.waitForMemoryReservation(5, 10); + }); + try { + Assert.assertTrue(waiterStarted.await(5, TimeUnit.SECONDS)); + assertStillBlocked(blockedReservation); + + Assert.assertSame(firstEvent, queue.directPoll()); + final PendingEventMemoryReservation secondReservation = + blockedReservation.get(5, TimeUnit.SECONDS); + Assert.assertNotNull(secondReservation); + Assert.assertEquals(5, queue.getPendingEventMemoryUsageInBytes()); + + Assert.assertTrue(queue.offer(secondEvent, secondReservation)); + Assert.assertSame(secondEvent, queue.pollLast()); + Assert.assertEquals(0, queue.getPendingEventMemoryUsageInBytes()); + } finally { + blockedReservation.cancel(true); + queue.discardAllEvents(); + executor.shutdownNow(); + Assert.assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + @Test + public void testClearWakesBlockedMemoryReservation() throws Exception { + final UnboundedBlockingPendingQueue queue = + new UnboundedBlockingPendingQueue<>(mock(PipeEventCounter.class)); + final PendingEventMemoryReservation firstReservation = queue.waitForMemoryReservation(8, 10); + Assert.assertNotNull(firstReservation); + + final ExecutorService executor = Executors.newSingleThreadExecutor(); + final CountDownLatch waiterStarted = new CountDownLatch(1); + final Future blockedReservation = + executor.submit( + () -> { + waiterStarted.countDown(); + return queue.waitForMemoryReservation(8, 10); + }); + try { + Assert.assertTrue(waiterStarted.await(5, TimeUnit.SECONDS)); + assertStillBlocked(blockedReservation); + + queue.clear(); + Assert.assertNull(blockedReservation.get(5, TimeUnit.SECONDS)); + Assert.assertEquals(0, queue.getPendingEventMemoryUsageInBytes()); + } finally { + firstReservation.close(); + blockedReservation.cancel(true); + executor.shutdownNow(); + Assert.assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + private static void assertStillBlocked(final Future future) throws Exception { + try { + future.get(200, TimeUnit.MILLISECONDS); + Assert.fail("Expected memory reservation to remain blocked"); + } catch (final TimeoutException expected) { + // Expected. + } + } +} From 8dec7f46fab76d8f1adbabe65af526ce53ca461d Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 17 Aug 2026 12:26:20 +0800 Subject: [PATCH 2/2] [Pipe] Fix built-in processor validation test fixture --- .../commons/pipe/plugin/builtin/BuiltinPipePluginTest.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/plugin/builtin/BuiltinPipePluginTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/plugin/builtin/BuiltinPipePluginTest.java index 95f549b5e6cca..bce6af995a150 100644 --- a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/plugin/builtin/BuiltinPipePluginTest.java +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/plugin/builtin/BuiltinPipePluginTest.java @@ -40,6 +40,8 @@ import org.junit.Assert; import org.junit.Test; +import java.util.Collections; + import static org.mockito.Mockito.mock; public class BuiltinPipePluginTest { @@ -75,7 +77,7 @@ public void testBuildInPipePlugin() { PipeProcessor processor = new DoNothingProcessor(); try { - processor.validate(mock(PipeParameterValidator.class)); + processor.validate(new PipeParameterValidator(new PipeParameters(Collections.emptyMap()))); } catch (Exception ignored) { Assert.fail(); }