apache · Caideyipi · Aug 11, 2026
diff --git a/...node/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java b/...node/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
@@ -25,6 +25,8 @@
 import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
 import org.apache.iotdb.commons.pipe.event.ProgressReportEvent;
 import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
+import org.apache.iotdb.db.pipe.agent.task.subtask.processor.PipeProcessorSubtaskExecutionGuard;
+import org.apache.iotdb.db.pipe.agent.task.subtask.processor.PipeProcessorSubtaskYieldException;
 import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent;
 import org.apache.iotdb.db.pipe.event.common.schema.PipeSchemaRegionWritePlanEvent;
 import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
@@ -56,6 +58,9 @@ public class PipeEventCollector implements EventCollector {
 
   private final boolean skipParsing;
 
+  private PipeProcessorSubtaskExecutionGuard processorExecutionGuard =
+      PipeProcessorSubtaskExecutionGuard.disabled();
+
   private final AtomicInteger collectInvocationCount = new AtomicInteger(0);
   private boolean hasNoGeneratedEvent = true;
   private boolean isFailedToIncreaseReferenceCount = false;
@@ -73,6 +78,11 @@ public PipeEventCollector(
     this.skipParsing = skipParsing;
   }
 
+  public void setProcessorExecutionGuard(
+      final PipeProcessorSubtaskExecutionGuard processorExecutionGuard) {
+    this.processorExecutionGuard = processorExecutionGuard;
+  }
+
   @Override
   public void collect(final Event event) {
     try {
@@ -91,6 +101,8 @@ public void collect(final Event event) {
       } else if (!(event instanceof ProgressReportEvent)) {
         collectEvent(event);
       }
+    } catch (final PipeProcessorSubtaskYieldException e) {
+      throw e;
     } catch (final PipeException e) {
       throw e;
     } catch (final Exception e) {
@@ -123,7 +135,7 @@ private void parseAndCollectEvent(final PipeRawTabletInsertionEvent sourceEvent)
   }
 
   private void parseAndCollectEvent(final PipeTsFileInsertionEvent sourceEvent) throws Exception {
-    if (!sourceEvent.waitForTsFileClose()) {
+    if (!sourceEvent.waitForTsFileClose(processorExecutionGuard)) {
       LOGGER.warn(
           "Pipe skipping temporary TsFile which shouldn't be transferred: {}",
           sourceEvent.getTsFile());
@@ -140,7 +152,9 @@ private void parseAndCollectEvent(final PipeTsFileInsertionEvent sourceEvent) th
     }
 
     sourceEvent.consumeTabletInsertionEventsWithRetry(
-        this::collectParsedRawTableEvent, "PipeEventCollector::parseAndCollectEvent");
+        this::collectParsedRawTableEvent,
+        "PipeEventCollector::parseAndCollectEvent",
+        processorExecutionGuard);
     sourceEvent.close();
     if (sourceEvent.isGeneratedByHistoricalExtractor()) {
       PipeTerminateEvent.markHistoricalTsFileSplit(

diff --git a/...main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java b/...main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java
@@ -33,6 +33,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;
@@ -44,12 +45,14 @@
 import org.apache.iotdb.pipe.api.exception.PipeException;
 
 import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.ListeningScheduledExecutorService;
 import org.apache.commons.lang3.exception.ExceptionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.Objects;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
 public class PipeProcessorSubtask extends PipeReportableSubtask {
@@ -67,6 +70,11 @@ public class PipeProcessorSubtask extends PipeReportableSubtask {
   private final EventSupplier inputEventSupplier;
   private final PipeProcessor pipeProcessor;
   private final PipeEventCollector outputEventCollector;
+  private final PipeProcessorSubtaskExecutionGuard executionGuard =
+      new PipeProcessorSubtaskExecutionGuard();
+  private final AtomicBoolean isResumingFromYield = new AtomicBoolean(false);
+  private final AtomicReference<EventProcessingContext> eventProcessingContext =
+      new AtomicReference<>();
 
   // This variable is used to distinguish between old and new subtasks before and after stuck
   // restart.
@@ -87,6 +95,7 @@ public PipeProcessorSubtask(
     this.inputEventSupplier = inputEventSupplier;
     this.pipeProcessor = pipeProcessor;
     this.outputEventCollector = outputEventCollector;
+    this.outputEventCollector.setProcessorExecutionGuard(executionGuard);
     this.subtaskCreationTime = System.currentTimeMillis();
 
     // Only register dataRegions
@@ -98,6 +107,7 @@ public PipeProcessorSubtask(
   @Override
   public void bindExecutors(
       final ListeningExecutorService subtaskWorkerThreadPoolExecutor,
+      final ListeningScheduledExecutorService subtaskWorkerScheduledExecutor,
       final ExecutorService ignored,
       final PipeSubtaskScheduler subtaskScheduler) {
     this.subtaskWorkerThreadPoolExecutor = subtaskWorkerThreadPoolExecutor;
@@ -108,19 +118,31 @@ public void bindExecutors(
       synchronized (PipeProcessorSubtaskWorkerManager.class) {
         if (subtaskWorkerManager.get() == null) {
           subtaskWorkerManager.set(
-              new PipeProcessorSubtaskWorkerManager(subtaskWorkerThreadPoolExecutor));
+              new PipeProcessorSubtaskWorkerManager(
+                  subtaskWorkerThreadPoolExecutor, subtaskWorkerScheduledExecutor));
         }
       }
     }
     subtaskWorkerManager.get().schedule(this);
   }
 
+  @Override
+  public Boolean call() throws Exception {
+    executionGuard.enter();
+    try {
+      return super.call();
+    } finally {
+      executionGuard.exit();
+    }
+  }
+
   @Override
   protected boolean executeOnce() throws Exception {
     if (isClosed.get()) {
       return false;
     }
 
+    executionGuard.check();
     final Event event =
         lastEvent != null
             ? lastEvent
@@ -132,7 +154,13 @@ protected boolean executeOnce() throws Exception {
       return false;
     }
 
-    outputEventCollector.resetFlags();
+    executionGuard.check();
+    if (!isResumingFromYield.getAndSet(false)) {
+      outputEventCollector.resetFlags();
+    }
+    final EventProcessingContext currentEventProcessingContext =
+        new EventProcessingContext(event, System.nanoTime());
+    eventProcessingContext.set(currentEventProcessingContext);
     try {
       // event can be supplied after the subtask is closed, so we need to check isClosed here
       if (!isClosed.get()) {
@@ -188,14 +216,22 @@ protected boolean executeOnce() throws Exception {
             .enrichWithCommitterKeyAndCommitId((EnrichedEvent) event, creationTime, regionId);
       }
       decreaseReferenceCountAndReleaseLastEvent(event, shouldReport);
+    } catch (final PipeProcessorSubtaskYieldException e) {
+      isResumingFromYield.set(true);
+      throw e;
     } catch (final PipeRuntimeOutOfMemoryCriticalException e) {
       PipeLogger.log(
           LOGGER::info,
           "Temporarily out of memory in pipe event processing, will wait for the memory to release. Message: %s",
           e.getMessage());
       return false;
     } catch (final Exception e) {
-      if (ExceptionUtils.getRootCause(e) instanceof PipeRuntimeOutOfMemoryCriticalException) {
+      final Throwable rootCause = ExceptionUtils.getRootCause(e);
+      if (rootCause instanceof PipeProcessorSubtaskYieldException) {
+        isResumingFromYield.set(true);
+        throw (PipeProcessorSubtaskYieldException) rootCause;
+      }
+      if (rootCause instanceof PipeRuntimeOutOfMemoryCriticalException) {
         PipeLogger.log(
             LOGGER::info,
             "Temporarily out of memory in pipe event processing, will wait for the memory to release. Message: %s",
@@ -218,6 +254,8 @@ protected boolean executeOnce() throws Exception {
             e.getMessage() != null ? " Message: " + e.getMessage() : "");
         clearReferenceCountAndReleaseLastEvent(event);
       }
+    } finally {
+      eventProcessingContext.compareAndSet(currentEventProcessingContext, null);
     }
 
     return true;
@@ -230,6 +268,20 @@ public void submitSelf() {
     // and the worker will be submitted to the executor
   }
 
+  @Override
+  protected void onAllowSubmittingSelf() {
+    executionGuard.start();
+  }
+
+  @Override
+  protected void onDisallowSubmittingSelf() {
+    executionGuard.stop();
+    final Event event = lastEvent;
+    if (event instanceof PipeTsFileInsertionEvent) {
+      ((PipeTsFileInsertionEvent) event).cancelTsFileParserMemoryReservationIfPending();
+    }
+  }
+
   public boolean isStoppedByException() {
     return lastEvent instanceof EnrichedEvent && retryCount.get() > MAX_RETRY_TIMES;
   }
@@ -259,6 +311,29 @@ boolean isClosed() {
     return isClosed.get();
   }
 
+  EventProcessingContext getEventProcessingContext() {
+    return eventProcessingContext.get();
+  }
+
+  static final class EventProcessingContext {
+
+    private final Event event;
+    private final long startTimeInNanos;
+
+    EventProcessingContext(final Event event, final long startTimeInNanos) {
+      this.event = event;
+      this.startTimeInNanos = startTimeInNanos;
+    }
+
+    Event getEvent() {
+      return event;
+    }
+
+    long getStartTimeInNanos() {
+      return startTimeInNanos;
+    }
+  }
+
   @Override
   public boolean equals(final Object obj) {
     if (this == obj) {

diff --git a/...apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuard.java b/...apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuard.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.agent.task.subtask.processor;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Guards one processor subtask invocation against concurrent STOP/START operations.
+ *
+ * <p>An invocation captures the current execution epoch. STOP invalidates that epoch before START
+ * can enable a new one, so an invocation started before STOP must yield even if the pipe is started
+ * again immediately.
+ */
+public class PipeProcessorSubtaskExecutionGuard {
+
+  private static final PipeProcessorSubtaskExecutionGuard DISABLED_GUARD =
+      new PipeProcessorSubtaskExecutionGuard(false);
+
+  private final boolean enabled;
+  private final AtomicBoolean isRunning = new AtomicBoolean(false);
+  private final AtomicLong executionEpoch = new AtomicLong(0);
+  private final ThreadLocal<Long> invocationEpoch = new ThreadLocal<>();
+
+  public PipeProcessorSubtaskExecutionGuard() {
+    this(true);
+  }
+
+  private PipeProcessorSubtaskExecutionGuard(final boolean enabled) {
+    this.enabled = enabled;
+  }
+
+  public static PipeProcessorSubtaskExecutionGuard disabled() {
+    return DISABLED_GUARD;
+  }
+
+  public boolean isEnabled() {
+    return enabled;
+  }
+
+  void start() {
+    if (enabled) {
+      isRunning.set(true);
+    }
+  }
+
+  void stop() {
+    if (enabled) {
+      isRunning.set(false);
+      executionEpoch.incrementAndGet();
+    }
+  }
+
+  void enter() {
+    if (!enabled) {
+      return;
+    }
+
+    final long currentEpoch = executionEpoch.get();
+    invocationEpoch.set(currentEpoch);
+    if (!isRunning.get() || currentEpoch != executionEpoch.get()) {
+      invocationEpoch.remove();
+      throw PipeProcessorSubtaskYieldException.pauseRequested();
+    }
+  }
+
+  void exit() {
+    if (enabled) {
+      invocationEpoch.remove();
+    }
+  }
+
+  public void check() {
+    if (!isCurrentInvocationValid()) {
+      throw PipeProcessorSubtaskYieldException.pauseRequested();
+    }
+  }
+
+  public boolean isCurrentInvocationValid() {
+    if (!enabled) {
+      return true;
+    }
+
+    final Long currentInvocationEpoch = invocationEpoch.get();
+    return currentInvocationEpoch != null
+        && isRunning.get()
+        && currentInvocationEpoch == executionEpoch.get();
+  }
+
+  public void yieldIfParserNotAdmitted() {
+    throw PipeProcessorSubtaskYieldException.parserNotAdmitted();
+  }
+}