This is an automated email from the ASF dual-hosted git repository.
Caideyipi pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new 22bc40ea6f3 [To dev/1.3] [Pipe] Fix processor worker starvation on
pipe stop (#18396) (#18443)
22bc40ea6f3 is described below
commit 22bc40ea6f3d77b43c020725649571e6d2a16dc3
Author: Caideyipi <[email protected]>
AuthorDate: Wed Aug 19 10:35:31 2026 +0800
[To dev/1.3] [Pipe] Fix processor worker starvation on pipe stop (#18396)
(#18443)
* [Pipe] Fix processor worker starvation on pipe stop (#18396)
* Fix processor worker starvation on pipe stop
* Add multi-pipe processor worker test
* Log long-running pipe processor event stacks
* refactor(pipe): reuse processor exception root cause
* Fix mixed line endings in dev/1.3 integration tests
---
LICENSE-binary | 2 +-
.../agent/task/connection/PipeEventCollector.java | 18 +-
.../subtask/processor/PipeProcessorSubtask.java | 81 ++++++-
.../PipeProcessorSubtaskExecutionGuard.java | 110 ++++++++++
.../processor/PipeProcessorSubtaskWorker.java | 128 ++++++++++-
.../PipeProcessorSubtaskWorkerManager.java | 15 +-
.../PipeProcessorSubtaskYieldException.java | 53 +++++
.../common/tsfile/PipeTsFileInsertionEvent.java | 103 +++++++--
.../task/PipeProcessorSubtaskExecutorTest.java | 6 +-
.../PipeProcessorSubtaskExecutionGuardTest.java | 240 +++++++++++++++++++++
.../processor/PipeProcessorSubtaskWorkerTest.java | 152 +++++++++++++
.../agent/task/execution/PipeSubtaskExecutor.java | 12 +-
.../task/subtask/PipeAbstractSinkSubtask.java | 2 +
.../pipe/agent/task/subtask/PipeSubtask.java | 17 +-
.../thrift-commons/src/main/thrift/common.thrift | 2 +-
.../src/main/thrift/confignode.thrift | 2 +-
.../src/main/thrift/datanode.thrift | 4 +-
17 files changed, 918 insertions(+), 29 deletions(-)
diff --git a/LICENSE-binary b/LICENSE-binary
index 3f412a0fc6b..6bcddb85554 100644
--- a/LICENSE-binary
+++ b/LICENSE-binary
@@ -236,7 +236,7 @@ org.eclipse.jetty:jetty-servlet:9.4.58.v20250814
org.eclipse.jetty:jetty-util:9.4.58.v20250814
com.google.code.findbugs:jsr305:3.0.2
com.librato.metrics:librato-java:2.1.0
-org.apache.thrift:libthrift:0.23.0
+org.apache.thrift:libthrift:0.23.0
io.dropwizard.metrics:metrics-core:4.2.19
io.dropwizard.metrics:metrics-jvm:3.2.2
com.librato.metrics:metrics-librato:5.1.0
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 df72ccb830d..d6710c83bd7 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
@@ -25,6 +25,8 @@ import
org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePatternOpera
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 class PipeEventCollector implements EventCollector {
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 class PipeEventCollector implements EventCollector {
} 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 @@ public class PipeEventCollector implements EventCollector {
}
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 @@ public class PipeEventCollector implements EventCollector {
}
sourceEvent.consumeTabletInsertionEventsWithRetry(
- this::collectParsedRawTableEvent,
"PipeEventCollector::parseAndCollectEvent");
+ this::collectParsedRawTableEvent,
+ "PipeEventCollector::parseAndCollectEvent",
+ processorExecutionGuard);
sourceEvent.close();
if (sourceEvent.isGeneratedByHistoricalExtractor()) {
PipeTerminateEvent.markHistoricalTsFileSplit(
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 3c4b3d55019..84485bc0c04 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
@@ -34,6 +34,7 @@ import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
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;
@@ -45,12 +46,14 @@ import
org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
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 {
@@ -68,6 +71,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.
@@ -88,6 +96,7 @@ public class PipeProcessorSubtask extends
PipeReportableSubtask {
this.inputEventSupplier = inputEventSupplier;
this.pipeProcessor = pipeProcessor;
this.outputEventCollector = outputEventCollector;
+ this.outputEventCollector.setProcessorExecutionGuard(executionGuard);
this.subtaskCreationTime = System.currentTimeMillis();
// Only register dataRegions
@@ -99,6 +108,7 @@ public class PipeProcessorSubtask extends
PipeReportableSubtask {
@Override
public void bindExecutors(
final ListeningExecutorService subtaskWorkerThreadPoolExecutor,
+ final ListeningScheduledExecutorService subtaskWorkerScheduledExecutor,
final ExecutorService ignored,
final PipeSubtaskScheduler subtaskScheduler) {
this.subtaskWorkerThreadPoolExecutor = subtaskWorkerThreadPoolExecutor;
@@ -109,19 +119,31 @@ public class PipeProcessorSubtask extends
PipeReportableSubtask {
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
@@ -133,7 +155,13 @@ public class PipeProcessorSubtask extends
PipeReportableSubtask {
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()) {
@@ -189,6 +217,9 @@ public class PipeProcessorSubtask extends
PipeReportableSubtask {
.enrichWithCommitterKeyAndCommitId((EnrichedEvent) event,
creationTime, regionId);
}
decreaseReferenceCountAndReleaseLastEvent(event, shouldReport);
+ } catch (final PipeProcessorSubtaskYieldException e) {
+ isResumingFromYield.set(true);
+ throw e;
} catch (final PipeRuntimeOutOfMemoryCriticalException e) {
recordResourceFailure(event, PipeResourceFailureType.MEMORY_TIMEOUT);
PipeLogger.log(
@@ -197,7 +228,12 @@ public class PipeProcessorSubtask extends
PipeReportableSubtask {
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) {
recordResourceFailure(event, PipeResourceFailureType.MEMORY_TIMEOUT);
PipeLogger.log(
LOGGER::info,
@@ -221,6 +257,8 @@ public class PipeProcessorSubtask extends
PipeReportableSubtask {
e.getMessage() != null ? " Message: " + e.getMessage() : "");
clearReferenceCountAndReleaseLastEvent(event);
}
+ } finally {
+ eventProcessingContext.compareAndSet(currentEventProcessingContext,
null);
}
return true;
@@ -233,6 +271,20 @@ public class PipeProcessorSubtask extends
PipeReportableSubtask {
// 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;
}
@@ -262,6 +314,29 @@ public class PipeProcessorSubtask extends
PipeReportableSubtask {
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/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuard.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuard.java
new file mode 100644
index 00000000000..a4eeab3482d
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/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();
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorker.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorker.java
index b9584d2c586..ad1548d9432 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorker.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorker.java
@@ -20,25 +20,50 @@
package org.apache.iotdb.db.pipe.agent.task.subtask.processor;
import org.apache.iotdb.commons.concurrent.WrappedRunnable;
+import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
+import org.apache.iotdb.pipe.api.event.Event;
+import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Collections;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
public class PipeProcessorSubtaskWorker extends WrappedRunnable {
private static final Logger LOGGER =
LoggerFactory.getLogger(PipeProcessorSubtaskWorker.class);
private static final int SLEEP_INTERVAL_ADJUSTMENT_ROUND_INTERVAL = 100;
+ private static final long LONG_RUNNING_EVENT_INITIAL_REPORT_DELAY_IN_NANOS =
+ TimeUnit.MINUTES.toNanos(10);
+ private static final long LONG_RUNNING_EVENT_REPORT_INTERVAL_IN_NANOS =
+ TimeUnit.MINUTES.toNanos(30);
+ private static final int MAX_EVENT_REPORT_LENGTH = 1024;
+ private static final int MAX_STACK_TRACE_DEPTH = 64;
+
private int totalRoundInAdjustmentInterval = 0;
private int workingRoundInAdjustmentInterval = 0;
private long sleepingTimeInMilliSecond = 50;
- private final Set<PipeProcessorSubtask> subtasks =
- Collections.newSetFromMap(new ConcurrentHashMap<>());
+ private final Set<PipeProcessorSubtask> subtasks;
+
+ private volatile Thread workerThread;
+ private volatile PipeProcessorSubtask currentSubtask;
+
+ private PipeProcessorSubtask.EventProcessingContext
lastReportedEventProcessingContext;
+ private long lastEventReportTimeInNanos = Long.MIN_VALUE;
+
+ public PipeProcessorSubtaskWorker() {
+ this(Collections.newSetFromMap(new ConcurrentHashMap<>()));
+ }
+
+ @VisibleForTesting
+ PipeProcessorSubtaskWorker(final Set<PipeProcessorSubtask> subtasks) {
+ this.subtasks = subtasks;
+ }
@Override
@SuppressWarnings("squid:S2189")
@@ -55,7 +80,8 @@ public class PipeProcessorSubtaskWorker extends
WrappedRunnable {
subtasks.removeIf(PipeProcessorSubtask::isClosed);
}
- private boolean runSubtasks() {
+ @VisibleForTesting
+ boolean runSubtasks() {
++totalRoundInAdjustmentInterval;
boolean canSleepBeforeNextRound = true;
@@ -65,18 +91,24 @@ public class PipeProcessorSubtaskWorker extends
WrappedRunnable {
continue;
}
+ workerThread = Thread.currentThread();
+ currentSubtask = subtask;
try {
final boolean hasAtLeastOneEventProcessed = subtask.call();
if (hasAtLeastOneEventProcessed) {
canSleepBeforeNextRound = false;
}
subtask.onSuccess(hasAtLeastOneEventProcessed);
+ } catch (final PipeProcessorSubtaskYieldException ignored) {
+ // The subtask voluntarily yields this worker without succeeding,
failing, or retrying.
} catch (final Exception e) {
if (subtask.isClosed()) {
LOGGER.warn("subtask {} is closed, ignore exception", subtask, e);
} else {
subtask.onFailure(e);
}
+ } finally {
+ currentSubtask = null;
}
}
@@ -117,4 +149,94 @@ public class PipeProcessorSubtaskWorker extends
WrappedRunnable {
public void schedule(final PipeProcessorSubtask pipeProcessorSubtask) {
subtasks.add(pipeProcessorSubtask);
}
+
+ void watchLongRunningEvent() {
+ final PipeProcessorSubtask subtask = currentSubtask;
+ final Thread thread = workerThread;
+ if (subtask == null || thread == null) {
+ return;
+ }
+
+ final PipeProcessorSubtask.EventProcessingContext context =
subtask.getEventProcessingContext();
+ final long currentTimeInNanos = System.nanoTime();
+ if (!isLongRunningEventReportDue(context, currentTimeInNanos)) {
+ return;
+ }
+
+ final StackTraceElement[] stackTrace = thread.getStackTrace();
+ // The event may finish while its stack is being captured. Do not
attribute a later event's
+ // stack to this event.
+ if (currentSubtask != subtask || subtask.getEventProcessingContext() !=
context) {
+ return;
+ }
+
+ markLongRunningEventReported(context, currentTimeInNanos);
+ LOGGER.warn(
+ "Pipe processor worker {} has been processing the same event for {}
ms. Pipe: {}, DataRegion: {}, subtask: {}, event: {}, thread state: {}.
Stack:{}",
+ thread.getName(),
+ TimeUnit.NANOSECONDS.toMillis(currentTimeInNanos -
context.getStartTimeInNanos()),
+ subtask.getPipeName(),
+ subtask.getRegionId(),
+ subtask.getTaskID(),
+ getEventReport(context.getEvent()),
+ thread.getState(),
+ formatStackTrace(stackTrace));
+ }
+
+ @VisibleForTesting
+ boolean isLongRunningEventReportDue(
+ final PipeProcessorSubtask.EventProcessingContext context, final long
currentTimeInNanos) {
+ if (context == null
+ || currentTimeInNanos - context.getStartTimeInNanos()
+ < LONG_RUNNING_EVENT_INITIAL_REPORT_DELAY_IN_NANOS) {
+ return false;
+ }
+
+ return lastReportedEventProcessingContext != context
+ || currentTimeInNanos - lastEventReportTimeInNanos
+ >= LONG_RUNNING_EVENT_REPORT_INTERVAL_IN_NANOS;
+ }
+
+ @VisibleForTesting
+ void markLongRunningEventReported(
+ final PipeProcessorSubtask.EventProcessingContext context, final long
currentTimeInNanos) {
+ lastReportedEventProcessingContext = context;
+ lastEventReportTimeInNanos = currentTimeInNanos;
+ }
+
+ @VisibleForTesting
+ static String getEventReport(final Event event) {
+ String report = event.getClass().getName();
+ if (event instanceof EnrichedEvent) {
+ try {
+ report =
+ event.getClass().getSimpleName() + ": " + ((EnrichedEvent)
event).coreReportMessage();
+ } catch (final RuntimeException ignored) {
+ // Keep the event class name if its diagnostic method fails.
+ }
+ }
+
+ report = report.replace('\n', ' ').replace('\r', ' ');
+ return report.length() <= MAX_EVENT_REPORT_LENGTH
+ ? report
+ : report.substring(0, MAX_EVENT_REPORT_LENGTH) + "...";
+ }
+
+ @VisibleForTesting
+ static String formatStackTrace(final StackTraceElement[] stackTrace) {
+ final StringBuilder builder = new StringBuilder();
+ final int frameCount = Math.min(stackTrace.length, MAX_STACK_TRACE_DEPTH);
+ for (int i = 0; i < frameCount; ++i) {
+ builder.append('\n').append('\t').append(stackTrace[i]);
+ }
+ if (stackTrace.length > frameCount) {
+ builder
+ .append('\n')
+ .append('\t')
+ .append("... (")
+ .append(stackTrace.length - frameCount)
+ .append(')');
+ }
+ return builder.toString();
+ }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerManager.java
index 33d58c4b5d4..ac2dd2cd7b5 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerManager.java
@@ -19,10 +19,13 @@
package org.apache.iotdb.db.pipe.agent.task.subtask.processor;
+import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.ListeningScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
public class PipeProcessorSubtaskWorkerManager {
@@ -34,7 +37,9 @@ public class PipeProcessorSubtaskWorkerManager {
private final AtomicLong scheduledTaskNumber;
- public PipeProcessorSubtaskWorkerManager(ListeningExecutorService
workerThreadPoolExecutor) {
+ public PipeProcessorSubtaskWorkerManager(
+ final ListeningExecutorService workerThreadPoolExecutor,
+ final ListeningScheduledExecutorService watcherScheduledExecutor) {
workers = new PipeProcessorSubtaskWorker[MAX_THREAD_NUM];
for (int i = 0; i < MAX_THREAD_NUM; i++) {
workers[i] = new PipeProcessorSubtaskWorker();
@@ -42,10 +47,18 @@ public class PipeProcessorSubtaskWorkerManager {
}
scheduledTaskNumber = new AtomicLong(0);
+ ScheduledExecutorUtil.safelyScheduleWithFixedDelay(
+ watcherScheduledExecutor, this::watchLongRunningEvents, 1, 1,
TimeUnit.MINUTES);
}
public void schedule(PipeProcessorSubtask pipeProcessorSubtask) {
workers[(int) (scheduledTaskNumber.getAndIncrement() %
MAX_THREAD_NUM)].schedule(
pipeProcessorSubtask);
}
+
+ private void watchLongRunningEvents() {
+ for (final PipeProcessorSubtaskWorker worker : workers) {
+ worker.watchLongRunningEvent();
+ }
+ }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskYieldException.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskYieldException.java
new file mode 100644
index 00000000000..3fe5242fa07
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskYieldException.java
@@ -0,0 +1,53 @@
+/*
+ * 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;
+
+/** Internal control-flow exception that immediately yields the current
processor worker. */
+public final class PipeProcessorSubtaskYieldException extends RuntimeException
{
+
+ private static final PipeProcessorSubtaskYieldException
PAUSE_REQUESTED_INSTANCE =
+ new PipeProcessorSubtaskYieldException(Reason.PAUSE_REQUESTED);
+ private static final PipeProcessorSubtaskYieldException
PARSER_NOT_ADMITTED_INSTANCE =
+ new PipeProcessorSubtaskYieldException(Reason.PARSER_NOT_ADMITTED);
+
+ private final Reason reason;
+
+ private PipeProcessorSubtaskYieldException(final Reason reason) {
+ super(null, null, false, false);
+ this.reason = reason;
+ }
+
+ public static PipeProcessorSubtaskYieldException pauseRequested() {
+ return PAUSE_REQUESTED_INSTANCE;
+ }
+
+ public static PipeProcessorSubtaskYieldException parserNotAdmitted() {
+ return PARSER_NOT_ADMITTED_INSTANCE;
+ }
+
+ public Reason getReason() {
+ return reason;
+ }
+
+ public enum Reason {
+ PAUSE_REQUESTED,
+ PARSER_NOT_ADMITTED
+ }
+}
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 19c845d5015..ae3708fde58 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
@@ -28,6 +28,8 @@ import
org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
import
org.apache.iotdb.commons.pipe.resource.ref.PipePhantomReferenceManager.PipeEventResource;
+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.ReferenceTrackableEvent;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import
org.apache.iotdb.db.pipe.event.common.tsfile.container.TsFileInsertionDataContainer;
@@ -247,6 +249,13 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent
* otherwise.
*/
public boolean waitForTsFileClose() throws InterruptedException {
+ return waitForTsFileClose(PipeProcessorSubtaskExecutionGuard.disabled());
+ }
+
+ public boolean waitForTsFileClose(
+ final PipeProcessorSubtaskExecutionGuard processorExecutionGuard)
+ throws InterruptedException {
+ processorExecutionGuard.check();
if (Objects.isNull(resource)) {
return true;
}
@@ -260,7 +269,9 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent
synchronized (isClosed) {
while (!isClosed.get()) {
+ processorExecutionGuard.check();
isClosed.wait(100);
+ processorExecutionGuard.check();
final boolean isClosedNow = resource.isClosed();
if (isClosedNow) {
@@ -527,19 +538,41 @@ public class PipeTsFileInsertionEvent extends
EnrichedEvent
public void consumeTabletInsertionEventsWithRetry(
final TabletInsertionEventConsumer consumer, final String callerName)
throws Exception {
+ consumeTabletInsertionEventsWithRetry(
+ consumer, callerName, PipeProcessorSubtaskExecutionGuard.disabled());
+ }
+
+ public void consumeTabletInsertionEventsWithRetry(
+ final TabletInsertionEventConsumer consumer,
+ final String callerName,
+ final PipeProcessorSubtaskExecutionGuard processorExecutionGuard)
+ throws Exception {
try {
while (true) {
+ processorExecutionGuard.check();
final PipeRawTabletInsertionEvent parsedEvent =
- getNextTabletInsertionEventFromSavedProgress();
+
getNextTabletInsertionEventFromSavedProgress(processorExecutionGuard);
if (parsedEvent == null) {
isTsFileParsingCompleted.set(true);
releaseTsFileParserMemoryIfReserved();
return;
}
+ processorExecutionGuard.check();
consumeParsedTabletInsertionEventWithRetry(
- consumer, callerName, parsedTabletInsertionEventCount.get(),
parsedEvent);
+ consumer,
+ callerName,
+ parsedTabletInsertionEventCount.get(),
+ parsedEvent,
+ processorExecutionGuard);
pendingTabletInsertionEvent.compareAndSet(parsedEvent, null);
+ processorExecutionGuard.check();
}
+ } catch (final PipeProcessorSubtaskYieldException e) {
+ releaseTsFileParserMemoryIfReserved();
+ if (!processorExecutionGuard.isCurrentInvocationValid()) {
+ cancelTsFileParserMemoryReservationIfPending();
+ }
+ throw e;
} catch (final PipeRuntimeOutOfMemoryCriticalException e) {
// Yield the active parser slot to the next pipe while retaining the
iterator and current
// tablet. The next retry resumes from this exact tablet instead of
reparsing the TsFile.
@@ -557,16 +590,15 @@ public class PipeTsFileInsertionEvent extends
EnrichedEvent
}
}
- private PipeRawTabletInsertionEvent
getNextTabletInsertionEventFromSavedProgress()
- throws Exception {
+ private PipeRawTabletInsertionEvent
getNextTabletInsertionEventFromSavedProgress(
+ final PipeProcessorSubtaskExecutionGuard processorExecutionGuard) throws
Exception {
if (isTsFileParsingCompleted.get()) {
return null;
}
- // Reacquire parser memory after a previous failure yielded the active
parser slot. This wait
- // is already bounded to 20-40 seconds, while the exponential backoff
below is only for retrying
- // the current tablet without yielding its parser slot.
- waitForResourceEnough4Parsing((long) ((1 + Math.random()) * 20 * 1000));
+ // Reacquire parser memory after a previous failure yielded the active
parser slot. Processor
+ // subtasks use non-blocking admission here, while other callers retain
the bounded wait.
+ reserveResource4Parsing(processorExecutionGuard);
final PipeRawTabletInsertionEvent pendingEvent =
pendingTabletInsertionEvent.get();
if (pendingEvent != null) {
@@ -575,7 +607,7 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent
Iterator<TabletInsertionEvent> iterator =
tabletInsertionEventIterator.get();
if (iterator == null) {
- if (!waitForTsFileClose()) {
+ if (!waitForTsFileClose(processorExecutionGuard)) {
LOGGER.warn(
"Pipe skipping temporary TsFile's parsing which shouldn't be
transferred: {}", tsFile);
return null;
@@ -598,12 +630,14 @@ public class PipeTsFileInsertionEvent extends
EnrichedEvent
final TabletInsertionEventConsumer consumer,
final String callerName,
final int tabletEventCount,
- final TabletInsertionEvent parsedEvent)
+ final TabletInsertionEvent parsedEvent,
+ final PipeProcessorSubtaskExecutionGuard processorExecutionGuard)
throws Exception {
final PipeMemoryManager memoryManager =
PipeDataNodeResourceManager.memory();
long firstOutOfMemoryTimeInMs = Long.MIN_VALUE;
int retryCount = 0;
while (true) {
+ processorExecutionGuard.check();
try {
consumer.consume((PipeRawTabletInsertionEvent) parsedEvent);
return;
@@ -617,7 +651,7 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent
}
logParserRetryOnOutOfMemory(callerName, tabletEventCount, retryCount,
e);
try {
- Thread.sleep(getParserRetryBackoffInMs(retryCount));
+ sleepForParserRetry(getParserRetryBackoffInMs(retryCount),
processorExecutionGuard);
} catch (final InterruptedException interruptedException) {
Thread.currentThread().interrupt();
throw e;
@@ -626,6 +660,24 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent
}
}
+ private void sleepForParserRetry(
+ final long sleepTimeInMs, final PipeProcessorSubtaskExecutionGuard
processorExecutionGuard)
+ throws InterruptedException {
+ if (!processorExecutionGuard.isEnabled()) {
+ Thread.sleep(sleepTimeInMs);
+ return;
+ }
+
+ final long deadlineInMs = System.currentTimeMillis() + sleepTimeInMs;
+ long remainingTimeInMs = sleepTimeInMs;
+ while (remainingTimeInMs > 0) {
+ processorExecutionGuard.check();
+ Thread.sleep(Math.min(remainingTimeInMs, 100));
+ processorExecutionGuard.check();
+ remainingTimeInMs = deadlineInMs - System.currentTimeMillis();
+ }
+ }
+
private long getParserRetryBackoffInMs(final int retryCount) {
final long initialBackoffInMs =
Math.max(1,
PipeConfig.getInstance().getPipeMemoryAllocateRetryIntervalInMs());
@@ -706,6 +758,33 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent
}
}
+ private void reserveResource4Parsing(
+ final PipeProcessorSubtaskExecutionGuard processorExecutionGuard)
+ throws InterruptedException {
+ if (!processorExecutionGuard.isEnabled()) {
+ waitForResourceEnough4Parsing((long) ((1 + Math.random()) * 20 * 1000));
+ return;
+ }
+
+ processorExecutionGuard.check();
+ final PipeMemoryManager memoryManager =
PipeDataNodeResourceManager.memory();
+ if (tryReserveTsFileParserMemory(memoryManager)) {
+ try {
+ processorExecutionGuard.check();
+ return;
+ } catch (final PipeProcessorSubtaskYieldException e) {
+ releaseTsFileParserMemoryIfReserved();
+ throw e;
+ }
+ }
+
+ if (!processorExecutionGuard.isCurrentInvocationValid()) {
+ cancelTsFileParserMemoryReservationIfPending();
+ processorExecutionGuard.check();
+ }
+ processorExecutionGuard.yieldIfParserNotAdmitted();
+ }
+
private void waitForResourceEnough4Parsing(final long timeoutMs) throws
InterruptedException {
final PipeMemoryManager memoryManager =
PipeDataNodeResourceManager.memory();
if (tryReserveTsFileParserMemory(memoryManager)) {
@@ -780,7 +859,7 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent
}
}
- private void cancelTsFileParserMemoryReservationIfPending() {
+ public void cancelTsFileParserMemoryReservationIfPending() {
if (!isTsFileParserMemoryReserved.get()) {
PipeDataNodeResourceManager.memory()
.cancelTsFileParserMemoryReservation(
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 267da1356c8..bd6e00191e5 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
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.pipe.agent.task;
import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException;
import org.apache.iotdb.commons.pipe.agent.task.connection.EventSupplier;
+import org.apache.iotdb.commons.pipe.agent.task.execution.PipeSubtaskScheduler;
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;
@@ -143,7 +144,10 @@ public class PipeProcessorSubtaskExecutorTest extends
PipeSubtaskExecutorTest {
}
private boolean executeOnceForTest() throws Exception {
- return executeOnce();
+ subtaskScheduler = mock(PipeSubtaskScheduler.class);
+ when(subtaskScheduler.schedule()).thenReturn(true, false);
+ allowSubmittingSelf();
+ return call();
}
}
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuardTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuardTest.java
new file mode 100644
index 00000000000..b7c5fdc5542
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuardTest.java
@@ -0,0 +1,240 @@
+/*
+ * 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 org.apache.iotdb.commons.conf.CommonConfig;
+import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixPipePattern;
+import org.apache.iotdb.commons.utils.FileUtils;
+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.PipeMemoryManager;
+import
org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager.TsFileParserMemoryReservation;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import
org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
+
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.file.metadata.PlainDeviceID;
+import org.apache.tsfile.utils.TsFileGeneratorUtils;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.File;
+import java.nio.file.Files;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+public class PipeProcessorSubtaskExecutionGuardTest {
+
+ @Test
+ public void testStopAndImmediateRestartInvalidateCurrentInvocation() {
+ final PipeProcessorSubtaskExecutionGuard executionGuard =
+ new PipeProcessorSubtaskExecutionGuard();
+
+ executionGuard.start();
+ executionGuard.enter();
+ executionGuard.check();
+
+ executionGuard.stop();
+ executionGuard.start();
+ Assert.assertThrows(PipeProcessorSubtaskYieldException.class,
executionGuard::check);
+
+ executionGuard.exit();
+ executionGuard.enter();
+ executionGuard.check();
+ executionGuard.exit();
+ }
+
+ @Test(timeout = 60000)
+ public void testParserAdmissionYieldsWithoutBlockingAndResumes() throws
Exception {
+ final CommonConfig commonConfig =
CommonDescriptor.getInstance().getConfig();
+ final PipeMemoryManager memoryManager =
PipeDataNodeResourceManager.memory();
+ final long originalParserMemoryInBytes =
commonConfig.getPipeTsFileParserMemory();
+ final int originalGlobalLimit =
commonConfig.getPipeTsFileParserInFlightMaxNum();
+ final int originalPerPipeRegionLimit =
+ commonConfig.getPipeTsFileParserInFlightMaxNumPerPipeRegion();
+ final TsFileParserMemoryReservation blockerReservation = new
TsFileParserMemoryReservation();
+ final TsFileParserMemoryReservation competitorReservation = new
TsFileParserMemoryReservation();
+
+ final File tempDir =
Files.createTempDirectory("pipeProcessorAdmissionYield").toFile();
+ PipeTsFileInsertionEvent event = null;
+ boolean isBlockerReserved = false;
+ boolean isCompetitorReserved = false;
+ try {
+ commonConfig.setPipeTsFileParserMemory(1);
+ commonConfig.setPipeTsFileParserInFlightMaxNum(1);
+ commonConfig.setPipeTsFileParserInFlightMaxNumPerPipeRegion(1);
+ isBlockerReserved =
+ memoryManager.tryReserveTsFileParserMemory("blocker", 0, "0",
blockerReservation);
+ Assert.assertTrue(isBlockerReserved);
+
+ event = createEvent(tempDir, "admission.tsfile", "admissionPipe");
+ final PipeTsFileInsertionEvent eventToConsume = event;
+ final PipeProcessorSubtaskExecutionGuard executionGuard =
+ new PipeProcessorSubtaskExecutionGuard();
+ executionGuard.start();
+ executionGuard.enter();
+
+ final long startTimeInNanos = System.nanoTime();
+ final PipeProcessorSubtaskYieldException admissionYield =
+ Assert.assertThrows(
+ PipeProcessorSubtaskYieldException.class,
+ () ->
+ eventToConsume.consumeTabletInsertionEventsWithRetry(
+ parsedEvent ->
parsedEvent.clearReferenceCount(getClass().getName()),
+ "test",
+ executionGuard));
+ Assert.assertEquals(
+ PipeProcessorSubtaskYieldException.Reason.PARSER_NOT_ADMITTED,
+ admissionYield.getReason());
+ Assert.assertTrue(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
startTimeInNanos) < 1000);
+ executionGuard.exit();
+
+ executionGuard.stop();
+ event.cancelTsFileParserMemoryReservationIfPending();
+ memoryManager.releaseTsFileParserMemory("blocker", 0, "0");
+ isBlockerReserved = false;
+ isCompetitorReserved =
+ memoryManager.tryReserveTsFileParserMemory("competitor", 0, "0",
competitorReservation);
+ Assert.assertTrue(isCompetitorReserved);
+ memoryManager.releaseTsFileParserMemory("competitor", 0, "0");
+ isCompetitorReserved = false;
+
+ final AtomicInteger consumedTabletCount = new AtomicInteger(0);
+ executionGuard.start();
+ executionGuard.enter();
+ event.consumeTabletInsertionEventsWithRetry(
+ parsedEvent -> {
+ consumedTabletCount.incrementAndGet();
+ parsedEvent.clearReferenceCount(getClass().getName());
+ },
+ "test",
+ executionGuard);
+ executionGuard.exit();
+ Assert.assertTrue(consumedTabletCount.get() > 0);
+ } finally {
+ if (event != null) {
+ event.close();
+ }
+ memoryManager.cancelTsFileParserMemoryReservation("blocker", 0, "0",
blockerReservation);
+ memoryManager.cancelTsFileParserMemoryReservation(
+ "competitor", 0, "0", competitorReservation);
+ if (isBlockerReserved) {
+ memoryManager.releaseTsFileParserMemory("blocker", 0, "0");
+ }
+ if (isCompetitorReserved) {
+ memoryManager.releaseTsFileParserMemory("competitor", 0, "0");
+ }
+ commonConfig.setPipeTsFileParserMemory(originalParserMemoryInBytes);
+ commonConfig.setPipeTsFileParserInFlightMaxNum(originalGlobalLimit);
+
commonConfig.setPipeTsFileParserInFlightMaxNumPerPipeRegion(originalPerPipeRegionLimit);
+ FileUtils.deleteFileOrDirectory(tempDir);
+ }
+ }
+
+ @Test(timeout = 60000)
+ public void testPauseAfterTabletResumesWithoutDuplicateConsumption() throws
Exception {
+ final CommonConfig commonConfig =
CommonDescriptor.getInstance().getConfig();
+ final long originalParserMemoryInBytes =
commonConfig.getPipeTsFileParserMemory();
+ final int originalGlobalLimit =
commonConfig.getPipeTsFileParserInFlightMaxNum();
+ final int originalPerPipeRegionLimit =
+ commonConfig.getPipeTsFileParserInFlightMaxNumPerPipeRegion();
+ final File tempDir =
Files.createTempDirectory("pipeProcessorPauseResume").toFile();
+ final PipeTsFileInsertionEvent event = createEvent(tempDir,
"resume.tsfile", "resumePipe");
+ final PipeProcessorSubtaskExecutionGuard executionGuard =
+ new PipeProcessorSubtaskExecutionGuard();
+ final AtomicInteger consumedTabletCount = new AtomicInteger(0);
+ final AtomicReference<Object> firstTablet = new AtomicReference<>();
+
+ try {
+ commonConfig.setPipeTsFileParserMemory(1);
+ commonConfig.setPipeTsFileParserInFlightMaxNum(1);
+ commonConfig.setPipeTsFileParserInFlightMaxNumPerPipeRegion(1);
+ executionGuard.start();
+ executionGuard.enter();
+ final PipeProcessorSubtaskYieldException pauseYield =
+ Assert.assertThrows(
+ PipeProcessorSubtaskYieldException.class,
+ () ->
+ event.consumeTabletInsertionEventsWithRetry(
+ parsedEvent -> {
+ firstTablet.set(parsedEvent);
+ consumedTabletCount.incrementAndGet();
+ parsedEvent.clearReferenceCount(getClass().getName());
+ executionGuard.stop();
+ },
+ "test",
+ executionGuard));
+ Assert.assertEquals(
+ PipeProcessorSubtaskYieldException.Reason.PAUSE_REQUESTED,
pauseYield.getReason());
+ executionGuard.exit();
+
+ executionGuard.start();
+ executionGuard.enter();
+ try {
+ event.consumeTabletInsertionEventsWithRetry(
+ parsedEvent -> {
+ Assert.assertNotSame(firstTablet.get(), parsedEvent);
+ consumedTabletCount.incrementAndGet();
+ parsedEvent.clearReferenceCount(getClass().getName());
+ },
+ "test",
+ executionGuard);
+ } catch (final PipeProcessorSubtaskYieldException e) {
+ Assert.fail("Unexpected yield reason: " + e.getReason());
+ }
+ executionGuard.exit();
+
+ Assert.assertTrue(consumedTabletCount.get() > 0);
+ } finally {
+ event.close();
+ commonConfig.setPipeTsFileParserMemory(originalParserMemoryInBytes);
+ commonConfig.setPipeTsFileParserInFlightMaxNum(originalGlobalLimit);
+
commonConfig.setPipeTsFileParserInFlightMaxNumPerPipeRegion(originalPerPipeRegionLimit);
+ FileUtils.deleteFileOrDirectory(tempDir);
+ }
+ }
+
+ private PipeTsFileInsertionEvent createEvent(
+ final File tempDir, final String fileName, final String pipeName) throws
Exception {
+ final File tsFile =
+ TsFileGeneratorUtils.generateNonAlignedTsFile(
+ new File(tempDir, fileName).getAbsolutePath(), 1, 1, 10, 0, 100,
10, 10);
+ final TsFileResource resource = new TsFileResource(tsFile);
+ resource.setStatusForTest(TsFileResourceStatus.NORMAL);
+ final IDeviceID deviceID = new PlainDeviceID("root.testsg.d0");
+ resource.updateStartTime(deviceID, 0);
+ resource.updateEndTime(deviceID, 9);
+
+ return new PipeTsFileInsertionEvent(
+ resource,
+ null,
+ false,
+ false,
+ false,
+ pipeName,
+ 0,
+ null,
+ new PrefixPipePattern("root"),
+ Long.MIN_VALUE,
+ Long.MAX_VALUE);
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerTest.java
new file mode 100644
index 00000000000..d1c34ccf471
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerTest.java
@@ -0,0 +1,152 @@
+/*
+ * 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 org.apache.iotdb.commons.pipe.event.EnrichedEvent;
+
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.ListeningScheduledExecutorService;
+import org.apache.commons.lang3.StringUtils;
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.InOrder;
+
+import java.util.LinkedHashSet;
+import java.util.concurrent.TimeUnit;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class PipeProcessorSubtaskWorkerTest {
+
+ @Test
+ public void testYieldingPipesDoNotBlockAnotherPipeOnSameWorker() throws
Exception {
+ final PipeProcessorSubtaskWorker worker = new
PipeProcessorSubtaskWorker(new LinkedHashSet<>());
+ final PipeProcessorSubtask stoppedPipe =
createRunnableSubtask("stoppedPipe");
+ final PipeProcessorSubtask parserWaitingPipe =
createRunnableSubtask("parserWaitingPipe");
+ final PipeProcessorSubtask runningPipe =
createRunnableSubtask("runningPipe");
+
+
when(stoppedPipe.call()).thenThrow(PipeProcessorSubtaskYieldException.pauseRequested());
+ when(parserWaitingPipe.call())
+ .thenThrow(PipeProcessorSubtaskYieldException.parserNotAdmitted());
+ when(runningPipe.call()).thenReturn(true);
+
+ worker.schedule(stoppedPipe);
+ worker.schedule(parserWaitingPipe);
+ worker.schedule(runningPipe);
+
+ Assert.assertFalse(worker.runSubtasks());
+
+ final InOrder inOrder = inOrder(stoppedPipe, parserWaitingPipe,
runningPipe);
+ inOrder.verify(stoppedPipe).call();
+ inOrder.verify(parserWaitingPipe).call();
+ inOrder.verify(runningPipe).call();
+ verify(runningPipe).onSuccess(true);
+ verify(stoppedPipe, never()).onSuccess(any());
+ verify(stoppedPipe, never()).onFailure(any());
+ verify(parserWaitingPipe, never()).onSuccess(any());
+ verify(parserWaitingPipe, never()).onFailure(any());
+ }
+
+ @Test
+ public void testLongRunningEventReportIsRateLimited() {
+ final PipeProcessorSubtaskWorker worker = new
PipeProcessorSubtaskWorker(new LinkedHashSet<>());
+ final long startTimeInNanos = 100;
+ final PipeProcessorSubtask.EventProcessingContext context =
+ new PipeProcessorSubtask.EventProcessingContext(
+ mock(EnrichedEvent.class), startTimeInNanos);
+ final long initialReportDelayInNanos = TimeUnit.MINUTES.toNanos(10);
+ final long reportIntervalInNanos = TimeUnit.MINUTES.toNanos(30);
+
+ Assert.assertFalse(
+ worker.isLongRunningEventReportDue(
+ context, startTimeInNanos + initialReportDelayInNanos - 1));
+ Assert.assertTrue(
+ worker.isLongRunningEventReportDue(context, startTimeInNanos +
initialReportDelayInNanos));
+
+ final long firstReportTimeInNanos = startTimeInNanos +
initialReportDelayInNanos;
+ worker.markLongRunningEventReported(context, firstReportTimeInNanos);
+ Assert.assertFalse(
+ worker.isLongRunningEventReportDue(
+ context, firstReportTimeInNanos + reportIntervalInNanos - 1));
+ Assert.assertTrue(
+ worker.isLongRunningEventReportDue(
+ context, firstReportTimeInNanos + reportIntervalInNanos));
+
+ final long nextEventStartTimeInNanos = firstReportTimeInNanos + 1;
+ final PipeProcessorSubtask.EventProcessingContext nextContext =
+ new PipeProcessorSubtask.EventProcessingContext(
+ mock(EnrichedEvent.class), nextEventStartTimeInNanos);
+ Assert.assertFalse(
+ worker.isLongRunningEventReportDue(
+ nextContext, nextEventStartTimeInNanos + initialReportDelayInNanos
- 1));
+ Assert.assertTrue(
+ worker.isLongRunningEventReportDue(
+ nextContext, nextEventStartTimeInNanos +
initialReportDelayInNanos));
+ }
+
+ @Test
+ public void testLongRunningEventLogPayloadIsBounded() {
+ final EnrichedEvent event = mock(EnrichedEvent.class);
+ when(event.coreReportMessage()).thenReturn(StringUtils.repeat('x', 2048) +
"\nmore");
+
+ final String eventReport =
PipeProcessorSubtaskWorker.getEventReport(event);
+ Assert.assertEquals(1027, eventReport.length());
+ Assert.assertFalse(eventReport.contains("\n"));
+ Assert.assertTrue(eventReport.endsWith("..."));
+
+ final StackTraceElement[] stackTrace = new StackTraceElement[100];
+ for (int i = 0; i < stackTrace.length; ++i) {
+ stackTrace[i] = new StackTraceElement("Class", "method" + i,
"File.java", i);
+ }
+ final String formattedStackTrace =
PipeProcessorSubtaskWorker.formatStackTrace(stackTrace);
+ Assert.assertTrue(formattedStackTrace.contains("method63"));
+ Assert.assertFalse(formattedStackTrace.contains("method64"));
+ Assert.assertTrue(formattedStackTrace.contains("... (36)"));
+ }
+
+ @Test
+ @SuppressWarnings("unsafeThreadSchedule")
+ public void testWorkerManagerSchedulesWatcher() {
+ final ListeningExecutorService workerThreadPoolExecutor =
mock(ListeningExecutorService.class);
+ final ListeningScheduledExecutorService watcherScheduledExecutor =
+ mock(ListeningScheduledExecutorService.class);
+
+ new PipeProcessorSubtaskWorkerManager(workerThreadPoolExecutor,
watcherScheduledExecutor);
+
+ verify(workerThreadPoolExecutor,
atLeastOnce()).submit(any(Runnable.class));
+ verify(watcherScheduledExecutor)
+ .scheduleWithFixedDelay(any(Runnable.class), eq(1L), eq(1L),
eq(TimeUnit.MINUTES));
+ }
+
+ private PipeProcessorSubtask createRunnableSubtask(final String mockName) {
+ final PipeProcessorSubtask subtask = mock(PipeProcessorSubtask.class,
mockName);
+ when(subtask.isClosed()).thenReturn(false);
+ when(subtask.isSubmittingSelf()).thenReturn(true);
+ when(subtask.isStoppedByException()).thenReturn(false);
+ return subtask;
+ }
+}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/execution/PipeSubtaskExecutor.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/execution/PipeSubtaskExecutor.java
index f16c6387cf5..61b821bba7f 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/execution/PipeSubtaskExecutor.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/execution/PipeSubtaskExecutor.java
@@ -26,6 +26,7 @@ import
org.apache.iotdb.commons.pipe.agent.task.subtask.PipeSubtask;
import org.apache.iotdb.commons.utils.TestOnly;
import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.ListeningScheduledExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -36,6 +37,7 @@ import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
public abstract class PipeSubtaskExecutor {
@@ -50,6 +52,7 @@ public abstract class PipeSubtaskExecutor {
protected final WrappedThreadPoolExecutor underlyingThreadPool;
protected final ListeningExecutorService subtaskWorkerThreadPoolExecutor;
+ protected final ListeningScheduledExecutorService
subtaskWorkerScheduledExecutor;
private final Map<String, PipeSubtask> registeredIdSubtaskMapper;
@@ -82,6 +85,9 @@ public abstract class PipeSubtaskExecutor {
underlyingThreadPool.disableErrorLog();
}
subtaskWorkerThreadPoolExecutor =
MoreExecutors.listeningDecorator(underlyingThreadPool);
+ final ScheduledExecutorService underlyingScheduledExecutor =
+
IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor(workingThreadName +
"-Scheduler");
+ subtaskWorkerScheduledExecutor =
MoreExecutors.listeningDecorator(underlyingScheduledExecutor);
subtaskCallbackListeningExecutor =
Objects.nonNull(callbackThreadName)
? IoTDBThreadPoolFactory.newSingleThreadExecutor(
@@ -104,7 +110,10 @@ public abstract class PipeSubtaskExecutor {
registeredIdSubtaskMapper.put(subtask.getTaskID(), subtask);
subtask.bindExecutors(
- subtaskWorkerThreadPoolExecutor, subtaskCallbackListeningExecutor,
schedulerSupplier(this));
+ subtaskWorkerThreadPoolExecutor,
+ subtaskWorkerScheduledExecutor,
+ subtaskCallbackListeningExecutor,
+ schedulerSupplier(this));
}
protected PipeSubtaskScheduler schedulerSupplier(final PipeSubtaskExecutor
executor) {
@@ -179,6 +188,7 @@ public abstract class PipeSubtaskExecutor {
}
subtaskWorkerThreadPoolExecutor.shutdown();
+ subtaskWorkerScheduledExecutor.shutdown();
if (subtaskCallbackListeningExecutor !=
globalSubtaskCallbackListeningExecutor) {
subtaskCallbackListeningExecutor.shutdown();
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java
index df43914494e..2b4f807caa4 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java
@@ -36,6 +36,7 @@ import org.apache.iotdb.pipe.api.exception.PipeException;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.ListeningScheduledExecutorService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -71,6 +72,7 @@ public abstract class PipeAbstractSinkSubtask extends
PipeReportableSubtask {
@Override
public void bindExecutors(
final ListeningExecutorService subtaskWorkerThreadPoolExecutor,
+ final ListeningScheduledExecutorService ignoredScheduledExecutor,
final ExecutorService subtaskCallbackListeningExecutor,
final PipeSubtaskScheduler subtaskScheduler) {
this.subtaskWorkerThreadPoolExecutor = subtaskWorkerThreadPoolExecutor;
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeSubtask.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeSubtask.java
index 1b58d1d6178..b583276fa90 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeSubtask.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeSubtask.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.pipe.api.event.Event;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.ListeningScheduledExecutorService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -65,6 +66,7 @@ public abstract class PipeSubtask
public abstract void bindExecutors(
ListeningExecutorService subtaskWorkerThreadPoolExecutor,
+ ListeningScheduledExecutorService subtaskWorkerScheduledExecutor,
ExecutorService subtaskCallbackListeningExecutor,
PipeSubtaskScheduler subtaskScheduler);
@@ -128,9 +130,14 @@ public abstract class PipeSubtask
public void allowSubmittingSelf() {
retryCount.set(0);
+ onAllowSubmittingSelf();
shouldStopSubmittingSelf.set(false);
}
+ protected void onAllowSubmittingSelf() {
+ // Do nothing by default.
+ }
+
/**
* Set the {@link PipeSubtask#shouldStopSubmittingSelf} state from {@code
false} to {@code true},
* in order to stop submitting the {@link PipeSubtask}.
@@ -139,7 +146,15 @@ public abstract class PipeSubtask
* {@code false} to {@code true}, {@code false} otherwise
*/
public boolean disallowSubmittingSelf() {
- return !shouldStopSubmittingSelf.getAndSet(true);
+ final boolean isChanged = !shouldStopSubmittingSelf.getAndSet(true);
+ if (isChanged) {
+ onDisallowSubmittingSelf();
+ }
+ return isChanged;
+ }
+
+ protected void onDisallowSubmittingSelf() {
+ // Do nothing by default.
}
public boolean isSubmittingSelf() {
diff --git a/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift
b/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift
index 5dd66be212c..b25188a4bcf 100644
--- a/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift
+++ b/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift
@@ -201,7 +201,7 @@ struct TPipeHeartbeatResp {
2: optional list<bool> pipeCompletedList
3: optional list<i64> pipeRemainingEventCountList
4: optional list<double> pipeRemainingTimeList
- 6: optional list<map<string, i64>> pipeRecentFailureList
+ 6: optional list<map<string, i64>> pipeRecentFailureList
}
struct TLicense {
diff --git a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
index b04274945da..86b5c857992 100644
--- a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
+++ b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
@@ -729,7 +729,7 @@ struct TShowPipeInfo {
7: required string exceptionMessage
8: optional i64 remainingEventCount
9: optional double EstimatedRemainingTime
- 11: optional map<string, i64> recentFailures
+ 11: optional map<string, i64> recentFailures
}
struct TGetAllPipeInfoResp {
diff --git a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
index 0d1f60d61a8..c569b24717e 100644
--- a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
+++ b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
@@ -307,7 +307,7 @@ struct TDataNodeHeartbeatResp {
14: optional list<bool> pipeCompletedList
15: optional list<i64> pipeRemainingEventCountList
16: optional list<double> pipeRemainingTimeList
- 19: optional list<map<string, i64>> pipeRecentFailureList
+ 19: optional list<map<string, i64>> pipeRecentFailureList
}
struct TPipeHeartbeatReq {
@@ -1077,4 +1077,4 @@ service MPPDataExchangeService {
/** Empty rpc, only for connection test */
common.TSStatus testConnectionEmptyRPC()
-}
+}
