You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ji...@apache.org on 2026/08/31 09:09:02 UTC
(iotdb) branch master updated: Fix pipe sink isolation between pipes (#18549)
This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new a9f1d5258e1 Fix pipe sink isolation between pipes (#18549)
a9f1d5258e1 is described below
commit a9f1d5258e1681258dbeb4c0ff326f48830451c5
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Mon Aug 31 17:08:56 2026 +0800
Fix pipe sink isolation between pipes (#18549)
---
.../runtime/heartbeat/PipeHeartbeatParser.java | 45 ----
.../runtime/heartbeat/PipeHeartbeatParserTest.java | 42 ++++
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 4 +
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 3 +
.../agent/runtime/PipeDataNodeRuntimeAgent.java | 17 +-
.../db/pipe/agent/task/PipeDataNodeTaskAgent.java | 9 +
.../pipe/agent/task/stage/PipeTaskSinkStage.java | 7 +-
.../agent/task/subtask/sink/PipeSinkSubtask.java | 46 ++++
.../task/subtask/sink/PipeSinkSubtaskManager.java | 219 +++++++++++++----
.../metric/schema/PipeSchemaRegionSinkMetrics.java | 69 +++---
.../metric/sink/PipeDataRegionSinkMetrics.java | 271 +++++++--------------
.../pipe/agent/task/PipeDataNodeTaskAgentTest.java | 207 ++++++++++++++++
.../subtask/sink/PipeSinkSubtaskManagerTest.java | 89 +++++++
.../schema/PipeSchemaRegionSinkMetricsTest.java | 77 ++++++
.../commons/pipe/agent/task/PipeTaskAgent.java | 246 ++++++++++++++-----
.../pipe/agent/task/meta/PipeRuntimeMeta.java | 4 +-
.../commons/pipe/agent/task/meta/PipeTaskMeta.java | 4 +-
.../iotdb/commons/service/metric/enums/Tag.java | 1 +
18 files changed, 984 insertions(+), 376 deletions(-)
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java
index f702312e5dc..f8374934811 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java
@@ -24,7 +24,6 @@ import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeCriticalException;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException;
-import org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkCriticalException;
import org.apache.iotdb.commons.pipe.agent.task.PipeTaskAgent;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeRuntimeMeta;
@@ -299,50 +298,6 @@ public class PipeHeartbeatParser {
exception,
pipeName);
}
-
- if (exception instanceof PipeRuntimeSinkCriticalException) {
- pipeTaskInfo
- .get()
- .getPipeMetaList()
- .forEach(
- pipeMeta -> {
- final PipeStaticMeta affectedStaticMeta = pipeMeta.getStaticMeta();
- if (!affectedStaticMeta
- .getSinkParameters()
- .equals(pipeMetaFromCoordinator.getStaticMeta().getSinkParameters())
- || affectedStaticMeta.equals(pipeMetaFromCoordinator.getStaticMeta())) {
- return;
- }
-
- final PipeRuntimeMeta runtimeMeta = pipeMeta.getRuntimeMeta();
- if (PipeStatus.PRE_DELETE.equals(runtimeMeta.getStatus().get())) {
- return;
- }
- if (!runtimeMeta.getStatus().get().equals(PipeStatus.STOPPED)) {
- // Record the connector exception for each pipe affected
- Map<Integer, PipeRuntimeException> exceptionMap =
- runtimeMeta.getNodeId2PipeRuntimeExceptionMap();
- if (!exceptionMap.containsKey(nodeId)
- || exceptionMap.get(nodeId).getTimeStamp()
- < exception.getTimeStamp()) {
- exceptionMap.put(nodeId, exception);
- }
- runtimeMeta.getStatus().set(PipeStatus.STOPPED);
- runtimeMeta.setIsStoppedByRuntimeException(true);
-
- needWriteConsensusOnConfigNodes.set(true);
- needPushPipeMetaToDataNodes.set(false);
-
- PipeLogger.log(
- LOGGER::warn,
- exception,
- ManagerMessages
- .DETECT_PIPERUNTIMESINKCRITICALEXCEPTION_FROM_AGENT_STOP_PIPE,
- exception,
- pipeName);
- }
- });
- }
}
}
}
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParserTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParserTest.java
index c36a09a7def..b76156bb619 100644
--- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParserTest.java
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParserTest.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.common.rpc.thrift.TPipeCompletedDataRegion;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeCriticalException;
+import org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkCriticalException;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeRuntimeMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
@@ -218,6 +219,47 @@ public class PipeHeartbeatParserTest {
verify(context.procedureManager, times(1)).pipeHandleMetaChange(true, false);
}
+ @Test
+ public void testParseHeartbeatDoesNotPropagateSinkExceptionToOtherPipes() throws Exception {
+ CommonDescriptor.getInstance().getConfig().setSeperatedPipeHeartbeatEnabled(false);
+
+ final String failedPipeName = "failedPipe";
+ final String unaffectedPipeName = "unaffectedPipe";
+ final PipeTaskInfo pipeTaskInfo = new PipeTaskInfo();
+ createPipe(pipeTaskInfo, failedPipeName, PipeStatus.RUNNING);
+ createPipe(pipeTaskInfo, unaffectedPipeName, PipeStatus.RUNNING);
+
+ final PipeMeta failedPipeMeta = pipeTaskInfo.getPipeMetaByPipeName(failedPipeName);
+ final PipeRuntimeMeta failedRuntimeMeta = failedPipeMeta.getRuntimeMeta();
+ final PipeRuntimeMeta unaffectedRuntimeMeta =
+ pipeTaskInfo.getPipeMetaByPipeName(unaffectedPipeName).getRuntimeMeta();
+
+ final PipeTaskMeta agentTaskMeta =
+ new PipeTaskMeta(MinimumProgressIndex.INSTANCE, DATA_NODE_ID);
+ agentTaskMeta.trackExceptionMessage(new PipeRuntimeSinkCriticalException("sink failure", 300L));
+ final ConcurrentMap<Integer, PipeTaskMeta> agentPipeTasks = new ConcurrentHashMap<>();
+ agentPipeTasks.put(DATA_NODE_ID, agentTaskMeta);
+ final PipeHeartbeat heartbeat =
+ new PipeHeartbeat(
+ Collections.singletonList(
+ new PipeMeta(failedPipeMeta.getStaticMeta(), new PipeRuntimeMeta(agentPipeTasks))
+ .serialize()),
+ Collections.singletonList(false),
+ Collections.singletonList(0L),
+ Collections.singletonList(0D),
+ null);
+
+ final ParserTestContext context = createParserTestContext(1, pipeTaskInfo);
+ context.parser.parseHeartbeat(DATA_NODE_ID, heartbeat);
+
+ Assert.assertEquals(PipeStatus.STOPPED, failedRuntimeMeta.getStatus().get());
+ Assert.assertTrue(failedRuntimeMeta.getIsStoppedByRuntimeException());
+ Assert.assertEquals(PipeStatus.RUNNING, unaffectedRuntimeMeta.getStatus().get());
+ Assert.assertFalse(unaffectedRuntimeMeta.getIsStoppedByRuntimeException());
+ Assert.assertTrue(unaffectedRuntimeMeta.getNodeId2PipeRuntimeExceptionMap().isEmpty());
+ verify(context.procedureManager, times(1)).pipeHandleMetaChange(true, false);
+ }
+
@Test
public void testParseHeartbeatDoesNotOverwritePreDeleteStatus() throws Exception {
CommonDescriptor.getInstance().getConfig().setSeperatedPipeHeartbeatEnabled(false);
diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index a61028e25da..41f5c2831bb 100644
--- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -134,6 +134,10 @@ public final class DataNodePipeMessages {
"Failed to decrease reference count for event {} in PipeRealtimePriorityBlockingQueue";
public static final String FAILED_TO_GET_PENDINGQUEUE_NO_SUCH_SUBTASK =
"Failed to get PendingQueue. No such subtask: ";
+ public static final String
+ EXCEPTION_MULTIPLE_PIPES_MATCH_THE_REQUESTED_SINK_SUBTASK_USE_THE_PIPE_SPECIFIC_PIPESINKSUBTASKMANAGER_API_C180D94C =
+ "Multiple pipes match the requested sink subtask. Use the pipe-specific "
+ + "PipeSinkSubtaskManager API.";
public static final String FAILED_TO_GET_PIPE_INFO_FROM_CONFIG_NODE_STATUS =
"Failed to get pipe info from config node, status is %s.";
public static final String FAILED_TO_GET_PIPE_METAS_WILL_BE =
diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index 4de5ee9aa99..5a4d9c4c568 100644
--- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -124,6 +124,9 @@ public final class DataNodePipeMessages {
"减少 reference count for event {} in PipeRealtimePriorityBlockingQueue 失败";
public static final String FAILED_TO_GET_PENDINGQUEUE_NO_SUCH_SUBTASK =
"获取 PendingQueue 失败，没有该子任务: ";
+ public static final String
+ EXCEPTION_MULTIPLE_PIPES_MATCH_THE_REQUESTED_SINK_SUBTASK_USE_THE_PIPE_SPECIFIC_PIPESINKSUBTASKMANAGER_API_C180D94C =
+ "多个 Pipe 匹配请求的 sink 子任务。请使用指定 Pipe 的 PipeSinkSubtaskManager API。";
public static final String FAILED_TO_GET_PIPE_INFO_FROM_CONFIG_NODE_STATUS =
"从 CN 处获取 pipe 信息失败，状态是 %s。";
public static final String FAILED_TO_GET_PIPE_METAS_WILL_BE =
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeDataNodeRuntimeAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeDataNodeRuntimeAgent.java
index 1a589d5970d..83c115a704e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeDataNodeRuntimeAgent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeDataNodeRuntimeAgent.java
@@ -232,7 +232,11 @@ public class PipeDataNodeRuntimeAgent implements IService {
public void report(EnrichedEvent event, PipeRuntimeException pipeRuntimeException) {
if (event.getPipeTaskMeta() != null) {
- report(event.getPipeTaskMeta(), pipeRuntimeException);
+ report(
+ event.getPipeName(),
+ event.getCreationTime(),
+ event.getPipeTaskMeta(),
+ pipeRuntimeException);
} else {
PipeLogger.log(
LOGGER::warn,
@@ -242,6 +246,14 @@ public class PipeDataNodeRuntimeAgent implements IService {
}
public void report(PipeTaskMeta pipeTaskMeta, PipeRuntimeException pipeRuntimeException) {
+ report(null, Long.MIN_VALUE, pipeTaskMeta, pipeRuntimeException);
+ }
+
+ private void report(
+ final String pipeName,
+ final long creationTime,
+ final PipeTaskMeta pipeTaskMeta,
+ final PipeRuntimeException pipeRuntimeException) {
PipeLogger.log(
LOGGER::warn,
pipeRuntimeException,
@@ -253,7 +265,8 @@ public class PipeDataNodeRuntimeAgent implements IService {
// no need to wait for the next heartbeat cycle.
if (pipeRuntimeException instanceof PipeRuntimeCriticalException) {
PipeDataNodeAgent.task()
- .stopAllPipesWithCriticalExceptionAndTrackException(pipeTaskMeta, pipeRuntimeException);
+ .stopAllPipesWithCriticalExceptionAndTrackException(
+ pipeName, creationTime, pipeTaskMeta, pipeRuntimeException);
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java
index 1f6416ad20f..b0f527110d3 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java
@@ -475,6 +475,15 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
CONFIG.getDataNodeId(), pipeTaskMeta, pipeRuntimeException);
}
+ public void stopAllPipesWithCriticalExceptionAndTrackException(
+ final String pipeName,
+ final long creationTime,
+ final PipeTaskMeta pipeTaskMeta,
+ final PipeRuntimeException pipeRuntimeException) {
+ super.stopAllPipesWithCriticalException(
+ CONFIG.getDataNodeId(), pipeName, creationTime, pipeTaskMeta, pipeRuntimeException);
+ }
+
///////////////////////// Heartbeat /////////////////////////
public void collectPipeMetaList(final TDataNodeHeartbeatResp resp) throws TException {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/stage/PipeTaskSinkStage.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/stage/PipeTaskSinkStage.java
index 88eac560cde..2bbcc1248d8 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/stage/PipeTaskSinkStage.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/stage/PipeTaskSinkStage.java
@@ -71,12 +71,12 @@ public class PipeTaskSinkStage extends PipeTaskStage {
@Override
public void startSubtask() throws PipeException {
- PipeSinkSubtaskManager.instance().start(sinkSubtaskId);
+ PipeSinkSubtaskManager.instance().start(pipeName, creationTime, sinkSubtaskId);
}
@Override
public void stopSubtask() throws PipeException {
- PipeSinkSubtaskManager.instance().stop(sinkSubtaskId);
+ PipeSinkSubtaskManager.instance().stop(pipeName, creationTime, sinkSubtaskId);
}
@Override
@@ -85,6 +85,7 @@ public class PipeTaskSinkStage extends PipeTaskStage {
}
public UnboundedBlockingPendingQueue<Event> getPipeSinkPendingQueue() {
- return PipeSinkSubtaskManager.instance().getPipeSinkPendingQueue(sinkSubtaskId);
+ return PipeSinkSubtaskManager.instance()
+ .getPipeSinkPendingQueue(pipeName, creationTime, sinkSubtaskId);
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java
index 83de468cba5..e65c2d3f619 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java
@@ -68,6 +68,7 @@ public class PipeSinkSubtask extends PipeAbstractSinkSubtask {
protected final UnboundedBlockingPendingQueue<Event> inputPendingQueue;
// Record these variables to provide corresponding value to tag key of monitoring metrics
+ private final String pipeName;
private final String attributeSortedString;
private final String attributeDisplayString;
private final int sinkIndex;
@@ -88,6 +89,7 @@ public class PipeSinkSubtask extends PipeAbstractSinkSubtask {
final UnboundedBlockingPendingQueue<Event> inputPendingQueue,
final PipeConnector outputPipeConnector) {
this(
+ null,
taskID,
creationTime,
attributeSortedString,
@@ -98,6 +100,45 @@ public class PipeSinkSubtask extends PipeAbstractSinkSubtask {
}
public PipeSinkSubtask(
+ final String pipeName,
+ final String taskID,
+ final long creationTime,
+ final String attributeSortedString,
+ final int sinkIndex,
+ final UnboundedBlockingPendingQueue<Event> inputPendingQueue,
+ final PipeConnector outputPipeConnector) {
+ this(
+ pipeName,
+ taskID,
+ creationTime,
+ attributeSortedString,
+ attributeSortedString,
+ sinkIndex,
+ inputPendingQueue,
+ outputPipeConnector);
+ }
+
+ public PipeSinkSubtask(
+ final String taskID,
+ final long creationTime,
+ final String attributeSortedString,
+ final String attributeDisplayString,
+ final int sinkIndex,
+ final UnboundedBlockingPendingQueue<Event> inputPendingQueue,
+ final PipeConnector outputPipeConnector) {
+ this(
+ null,
+ taskID,
+ creationTime,
+ attributeSortedString,
+ attributeDisplayString,
+ sinkIndex,
+ inputPendingQueue,
+ outputPipeConnector);
+ }
+
+ public PipeSinkSubtask(
+ final String pipeName,
final String taskID,
final long creationTime,
final String attributeSortedString,
@@ -106,6 +147,7 @@ public class PipeSinkSubtask extends PipeAbstractSinkSubtask {
final UnboundedBlockingPendingQueue<Event> inputPendingQueue,
final PipeConnector outputPipeConnector) {
super(taskID, creationTime, outputPipeConnector);
+ this.pipeName = pipeName;
this.attributeSortedString = attributeSortedString;
this.attributeDisplayString = attributeDisplayString;
this.sinkIndex = sinkIndex;
@@ -467,6 +509,10 @@ public class PipeSinkSubtask extends PipeAbstractSinkSubtask {
//////////////////////////// APIs provided for metric framework ////////////////////////////
+ public String getPipeName() {
+ return pipeName;
+ }
+
public String getAttributeSortedString() {
return attributeSortedString;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManager.java
index 1d94d4d9419..27a38c76d52 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManager.java
@@ -31,6 +31,7 @@ import org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskSinkRuntimeEnviro
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
import org.apache.iotdb.db.pipe.agent.task.execution.PipeSinkSubtaskExecutor;
+import org.apache.iotdb.db.pipe.consensus.IoTConsensusV2SubtaskExecutor;
import org.apache.iotdb.db.pipe.consensus.ReplicateProgressDataNodeManager;
import org.apache.iotdb.db.pipe.metric.source.PipeDataRegionEventCounter;
import org.apache.iotdb.db.pipe.sink.protocol.thrift.async.IoTDBDataRegionAsyncSink;
@@ -49,6 +50,7 @@ import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.TreeMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;
@@ -60,10 +62,10 @@ public class PipeSinkSubtaskManager {
private static final String FAILED_TO_DEREGISTER_EXCEPTION_MESSAGE =
"Failed to deregister PipeConnectorSubtask. No such subtask: ";
- private final Map<String, List<PipeSinkSubtaskLifeCycle>>
- attributeSortedString2SubtaskLifeCycleMap = new HashMap<>();
+ private final Map<PipeSinkSubtaskKey, List<PipeSinkSubtaskLifeCycle>>
+ pipeSinkSubtaskKey2SubtaskLifeCycleMap = new HashMap<>();
- private final Map<String, String> attributeSortedString2DisplayString = new HashMap<>();
+ private final Map<PipeSinkSubtaskKey, String> pipeSinkSubtaskKey2DisplayString = new HashMap<>();
public synchronized String register(
final Supplier<? extends PipeSinkSubtaskExecutor> executorSupplier,
@@ -98,8 +100,11 @@ public class PipeSinkSubtaskManager {
: "data_" + attributeDisplayString
: "schema_" + attributeDisplayString;
environment.setAttributeSortedString(attributeDisplayStringWithPrefix);
+ final PipeSinkSubtaskKey pipeSinkSubtaskKey =
+ new PipeSinkSubtaskKey(
+ environment.getPipeName(), environment.getCreationTime(), attributeSortedString);
- if (!attributeSortedString2SubtaskLifeCycleMap.containsKey(attributeSortedString)) {
+ if (!pipeSinkSubtaskKey2SubtaskLifeCycleMap.containsKey(pipeSinkSubtaskKey)) {
final PipeSinkSubtaskExecutor executor = executorSupplier.get();
final List<PipeSinkSubtaskLifeCycle> pipeSinkSubtaskLifeCycleList = new ArrayList<>(sinkNum);
@@ -117,7 +122,11 @@ public class PipeSinkSubtaskManager {
for (int sinkIndex = 0; sinkIndex < sinkNum; sinkIndex++) {
final String taskID =
String.format(
- "%s_%s_%s", attributeSortedString, environment.getCreationTime(), sinkIndex);
+ "%s_%s_%s_%s",
+ environment.getPipeName(),
+ attributeSortedString,
+ environment.getCreationTime(),
+ sinkIndex);
environment.setSinkTaskId(taskID);
final PipeConnector pipeSink =
@@ -148,6 +157,7 @@ public class PipeSinkSubtaskManager {
// 2. Construct PipeConnectorSubtaskLifeCycle to manage PipeConnectorSubtask's life cycle
final PipeSinkSubtask pipeSinkSubtask =
new PipeSinkSubtask(
+ environment.getPipeName(),
taskID,
environment.getCreationTime(),
attributeSortedString,
@@ -165,14 +175,12 @@ public class PipeSinkSubtaskManager {
attributeDisplayStringWithPrefix,
executor.getWorkingThreadName(),
executor.getCallbackThreadName());
- attributeSortedString2SubtaskLifeCycleMap.put(
- attributeSortedString, pipeSinkSubtaskLifeCycleList);
- attributeSortedString2DisplayString.put(
- attributeSortedString, attributeDisplayStringWithPrefix);
+ pipeSinkSubtaskKey2SubtaskLifeCycleMap.put(pipeSinkSubtaskKey, pipeSinkSubtaskLifeCycleList);
+ pipeSinkSubtaskKey2DisplayString.put(pipeSinkSubtaskKey, attributeDisplayStringWithPrefix);
}
for (final PipeSinkSubtaskLifeCycle lifeCycle :
- attributeSortedString2SubtaskLifeCycleMap.get(attributeSortedString)) {
+ pipeSinkSubtaskKey2SubtaskLifeCycleMap.get(pipeSinkSubtaskKey)) {
lifeCycle.register();
}
@@ -184,12 +192,14 @@ public class PipeSinkSubtaskManager {
final long creationTime,
final int regionId,
final String attributeSortedString) {
- if (!attributeSortedString2SubtaskLifeCycleMap.containsKey(attributeSortedString)) {
- throwNoSuchSubtaskException(attributeSortedString);
+ final PipeSinkSubtaskKey pipeSinkSubtaskKey =
+ new PipeSinkSubtaskKey(pipeName, creationTime, attributeSortedString);
+ if (!pipeSinkSubtaskKey2SubtaskLifeCycleMap.containsKey(pipeSinkSubtaskKey)) {
+ throwNoSuchSubtaskException(pipeSinkSubtaskKey);
}
final List<PipeSinkSubtaskLifeCycle> lifeCycles =
- attributeSortedString2SubtaskLifeCycleMap.get(attributeSortedString);
+ pipeSinkSubtaskKey2SubtaskLifeCycleMap.get(pipeSinkSubtaskKey);
// Shall not be empty
final PipeSinkSubtaskExecutor executor = lifeCycles.get(0).executor;
@@ -200,13 +210,17 @@ public class PipeSinkSubtaskManager {
lifeCycles.removeIf(o -> o.deregister(committerKey));
if (lifeCycles.isEmpty()) {
- attributeSortedString2SubtaskLifeCycleMap.remove(attributeSortedString);
- attributeSortedString2DisplayString.remove(attributeSortedString);
- executor.shutdown();
- LOGGER.info(
- DataNodePipeMessages.THE_EXECUTOR_AND_HAS_BEEN_SUCCESSFULLY_SHUTDOWN,
- executor.getWorkingThreadName(),
- executor.getCallbackThreadName());
+ pipeSinkSubtaskKey2SubtaskLifeCycleMap.remove(pipeSinkSubtaskKey);
+ pipeSinkSubtaskKey2DisplayString.remove(pipeSinkSubtaskKey);
+ // The IoTConsensusV2 executor is owned by the node-wide global component container and is
+ // shared by all consensus pipes. A single pipe must not shut it down for the others.
+ if (!(executor instanceof IoTConsensusV2SubtaskExecutor)) {
+ executor.shutdown();
+ LOGGER.info(
+ DataNodePipeMessages.THE_EXECUTOR_AND_HAS_BEEN_SUCCESSFULLY_SHUTDOWN,
+ executor.getWorkingThreadName(),
+ executor.getCallbackThreadName());
+ }
}
PipeEventCommitManager.getInstance().deregister(pipeName, creationTime, regionId);
@@ -215,47 +229,122 @@ public class PipeSinkSubtaskManager {
ReplicateProgressDataNodeManager.resetReplicateIndexForIoTV2(pipeName);
}
+ public synchronized void start(
+ final String pipeName, final long creationTime, final String attributeSortedString) {
+ final PipeSinkSubtaskKey pipeSinkSubtaskKey =
+ new PipeSinkSubtaskKey(pipeName, creationTime, attributeSortedString);
+ if (!pipeSinkSubtaskKey2SubtaskLifeCycleMap.containsKey(pipeSinkSubtaskKey)) {
+ throwNoSuchSubtaskException(pipeSinkSubtaskKey);
+ }
+
+ for (final PipeSinkSubtaskLifeCycle lifeCycle :
+ pipeSinkSubtaskKey2SubtaskLifeCycleMap.get(pipeSinkSubtaskKey)) {
+ lifeCycle.start();
+ }
+ }
+
+ /**
+ * @deprecated Use {@link #start(String, long, String)} to identify the pipe explicitly.
+ */
+ @Deprecated
public synchronized void start(final String attributeSortedString) {
- if (!attributeSortedString2SubtaskLifeCycleMap.containsKey(attributeSortedString)) {
- throwNoSuchSubtaskException(attributeSortedString);
+ final PipeSinkSubtaskKey pipeSinkSubtaskKey =
+ getUniquePipeSinkSubtaskKey(attributeSortedString);
+ if (pipeSinkSubtaskKey == null) {
+ throwNoSuchSubtaskException(
+ new PipeSinkSubtaskKey(null, Long.MIN_VALUE, attributeSortedString));
}
for (final PipeSinkSubtaskLifeCycle lifeCycle :
- attributeSortedString2SubtaskLifeCycleMap.get(attributeSortedString)) {
+ pipeSinkSubtaskKey2SubtaskLifeCycleMap.get(pipeSinkSubtaskKey)) {
lifeCycle.start();
}
}
+ public synchronized void stop(
+ final String pipeName, final long creationTime, final String attributeSortedString) {
+ final PipeSinkSubtaskKey pipeSinkSubtaskKey =
+ new PipeSinkSubtaskKey(pipeName, creationTime, attributeSortedString);
+ if (!pipeSinkSubtaskKey2SubtaskLifeCycleMap.containsKey(pipeSinkSubtaskKey)) {
+ throwNoSuchSubtaskException(pipeSinkSubtaskKey);
+ }
+
+ for (final PipeSinkSubtaskLifeCycle lifeCycle :
+ pipeSinkSubtaskKey2SubtaskLifeCycleMap.get(pipeSinkSubtaskKey)) {
+ lifeCycle.stop();
+ }
+ }
+
+ /**
+ * @deprecated Use {@link #stop(String, long, String)} to identify the pipe explicitly.
+ */
+ @Deprecated
public synchronized void stop(final String attributeSortedString) {
- if (!attributeSortedString2SubtaskLifeCycleMap.containsKey(attributeSortedString)) {
- throwNoSuchSubtaskException(attributeSortedString);
+ final PipeSinkSubtaskKey pipeSinkSubtaskKey =
+ getUniquePipeSinkSubtaskKey(attributeSortedString);
+ if (pipeSinkSubtaskKey == null) {
+ throwNoSuchSubtaskException(
+ new PipeSinkSubtaskKey(null, Long.MIN_VALUE, attributeSortedString));
}
for (final PipeSinkSubtaskLifeCycle lifeCycle :
- attributeSortedString2SubtaskLifeCycleMap.get(attributeSortedString)) {
+ pipeSinkSubtaskKey2SubtaskLifeCycleMap.get(pipeSinkSubtaskKey)) {
lifeCycle.stop();
}
}
- public UnboundedBlockingPendingQueue<Event> getPipeSinkPendingQueue(
- final String attributeSortedString) {
- if (!attributeSortedString2SubtaskLifeCycleMap.containsKey(attributeSortedString)) {
+ public synchronized UnboundedBlockingPendingQueue<Event> getPipeSinkPendingQueue(
+ final String pipeName, final long creationTime, final String attributeSortedString) {
+ final PipeSinkSubtaskKey pipeSinkSubtaskKey =
+ new PipeSinkSubtaskKey(pipeName, creationTime, attributeSortedString);
+ if (!pipeSinkSubtaskKey2SubtaskLifeCycleMap.containsKey(pipeSinkSubtaskKey)) {
throw new PipeException(
DataNodePipeMessages.FAILED_TO_GET_PENDINGQUEUE_NO_SUCH_SUBTASK
- + getDisplayStringForException(attributeSortedString));
+ + getDisplayStringForException(pipeSinkSubtaskKey));
}
// All subtasks share the same pending queue
- return attributeSortedString2SubtaskLifeCycleMap
- .get(attributeSortedString)
- .get(0)
- .getPendingQueue();
+ return pipeSinkSubtaskKey2SubtaskLifeCycleMap.get(pipeSinkSubtaskKey).get(0).getPendingQueue();
+ }
+
+ /**
+ * @deprecated Use {@link #getPipeSinkPendingQueue(String, long, String)} to identify the pipe
+ * explicitly.
+ */
+ @Deprecated
+ public synchronized UnboundedBlockingPendingQueue<Event> getPipeSinkPendingQueue(
+ final String attributeSortedString) {
+ final PipeSinkSubtaskKey pipeSinkSubtaskKey =
+ getUniquePipeSinkSubtaskKey(attributeSortedString);
+ if (pipeSinkSubtaskKey == null) {
+ throw new PipeException(
+ DataNodePipeMessages.FAILED_TO_GET_PENDINGQUEUE_NO_SUCH_SUBTASK
+ + getDisplayStringForException(
+ new PipeSinkSubtaskKey(null, Long.MIN_VALUE, attributeSortedString)));
+ }
+
+ return pipeSinkSubtaskKey2SubtaskLifeCycleMap.get(pipeSinkSubtaskKey).get(0).getPendingQueue();
}
+ public synchronized boolean hasRegisteredSubtasks(
+ final String pipeName,
+ final long creationTime,
+ final PipeParameters pipeSinkParameters,
+ final int regionId) {
+ return pipeSinkSubtaskKey2SubtaskLifeCycleMap.containsKey(
+ new PipeSinkSubtaskKey(
+ pipeName, creationTime, generateAttributeSortedString(pipeSinkParameters, regionId)));
+ }
+
+ /**
+ * @deprecated Use {@link #hasRegisteredSubtasks(String, long, PipeParameters, int)} to identify
+ * the pipe explicitly.
+ */
+ @Deprecated
public synchronized boolean hasRegisteredSubtasks(
final PipeParameters pipeSinkParameters, final int regionId) {
- return attributeSortedString2SubtaskLifeCycleMap.containsKey(
- generateAttributeSortedString(pipeSinkParameters, regionId));
+ return getUniquePipeSinkSubtaskKey(generateAttributeSortedString(pipeSinkParameters, regionId))
+ != null;
}
public static int calculateSinkSubtaskNum(
@@ -324,14 +413,62 @@ public class PipeSinkSubtaskManager {
return filteredAttributes.toString();
}
- private void throwNoSuchSubtaskException(final String attributeSortedString) {
+ private void throwNoSuchSubtaskException(final PipeSinkSubtaskKey pipeSinkSubtaskKey) {
throw new PipeException(
- FAILED_TO_DEREGISTER_EXCEPTION_MESSAGE
- + getDisplayStringForException(attributeSortedString));
+ FAILED_TO_DEREGISTER_EXCEPTION_MESSAGE + getDisplayStringForException(pipeSinkSubtaskKey));
+ }
+
+ private String getDisplayStringForException(final PipeSinkSubtaskKey pipeSinkSubtaskKey) {
+ return pipeSinkSubtaskKey2DisplayString.getOrDefault(pipeSinkSubtaskKey, "unknown");
}
- private String getDisplayStringForException(final String attributeSortedString) {
- return attributeSortedString2DisplayString.getOrDefault(attributeSortedString, "unknown");
+ private PipeSinkSubtaskKey getUniquePipeSinkSubtaskKey(final String attributeSortedString) {
+ PipeSinkSubtaskKey matchedKey = null;
+ for (final PipeSinkSubtaskKey key : pipeSinkSubtaskKey2SubtaskLifeCycleMap.keySet()) {
+ if (!Objects.equals(attributeSortedString, key.attributeSortedString)) {
+ continue;
+ }
+ if (matchedKey != null) {
+ throw new PipeException(
+ DataNodePipeMessages
+ .EXCEPTION_MULTIPLE_PIPES_MATCH_THE_REQUESTED_SINK_SUBTASK_USE_THE_PIPE_SPECIFIC_PIPESINKSUBTASKMANAGER_API_C180D94C);
+ }
+ matchedKey = key;
+ }
+ return matchedKey;
+ }
+
+ private static final class PipeSinkSubtaskKey {
+
+ private final String pipeName;
+ private final long creationTime;
+ private final String attributeSortedString;
+
+ private PipeSinkSubtaskKey(
+ final String pipeName, final long creationTime, final String attributeSortedString) {
+ this.pipeName = pipeName;
+ this.creationTime = creationTime;
+ this.attributeSortedString = attributeSortedString;
+ }
+
+ @Override
+ public boolean equals(final Object object) {
+ if (this == object) {
+ return true;
+ }
+ if (!(object instanceof PipeSinkSubtaskKey)) {
+ return false;
+ }
+ final PipeSinkSubtaskKey that = (PipeSinkSubtaskKey) object;
+ return creationTime == that.creationTime
+ && Objects.equals(pipeName, that.pipeName)
+ && Objects.equals(attributeSortedString, that.attributeSortedString);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(pipeName, creationTime, attributeSortedString);
+ }
}
///////////////////////// Singleton Instance Holder /////////////////////////
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/schema/PipeSchemaRegionSinkMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/schema/PipeSchemaRegionSinkMetrics.java
index 869b2616e60..cc8c74e2984 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/schema/PipeSchemaRegionSinkMetrics.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/schema/PipeSchemaRegionSinkMetrics.java
@@ -61,6 +61,37 @@ public class PipeSchemaRegionSinkMetrics implements IMetricSet {
createHistogram(taskID);
}
+ private static String[] getCreationTimeTags(final PipeSinkSubtask connector) {
+ return connector.getPipeName() == null
+ ? new String[] {
+ Tag.NAME.toString(),
+ connector.getAttributeSortedString(),
+ Tag.CREATION_TIME.toString(),
+ String.valueOf(connector.getCreationTime())
+ }
+ : new String[] {
+ Tag.NAME.toString(),
+ connector.getAttributeSortedString(),
+ Tag.PIPE.toString(),
+ connector.getPipeName(),
+ Tag.CREATION_TIME.toString(),
+ String.valueOf(connector.getCreationTime())
+ };
+ }
+
+ private static String[] getNameTags(final PipeSinkSubtask connector) {
+ return connector.getPipeName() == null
+ ? new String[] {Tag.NAME.toString(), connector.getAttributeSortedString()}
+ : new String[] {
+ Tag.NAME.toString(),
+ connector.getAttributeSortedString(),
+ Tag.PIPE.toString(),
+ connector.getPipeName(),
+ Tag.CREATION_TIME.toString(),
+ String.valueOf(connector.getCreationTime())
+ };
+ }
+
private void createRate(final String taskID) {
final PipeSinkSubtask connector = connectorMap.get(taskID);
// Transfer event rate
@@ -69,10 +100,7 @@ public class PipeSchemaRegionSinkMetrics implements IMetricSet {
metricService.getOrCreateRate(
Metric.PIPE_CONNECTOR_SCHEMA_TRANSFER.toString(),
MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime())));
+ getCreationTimeTags(connector)));
}
private void createHistogram(final String taskID) {
@@ -82,28 +110,21 @@ public class PipeSchemaRegionSinkMetrics implements IMetricSet {
metricService.getOrCreateHistogram(
Metric.PIPE_SCHEMA_BATCH_SIZE.toString(),
MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime()));
+ getCreationTimeTags(connector));
connector.setSchemaBatchSizeHistogram(schemaBatchSizeHistogram);
final Histogram schemaBatchTimeIntervalHistogram =
metricService.getOrCreateHistogram(
Metric.PIPE_SCHEMA_BATCH_TIME_COST.toString(),
MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime()));
+ getCreationTimeTags(connector));
connector.setSchemaBatchTimeIntervalHistogram(schemaBatchTimeIntervalHistogram);
final Histogram schemaBatchEventSizeHistogram =
metricService.getOrCreateHistogram(
Metric.PIPE_CONNECTOR_BATCH_SIZE.toString(),
MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- connector.getAttributeSortedString());
+ getNameTags(connector));
connector.setEventSizeHistogram(schemaBatchEventSizeHistogram);
}
@@ -126,10 +147,7 @@ public class PipeSchemaRegionSinkMetrics implements IMetricSet {
metricService.remove(
MetricType.RATE,
Metric.PIPE_CONNECTOR_SCHEMA_TRANSFER.toString(),
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime()));
+ getCreationTimeTags(connector));
schemaRateMap.remove(taskID);
}
@@ -138,22 +156,13 @@ public class PipeSchemaRegionSinkMetrics implements IMetricSet {
metricService.remove(
MetricType.HISTOGRAM,
Metric.PIPE_SCHEMA_BATCH_SIZE.toString(),
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime()));
+ getCreationTimeTags(connector));
metricService.remove(
MetricType.HISTOGRAM,
Metric.PIPE_SCHEMA_BATCH_TIME_COST.toString(),
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime()));
+ getCreationTimeTags(connector));
metricService.remove(
- MetricType.HISTOGRAM,
- Metric.PIPE_CONNECTOR_BATCH_SIZE.toString(),
- Tag.NAME.toString(),
- connector.getAttributeSortedString());
+ MetricType.HISTOGRAM, Metric.PIPE_CONNECTOR_BATCH_SIZE.toString(), getNameTags(connector));
}
//////////////////////////// Register & deregister (pipe integration) ////////////////////////////
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/sink/PipeDataRegionSinkMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/sink/PipeDataRegionSinkMetrics.java
index e2a391e8f70..a52d32ed0d7 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/sink/PipeDataRegionSinkMetrics.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/sink/PipeDataRegionSinkMetrics.java
@@ -74,6 +74,59 @@ public class PipeDataRegionSinkMetrics implements IMetricSet {
createHistogram(taskID);
}
+ private static String[] getIndexedTags(final PipeSinkSubtask sink) {
+ return sink.getPipeName() == null
+ ? new String[] {
+ Tag.NAME.toString(),
+ sink.getAttributeSortedString(),
+ Tag.INDEX.toString(),
+ String.valueOf(sink.getSinkIndex()),
+ Tag.CREATION_TIME.toString(),
+ String.valueOf(sink.getCreationTime())
+ }
+ : new String[] {
+ Tag.NAME.toString(),
+ sink.getAttributeSortedString(),
+ Tag.PIPE.toString(),
+ sink.getPipeName(),
+ Tag.INDEX.toString(),
+ String.valueOf(sink.getSinkIndex()),
+ Tag.CREATION_TIME.toString(),
+ String.valueOf(sink.getCreationTime())
+ };
+ }
+
+ private static String[] getCreationTimeTags(final PipeSinkSubtask sink) {
+ return sink.getPipeName() == null
+ ? new String[] {
+ Tag.NAME.toString(),
+ sink.getAttributeSortedString(),
+ Tag.CREATION_TIME.toString(),
+ String.valueOf(sink.getCreationTime())
+ }
+ : new String[] {
+ Tag.NAME.toString(),
+ sink.getAttributeSortedString(),
+ Tag.PIPE.toString(),
+ sink.getPipeName(),
+ Tag.CREATION_TIME.toString(),
+ String.valueOf(sink.getCreationTime())
+ };
+ }
+
+ private static String[] getNameTags(final PipeSinkSubtask sink) {
+ return sink.getPipeName() == null
+ ? new String[] {Tag.NAME.toString(), sink.getAttributeSortedString()}
+ : new String[] {
+ Tag.NAME.toString(),
+ sink.getAttributeSortedString(),
+ Tag.PIPE.toString(),
+ sink.getPipeName(),
+ Tag.CREATION_TIME.toString(),
+ String.valueOf(sink.getCreationTime())
+ };
+ }
+
private void createAutoGauge(final String taskID) {
final PipeSinkSubtask sink = sinkMap.get(taskID);
// Pending event count
@@ -82,80 +135,45 @@ public class PipeDataRegionSinkMetrics implements IMetricSet {
MetricLevel.IMPORTANT,
sink,
PipeSinkSubtask::getTabletInsertionEventCount,
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(sink.getSinkIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime()));
+ getIndexedTags(sink));
metricService.createAutoGauge(
Metric.UNTRANSFERRED_TSFILE_COUNT.toString(),
MetricLevel.IMPORTANT,
sink,
PipeSinkSubtask::getTsFileInsertionEventCount,
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(sink.getSinkIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime()));
+ getIndexedTags(sink));
metricService.createAutoGauge(
Metric.UNTRANSFERRED_HEARTBEAT_COUNT.toString(),
MetricLevel.IMPORTANT,
sink,
PipeSinkSubtask::getPipeHeartbeatEventCount,
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(sink.getSinkIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime()));
+ getIndexedTags(sink));
// Metrics related to IoTDBThriftAsyncSink
metricService.createAutoGauge(
Metric.PIPE_ASYNC_CONNECTOR_RETRY_EVENT_QUEUE_SIZE.toString(),
MetricLevel.IMPORTANT,
sink,
PipeSinkSubtask::getAsyncSinkRetryEventQueueSize,
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(sink.getSinkIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime()));
+ getIndexedTags(sink));
metricService.createAutoGauge(
Metric.PIPE_PENDING_HANDLERS_SIZE.toString(),
MetricLevel.IMPORTANT,
sink,
PipeSinkSubtask::getPendingHandlersSize,
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(sink.getSinkIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime()));
+ getIndexedTags(sink));
// Metrics related to IoTDB sink
metricService.createAutoGauge(
Metric.PIPE_TOTAL_UNCOMPRESSED_SIZE.toString(),
MetricLevel.IMPORTANT,
sink,
PipeSinkSubtask::getTotalUncompressedSize,
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(sink.getSinkIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime()));
+ getIndexedTags(sink));
metricService.createAutoGauge(
Metric.PIPE_TOTAL_COMPRESSED_SIZE.toString(),
MetricLevel.IMPORTANT,
sink,
PipeSinkSubtask::getTotalCompressedSize,
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(sink.getSinkIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime()));
+ getIndexedTags(sink));
}
private void createRate(final String taskID) {
@@ -166,34 +184,19 @@ public class PipeDataRegionSinkMetrics implements IMetricSet {
metricService.getOrCreateRate(
Metric.PIPE_CONNECTOR_TABLET_TRANSFER.toString(),
MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(sink.getSinkIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime())));
+ getIndexedTags(sink)));
tsFileRateMap.put(
taskID,
metricService.getOrCreateRate(
Metric.PIPE_CONNECTOR_TSFILE_TRANSFER.toString(),
MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(sink.getSinkIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime())));
+ getIndexedTags(sink)));
pipeHeartbeatRateMap.put(
taskID,
metricService.getOrCreateRate(
Metric.PIPE_CONNECTOR_HEARTBEAT_TRANSFER.toString(),
MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(sink.getSinkIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime())));
+ getIndexedTags(sink)));
}
private void createTimer(final String taskID) {
@@ -203,10 +206,7 @@ public class PipeDataRegionSinkMetrics implements IMetricSet {
metricService.getOrCreateTimer(
Metric.PIPE_COMPRESSION_TIME.toString(),
MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime())));
+ getCreationTimeTags(sink)));
}
private void createHistogram(final String taskID) {
@@ -216,48 +216,33 @@ public class PipeDataRegionSinkMetrics implements IMetricSet {
metricService.getOrCreateHistogram(
Metric.PIPE_INSERT_NODE_BATCH_SIZE.toString(),
MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime()));
+ getCreationTimeTags(sink));
sink.setTabletBatchSizeHistogram(tabletBatchSizeHistogram);
final Histogram tsFileBatchSizeHistogram =
metricService.getOrCreateHistogram(
Metric.PIPE_TSFILE_BATCH_SIZE.toString(),
MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime()));
+ getCreationTimeTags(sink));
sink.setTsFileBatchSizeHistogram(tsFileBatchSizeHistogram);
final Histogram tabletBatchTimeIntervalHistogram =
metricService.getOrCreateHistogram(
Metric.PIPE_INSERT_NODE_BATCH_TIME_COST.toString(),
MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime()));
+ getCreationTimeTags(sink));
sink.setTabletBatchTimeIntervalHistogram(tabletBatchTimeIntervalHistogram);
final Histogram tsFileBatchTimeIntervalHistogram =
metricService.getOrCreateHistogram(
Metric.PIPE_TSFILE_BATCH_TIME_COST.toString(),
MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime()));
+ getCreationTimeTags(sink));
sink.setTsFileBatchTimeIntervalHistogram(tsFileBatchTimeIntervalHistogram);
Histogram eventSizeHistogram =
metricService.getOrCreateHistogram(
- Metric.PIPE_CONNECTOR_BATCH_SIZE.toString(),
- MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- sink.getAttributeSortedString());
+ Metric.PIPE_CONNECTOR_BATCH_SIZE.toString(), MetricLevel.IMPORTANT, getNameTags(sink));
sink.setEventSizeHistogram(eventSizeHistogram);
}
@@ -283,102 +268,38 @@ public class PipeDataRegionSinkMetrics implements IMetricSet {
final PipeSinkSubtask sink = sinkMap.get(taskID);
// Pending event count
metricService.remove(
- MetricType.AUTO_GAUGE,
- Metric.UNTRANSFERRED_TABLET_COUNT.toString(),
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(sink.getSinkIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime()));
+ MetricType.AUTO_GAUGE, Metric.UNTRANSFERRED_TABLET_COUNT.toString(), getIndexedTags(sink));
metricService.remove(
- MetricType.AUTO_GAUGE,
- Metric.UNTRANSFERRED_TSFILE_COUNT.toString(),
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(sink.getSinkIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime()));
+ MetricType.AUTO_GAUGE, Metric.UNTRANSFERRED_TSFILE_COUNT.toString(), getIndexedTags(sink));
metricService.remove(
MetricType.AUTO_GAUGE,
Metric.UNTRANSFERRED_HEARTBEAT_COUNT.toString(),
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(sink.getSinkIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime()));
+ getIndexedTags(sink));
// Metrics related to IoTDBThriftAsyncSink
metricService.remove(
MetricType.AUTO_GAUGE,
Metric.PIPE_ASYNC_CONNECTOR_RETRY_EVENT_QUEUE_SIZE.toString(),
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(sink.getSinkIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime()));
+ getIndexedTags(sink));
metricService.remove(
- MetricType.AUTO_GAUGE,
- Metric.PIPE_PENDING_HANDLERS_SIZE.toString(),
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(sink.getSinkIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime()));
+ MetricType.AUTO_GAUGE, Metric.PIPE_PENDING_HANDLERS_SIZE.toString(), getIndexedTags(sink));
// Metrics related to IoTDB sink
metricService.remove(
MetricType.AUTO_GAUGE,
Metric.PIPE_TOTAL_UNCOMPRESSED_SIZE.toString(),
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(sink.getSinkIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime()));
+ getIndexedTags(sink));
metricService.remove(
- MetricType.AUTO_GAUGE,
- Metric.PIPE_TOTAL_COMPRESSED_SIZE.toString(),
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(sink.getSinkIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime()));
+ MetricType.AUTO_GAUGE, Metric.PIPE_TOTAL_COMPRESSED_SIZE.toString(), getIndexedTags(sink));
}
private void removeRate(final String taskID) {
final PipeSinkSubtask sink = sinkMap.get(taskID);
// Transfer event rate
metricService.remove(
- MetricType.RATE,
- Metric.PIPE_CONNECTOR_TABLET_TRANSFER.toString(),
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(sink.getSinkIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime()));
+ MetricType.RATE, Metric.PIPE_CONNECTOR_TABLET_TRANSFER.toString(), getIndexedTags(sink));
metricService.remove(
- MetricType.RATE,
- Metric.PIPE_CONNECTOR_TSFILE_TRANSFER.toString(),
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(sink.getSinkIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime()));
+ MetricType.RATE, Metric.PIPE_CONNECTOR_TSFILE_TRANSFER.toString(), getIndexedTags(sink));
metricService.remove(
- MetricType.RATE,
- Metric.PIPE_CONNECTOR_HEARTBEAT_TRANSFER.toString(),
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(sink.getSinkIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime()));
+ MetricType.RATE, Metric.PIPE_CONNECTOR_HEARTBEAT_TRANSFER.toString(), getIndexedTags(sink));
tabletRateMap.remove(taskID);
tsFileRateMap.remove(taskID);
pipeHeartbeatRateMap.remove(taskID);
@@ -387,12 +308,7 @@ public class PipeDataRegionSinkMetrics implements IMetricSet {
private void removeTimer(final String taskID) {
final PipeSinkSubtask sink = sinkMap.get(taskID);
metricService.remove(
- MetricType.TIMER,
- Metric.PIPE_COMPRESSION_TIME.toString(),
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime()));
+ MetricType.TIMER, Metric.PIPE_COMPRESSION_TIME.toString(), getCreationTimeTags(sink));
compressionTimerMap.remove(taskID);
}
@@ -401,37 +317,20 @@ public class PipeDataRegionSinkMetrics implements IMetricSet {
metricService.remove(
MetricType.HISTOGRAM,
Metric.PIPE_INSERT_NODE_BATCH_SIZE.toString(),
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime()));
+ getCreationTimeTags(sink));
metricService.remove(
- MetricType.HISTOGRAM,
- Metric.PIPE_TSFILE_BATCH_SIZE.toString(),
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime()));
+ MetricType.HISTOGRAM, Metric.PIPE_TSFILE_BATCH_SIZE.toString(), getCreationTimeTags(sink));
metricService.remove(
MetricType.HISTOGRAM,
Metric.PIPE_INSERT_NODE_BATCH_TIME_COST.toString(),
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime()));
+ getCreationTimeTags(sink));
metricService.remove(
MetricType.HISTOGRAM,
Metric.PIPE_TSFILE_BATCH_TIME_COST.toString(),
- Tag.NAME.toString(),
- sink.getAttributeSortedString(),
- Tag.CREATION_TIME.toString(),
- String.valueOf(sink.getCreationTime()));
+ getCreationTimeTags(sink));
metricService.remove(
- MetricType.HISTOGRAM,
- Metric.PIPE_CONNECTOR_BATCH_SIZE.toString(),
- Tag.NAME.toString(),
- sink.getAttributeSortedString());
+ MetricType.HISTOGRAM, Metric.PIPE_CONNECTOR_BATCH_SIZE.toString(), getNameTags(sink));
}
//////////////////////////// register & deregister (pipe integration) ////////////////////////////
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgentTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgentTest.java
index 018ec371d91..83709d41c3f 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgentTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgentTest.java
@@ -23,24 +23,32 @@ import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
import org.apache.iotdb.commons.consensus.index.impl.SimpleProgressIndex;
+import org.apache.iotdb.commons.exception.pipe.PipeRuntimeCriticalException;
+import org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkCriticalException;
import org.apache.iotdb.commons.pipe.agent.task.PipeTaskAgent;
+import org.apache.iotdb.commons.pipe.agent.task.PipeTaskManager;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMetaKeeper;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeRuntimeMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
import org.apache.iotdb.pipe.api.exception.PipeException;
+import org.awaitility.Awaitility;
import org.junit.Assert;
import org.junit.Test;
import java.lang.reflect.Field;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.TimeUnit;
public class PipeDataNodeTaskAgentTest {
@@ -103,6 +111,197 @@ public class PipeDataNodeTaskAgentTest {
}
}
+ @Test
+ public void testSinkCriticalExceptionIsPropagatedOnlyWithinItsPipe() throws Exception {
+ final PipeDataNodeTaskAgent taskAgent = new PipeDataNodeTaskAgent();
+ final PipeMetaKeeper pipeMetaKeeper = getField(taskAgent, "pipeMetaKeeper");
+ final int localNodeId = IoTDBDescriptor.getInstance().getConfig().getDataNodeId();
+
+ final PipeTaskMeta failedTaskMeta =
+ new PipeTaskMeta(MinimumProgressIndex.INSTANCE, localNodeId);
+ final PipeTaskMeta failedPipeSecondTaskMeta =
+ new PipeTaskMeta(MinimumProgressIndex.INSTANCE, localNodeId);
+ final ConcurrentMap<Integer, PipeTaskMeta> failedPipeTaskMetaMap = new ConcurrentHashMap<>();
+ failedPipeTaskMetaMap.put(REGION_ID, failedTaskMeta);
+ failedPipeTaskMetaMap.put(REGION_ID + 1, failedPipeSecondTaskMeta);
+ final PipeMeta failedPipeMeta =
+ new PipeMeta(
+ new PipeStaticMeta("failedPipe", 1L, new HashMap<>(), new HashMap<>(), new HashMap<>()),
+ new PipeRuntimeMeta(failedPipeTaskMetaMap));
+ failedPipeMeta.getRuntimeMeta().getStatus().set(PipeStatus.RUNNING);
+
+ final PipeTaskMeta unaffectedTaskMeta =
+ new PipeTaskMeta(MinimumProgressIndex.INSTANCE, localNodeId);
+ final ConcurrentMap<Integer, PipeTaskMeta> unaffectedPipeTaskMetaMap =
+ new ConcurrentHashMap<>();
+ unaffectedPipeTaskMetaMap.put(REGION_ID, unaffectedTaskMeta);
+ final PipeMeta unaffectedPipeMeta =
+ new PipeMeta(
+ new PipeStaticMeta(
+ "unaffectedPipe", 1L, new HashMap<>(), new HashMap<>(), new HashMap<>()),
+ new PipeRuntimeMeta(unaffectedPipeTaskMetaMap));
+ unaffectedPipeMeta.getRuntimeMeta().getStatus().set(PipeStatus.RUNNING);
+
+ pipeMetaKeeper.addPipeMeta(failedPipeMeta);
+ pipeMetaKeeper.addPipeMeta(unaffectedPipeMeta);
+
+ final PipeRuntimeSinkCriticalException exception =
+ new PipeRuntimeSinkCriticalException("sink failure", 1L);
+ taskAgent.stopAllPipesWithCriticalExceptionAndTrackException(failedTaskMeta, exception);
+
+ Awaitility.await()
+ .atMost(5, TimeUnit.SECONDS)
+ .until(
+ () ->
+ PipeStatus.STOPPED.equals(failedPipeMeta.getRuntimeMeta().getStatus().get())
+ && failedPipeSecondTaskMeta.containsExceptionMessage(exception));
+
+ Assert.assertEquals(PipeStatus.RUNNING, unaffectedPipeMeta.getRuntimeMeta().getStatus().get());
+ Assert.assertFalse(unaffectedTaskMeta.hasExceptionMessages());
+ }
+
+ @Test
+ public void testExplicitPipeIdentityDoesNotFallBackToTaskMeta() throws Exception {
+ final PipeDataNodeTaskAgent taskAgent = new PipeDataNodeTaskAgent();
+ final PipeMetaKeeper pipeMetaKeeper = getField(taskAgent, "pipeMetaKeeper");
+ final int localNodeId = IoTDBDescriptor.getInstance().getConfig().getDataNodeId();
+
+ final PipeTaskMeta taskMeta = new PipeTaskMeta(MinimumProgressIndex.INSTANCE, localNodeId);
+ final ConcurrentMap<Integer, PipeTaskMeta> taskMetaMap = new ConcurrentHashMap<>();
+ taskMetaMap.put(REGION_ID, taskMeta);
+ final PipeMeta pipeMeta =
+ new PipeMeta(
+ new PipeStaticMeta(
+ "existingPipe", 1L, new HashMap<>(), new HashMap<>(), new HashMap<>()),
+ new PipeRuntimeMeta(taskMetaMap));
+ pipeMeta.getRuntimeMeta().getStatus().set(PipeStatus.RUNNING);
+ pipeMetaKeeper.addPipeMeta(pipeMeta);
+
+ taskAgent.stopAllPipesWithCriticalExceptionAndTrackException(
+ "existingPipe",
+ Long.MIN_VALUE,
+ taskMeta,
+ new PipeRuntimeCriticalException("stale pipe identity", 1L));
+
+ Awaitility.await()
+ .during(500, TimeUnit.MILLISECONDS)
+ .atMost(2, TimeUnit.SECONDS)
+ .until(
+ () ->
+ PipeStatus.RUNNING.equals(pipeMeta.getRuntimeMeta().getStatus().get())
+ && !taskMeta.hasExceptionMessages());
+ }
+
+ @Test
+ public void testDetachedTaskMetaMustIdentifyOnePipeUniquely() throws Exception {
+ final PipeDataNodeTaskAgent taskAgent = new PipeDataNodeTaskAgent();
+ final PipeMetaKeeper pipeMetaKeeper = getField(taskAgent, "pipeMetaKeeper");
+ final int localNodeId = IoTDBDescriptor.getInstance().getConfig().getDataNodeId();
+
+ final PipeTaskMeta firstTaskMeta = new PipeTaskMeta(MinimumProgressIndex.INSTANCE, localNodeId);
+ final PipeTaskMeta secondTaskMeta =
+ new PipeTaskMeta(MinimumProgressIndex.INSTANCE, localNodeId);
+ final ConcurrentMap<Integer, PipeTaskMeta> firstTaskMetaMap = new ConcurrentHashMap<>();
+ firstTaskMetaMap.put(REGION_ID, firstTaskMeta);
+ final ConcurrentMap<Integer, PipeTaskMeta> secondTaskMetaMap = new ConcurrentHashMap<>();
+ secondTaskMetaMap.put(REGION_ID, secondTaskMeta);
+ final PipeMeta firstPipeMeta =
+ new PipeMeta(
+ new PipeStaticMeta("firstPipe", 1L, new HashMap<>(), new HashMap<>(), new HashMap<>()),
+ new PipeRuntimeMeta(firstTaskMetaMap));
+ final PipeMeta secondPipeMeta =
+ new PipeMeta(
+ new PipeStaticMeta("secondPipe", 1L, new HashMap<>(), new HashMap<>(), new HashMap<>()),
+ new PipeRuntimeMeta(secondTaskMetaMap));
+ firstPipeMeta.getRuntimeMeta().getStatus().set(PipeStatus.RUNNING);
+ secondPipeMeta.getRuntimeMeta().getStatus().set(PipeStatus.RUNNING);
+ pipeMetaKeeper.addPipeMeta(firstPipeMeta);
+ pipeMetaKeeper.addPipeMeta(secondPipeMeta);
+
+ final PipeTaskMeta detachedTaskMeta =
+ new PipeTaskMeta(MinimumProgressIndex.INSTANCE, localNodeId);
+ taskAgent.stopAllPipesWithCriticalExceptionAndTrackException(
+ detachedTaskMeta, new PipeRuntimeCriticalException("ambiguous task meta", 1L));
+
+ Awaitility.await().atMost(2, TimeUnit.SECONDS).until(detachedTaskMeta::hasExceptionMessages);
+ pipeMetaKeeper.acquireWriteLock();
+ pipeMetaKeeper.releaseWriteLock();
+
+ Assert.assertEquals(PipeStatus.RUNNING, firstPipeMeta.getRuntimeMeta().getStatus().get());
+ Assert.assertEquals(PipeStatus.RUNNING, secondPipeMeta.getRuntimeMeta().getStatus().get());
+ Assert.assertFalse(firstTaskMeta.hasExceptionMessages());
+ Assert.assertFalse(secondTaskMeta.hasExceptionMessages());
+ }
+
+ @Test
+ public void testDetachedTaskMetaIsRecordedOnIdentifiedPipe() throws Exception {
+ final PipeDataNodeTaskAgent taskAgent = new PipeDataNodeTaskAgent();
+ final PipeMetaKeeper pipeMetaKeeper = getField(taskAgent, "pipeMetaKeeper");
+ final PipeTaskManager pipeTaskManager = getField(taskAgent, "pipeTaskManager");
+ final int localNodeId = IoTDBDescriptor.getInstance().getConfig().getDataNodeId();
+
+ final PipeTaskMeta localTaskMeta = new PipeTaskMeta(MinimumProgressIndex.INSTANCE, localNodeId);
+ final ConcurrentMap<Integer, PipeTaskMeta> taskMetaMap = new ConcurrentHashMap<>();
+ taskMetaMap.put(REGION_ID, localTaskMeta);
+ final PipeMeta pipeMeta =
+ new PipeMeta(
+ new PipeStaticMeta(
+ "detachedPipe", 2L, new HashMap<>(), new HashMap<>(), new HashMap<>()),
+ new PipeRuntimeMeta(taskMetaMap));
+ pipeMeta.getRuntimeMeta().getStatus().set(PipeStatus.RUNNING);
+ pipeMetaKeeper.addPipeMeta(pipeMeta);
+ pipeTaskManager.addPipeTasks(pipeMeta.getStaticMeta(), Collections.emptyMap());
+
+ final PipeTaskMeta detachedTaskMeta =
+ new PipeTaskMeta(MinimumProgressIndex.INSTANCE, localNodeId);
+ final PipeRuntimeCriticalException exception =
+ new PipeRuntimeCriticalException("detached failure", 2L);
+ taskAgent.stopAllPipesWithCriticalExceptionAndTrackException(
+ "detachedPipe", 2L, detachedTaskMeta, exception);
+
+ Awaitility.await()
+ .atMost(5, TimeUnit.SECONDS)
+ .until(
+ () ->
+ PipeStatus.STOPPED.equals(pipeMeta.getRuntimeMeta().getStatus().get())
+ && localTaskMeta.containsExceptionMessage(exception));
+ Assert.assertTrue(pipeMeta.getRuntimeMeta().getIsStoppedByRuntimeException());
+ }
+
+ @Test
+ public void testCriticalExceptionWithNullTaskMetaStopsIdentifiedPipe() throws Exception {
+ final PipeDataNodeTaskAgent taskAgent = new PipeDataNodeTaskAgent();
+ final PipeMetaKeeper pipeMetaKeeper = getField(taskAgent, "pipeMetaKeeper");
+
+ final PipeMeta failedPipeMeta =
+ new PipeMeta(
+ new PipeStaticMeta(
+ "nullTaskMetaPipe", 3L, new HashMap<>(), new HashMap<>(), new HashMap<>()),
+ new PipeRuntimeMeta());
+ failedPipeMeta.getRuntimeMeta().getStatus().set(PipeStatus.RUNNING);
+ final PipeMeta unaffectedPipeMeta =
+ new PipeMeta(
+ new PipeStaticMeta(
+ "stillRunningPipe", 3L, new HashMap<>(), new HashMap<>(), new HashMap<>()),
+ new PipeRuntimeMeta());
+ unaffectedPipeMeta.getRuntimeMeta().getStatus().set(PipeStatus.RUNNING);
+ pipeMetaKeeper.addPipeMeta(failedPipeMeta);
+ pipeMetaKeeper.addPipeMeta(unaffectedPipeMeta);
+
+ final PipeRuntimeSinkCriticalException exception =
+ new PipeRuntimeSinkCriticalException("null task meta failure", 3L);
+ taskAgent.stopAllPipesWithCriticalExceptionAndTrackException(
+ "nullTaskMetaPipe", 3L, null, exception);
+
+ Awaitility.await()
+ .atMost(5, TimeUnit.SECONDS)
+ .until(
+ () ->
+ PipeStatus.STOPPED.equals(failedPipeMeta.getRuntimeMeta().getStatus().get())
+ && failedPipeMeta.getRuntimeMeta().getIsStoppedByRuntimeException());
+ Assert.assertEquals(PipeStatus.RUNNING, unaffectedPipeMeta.getRuntimeMeta().getStatus().get());
+ }
+
@Test
public void testCarryOverCommittedProgressForResumeAlter() {
final PipeStaticMeta oldStaticMeta = createStaticMeta(1, false);
@@ -250,4 +449,12 @@ public class PipeDataNodeTaskAgentTest {
taskMetaMap.put(REGION_ID, new PipeTaskMeta(progressIndex, leaderId));
return new PipeMeta(staticMeta, new PipeRuntimeMeta(taskMetaMap));
}
+
+ @SuppressWarnings("unchecked")
+ private <T> T getField(final PipeDataNodeTaskAgent taskAgent, final String fieldName)
+ throws ReflectiveOperationException {
+ final Field field = PipeTaskAgent.class.getDeclaredField(fieldName);
+ field.setAccessible(true);
+ return (T) field.get(taskAgent);
+ }
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManagerTest.java
index dad19044251..8c2fd176587 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManagerTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManagerTest.java
@@ -20,9 +20,15 @@
package org.apache.iotdb.db.pipe.agent.task.subtask.sink;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin;
+import org.apache.iotdb.commons.pipe.agent.task.connection.UnboundedBlockingPendingQueue;
import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant;
import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
+import org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskSinkRuntimeEnvironment;
+import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
+import org.apache.iotdb.db.pipe.agent.task.execution.PipeSinkSubtaskExecutor;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
+import org.apache.iotdb.pipe.api.event.Event;
+import org.apache.iotdb.pipe.api.exception.PipeException;
import org.junit.Assert;
import org.junit.Test;
@@ -32,6 +38,89 @@ import java.util.Map;
public class PipeSinkSubtaskManagerTest {
+ @Test
+ public void testSubtasksAreSharedOnlyWithinSamePipe() {
+ // Initialize the task agent used by PipeEventCommitManager before registering subtasks.
+ PipeDataNodeAgent.task();
+
+ final String firstPipeName = "firstPipe";
+ final String secondPipeName = "secondPipe";
+ final long creationTime = 1L;
+ final int firstRegionId = -1;
+ final int secondRegionId = -2;
+ final Map<String, String> attributes = new HashMap<>();
+ attributes.put(
+ PipeSinkConstant.CONNECTOR_KEY, BuiltinPipePlugin.DO_NOTHING_CONNECTOR.getPipePluginName());
+ attributes.put(PipeSinkConstant.CONNECTOR_IOTDB_PARALLEL_TASKS_KEY, "1");
+ final PipeParameters parameters = new PipeParameters(attributes);
+ final PipeSinkSubtaskManager manager = PipeSinkSubtaskManager.instance();
+
+ boolean firstRegionRegistered = false;
+ boolean secondRegionRegistered = false;
+ boolean secondPipeRegistered = false;
+ try {
+ final String firstPipeSubtaskId =
+ manager.register(
+ PipeSinkSubtaskExecutor::new,
+ parameters,
+ new PipeTaskSinkRuntimeEnvironment(firstPipeName, creationTime, firstRegionId));
+ firstRegionRegistered = true;
+ final String firstPipeSecondRegionSubtaskId =
+ manager.register(
+ PipeSinkSubtaskExecutor::new,
+ parameters,
+ new PipeTaskSinkRuntimeEnvironment(firstPipeName, creationTime, secondRegionId));
+ secondRegionRegistered = true;
+ final UnboundedBlockingPendingQueue<Event> firstPipeQueue =
+ manager.getPipeSinkPendingQueue(firstPipeName, creationTime, firstPipeSubtaskId);
+ Assert.assertSame(firstPipeQueue, manager.getPipeSinkPendingQueue(firstPipeSubtaskId));
+ Assert.assertTrue(manager.hasRegisteredSubtasks(parameters, firstRegionId));
+
+ final String secondPipeSubtaskId =
+ manager.register(
+ PipeSinkSubtaskExecutor::new,
+ parameters,
+ new PipeTaskSinkRuntimeEnvironment(secondPipeName, creationTime, firstRegionId));
+ secondPipeRegistered = true;
+
+ Assert.assertSame(
+ firstPipeQueue,
+ manager.getPipeSinkPendingQueue(
+ firstPipeName, creationTime, firstPipeSecondRegionSubtaskId));
+ Assert.assertNotSame(
+ firstPipeQueue,
+ manager.getPipeSinkPendingQueue(secondPipeName, creationTime, secondPipeSubtaskId));
+ Assert.assertThrows(
+ PipeException.class, () -> manager.getPipeSinkPendingQueue(firstPipeSubtaskId));
+ Assert.assertThrows(
+ PipeException.class, () -> manager.hasRegisteredSubtasks(parameters, firstRegionId));
+ Assert.assertThrows(PipeException.class, () -> manager.start(firstPipeSubtaskId));
+ Assert.assertThrows(PipeException.class, () -> manager.stop(firstPipeSubtaskId));
+ } finally {
+ if (secondRegionRegistered) {
+ manager.deregister(
+ firstPipeName,
+ creationTime,
+ secondRegionId,
+ PipeSinkSubtaskManager.generateAttributeSortedString(parameters, secondRegionId));
+ }
+ if (firstRegionRegistered) {
+ manager.deregister(
+ firstPipeName,
+ creationTime,
+ firstRegionId,
+ PipeSinkSubtaskManager.generateAttributeSortedString(parameters, firstRegionId));
+ }
+ if (secondPipeRegistered) {
+ manager.deregister(
+ secondPipeName,
+ creationTime,
+ firstRegionId,
+ PipeSinkSubtaskManager.generateAttributeSortedString(parameters, firstRegionId));
+ }
+ }
+ }
+
@Test
public void testGenerateAttributeSortedStringUsesSerializeByRegionAndIgnoresRestartFlag() {
final Map<String, String> attributes = new HashMap<>();
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/metric/schema/PipeSchemaRegionSinkMetricsTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/metric/schema/PipeSchemaRegionSinkMetricsTest.java
index 744e640189c..f695c604ea1 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/metric/schema/PipeSchemaRegionSinkMetricsTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/metric/schema/PipeSchemaRegionSinkMetricsTest.java
@@ -151,4 +151,81 @@ public class PipeSchemaRegionSinkMetricsTest {
metricServiceField.set(metrics, null);
}
}
+
+ @Test
+ public void testBatchHistogramIsIsolatedByPipeIdentity() throws Exception {
+ final String taskId = "schema-pipe-task-" + System.nanoTime();
+ boolean deregistered = false;
+ final AbstractMetricService metricService = Mockito.mock(AbstractMetricService.class);
+ final PipeSinkSubtask subtask = Mockito.mock(PipeSinkSubtask.class);
+ final Rate rate = Mockito.mock(Rate.class);
+ final Histogram eventSizeHistogram = Mockito.mock(Histogram.class);
+
+ when(subtask.getTaskID()).thenReturn(taskId);
+ when(subtask.getAttributeSortedString()).thenReturn("schema_test");
+ when(subtask.getPipeName()).thenReturn("pipe");
+ when(subtask.getCreationTime()).thenReturn(2L);
+ when(metricService.getOrCreateRate(
+ eq(Metric.PIPE_CONNECTOR_SCHEMA_TRANSFER.toString()),
+ eq(MetricLevel.IMPORTANT),
+ eq(Tag.NAME.toString()),
+ eq("schema_test"),
+ eq(Tag.PIPE.toString()),
+ eq("pipe"),
+ eq(Tag.CREATION_TIME.toString()),
+ eq("2")))
+ .thenReturn(rate);
+ when(metricService.getOrCreateHistogram(
+ eq(Metric.PIPE_CONNECTOR_BATCH_SIZE.toString()),
+ eq(MetricLevel.IMPORTANT),
+ eq(Tag.NAME.toString()),
+ eq("schema_test"),
+ eq(Tag.PIPE.toString()),
+ eq("pipe"),
+ eq(Tag.CREATION_TIME.toString()),
+ eq("2")))
+ .thenReturn(eventSizeHistogram);
+
+ final PipeSchemaRegionSinkMetrics metrics = PipeSchemaRegionSinkMetrics.getInstance();
+ final Field metricServiceField =
+ PipeSchemaRegionSinkMetrics.class.getDeclaredField("metricService");
+ metricServiceField.setAccessible(true);
+ final Field connectorMapField =
+ PipeSchemaRegionSinkMetrics.class.getDeclaredField("connectorMap");
+ connectorMapField.setAccessible(true);
+ final Field schemaRateMapField =
+ PipeSchemaRegionSinkMetrics.class.getDeclaredField("schemaRateMap");
+ schemaRateMapField.setAccessible(true);
+
+ ((Map<?, ?>) connectorMapField.get(metrics)).clear();
+ ((Map<?, ?>) schemaRateMapField.get(metrics)).clear();
+ metricServiceField.set(metrics, null);
+
+ try {
+ metrics.register(subtask);
+ metrics.bindTo(metricService);
+
+ verify(subtask).setEventSizeHistogram(eventSizeHistogram);
+
+ metrics.deregister(taskId);
+ verify(metricService)
+ .remove(
+ MetricType.HISTOGRAM,
+ Metric.PIPE_CONNECTOR_BATCH_SIZE.toString(),
+ Tag.NAME.toString(),
+ "schema_test",
+ Tag.PIPE.toString(),
+ "pipe",
+ Tag.CREATION_TIME.toString(),
+ "2");
+ deregistered = true;
+ } finally {
+ if (!deregistered) {
+ metrics.deregister(taskId);
+ }
+ ((Map<?, ?>) connectorMapField.get(metrics)).clear();
+ ((Map<?, ?>) schemaRateMapField.get(metrics)).clear();
+ metricServiceField.set(metrics, null);
+ }
+ }
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java
index 5523717b093..ceb4f257733 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java
@@ -54,7 +54,6 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
-import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
@@ -970,6 +969,21 @@ public abstract class PipeTaskAgent {
final int currentNodeId,
final PipeTaskMeta pipeTaskMeta,
final PipeRuntimeException pipeRuntimeException) {
+ stopAllPipesWithCriticalException(
+ currentNodeId, null, Long.MIN_VALUE, pipeTaskMeta, pipeRuntimeException);
+ }
+
+ /**
+ * Stops the pipe that owns a critical exception and, for sink failures, its other region tasks.
+ * The pipe identity is passed separately because the task meta in an event may be a serialized
+ * copy rather than the object currently held by the task agent.
+ */
+ protected void stopAllPipesWithCriticalException(
+ final int currentNodeId,
+ final String pipeName,
+ final long creationTime,
+ final PipeTaskMeta pipeTaskMeta,
+ final PipeRuntimeException pipeRuntimeException) {
// To avoid deadlock, we use a new thread to stop all pipes.
CompletableFuture.runAsync(
() -> {
@@ -978,8 +992,41 @@ public abstract class PipeTaskAgent {
while (true) {
if (tryWriteLockWithTimeOut(5)) {
try {
- pipeTaskMeta.trackExceptionMessage(pipeRuntimeException);
- stopAllPipesWithCriticalExceptionInternal(currentNodeId);
+ final PipeMeta failedPipeMeta =
+ findPipeMeta(pipeName, creationTime, pipeTaskMeta);
+ final PipeTaskMeta localFailedPipeTaskMeta =
+ findLocalPipeTaskMeta(failedPipeMeta, pipeTaskMeta, currentNodeId);
+
+ // An explicit pipe identity is authoritative. If it is stale, do not fall back
+ // to the supplied task meta because identical task metadata is valid in another
+ // pipe.
+ if (pipeName != null && failedPipeMeta == null) {
+ return;
+ }
+
+ if (failedPipeMeta != null
+ && failedPipeMeta.getRuntimeMeta().getStatus().get() == PipeStatus.RUNNING) {
+ failedPipeMeta.getRuntimeMeta().setIsStoppedByRuntimeException(true);
+ }
+
+ if (failedPipeMeta == null) {
+ // Keep the legacy behavior for callers that only provide a task meta. When
+ // the object is detached and cannot be mapped to a unique local pipe, there
+ // is no safe pipe to stop.
+ if (pipeTaskMeta != null) {
+ pipeTaskMeta.trackExceptionMessage(pipeRuntimeException);
+ }
+ } else if (localFailedPipeTaskMeta != null) {
+ localFailedPipeTaskMeta.trackExceptionMessage(pipeRuntimeException);
+ }
+
+ stopAllPipesWithCriticalExceptionInternal(
+ currentNodeId, failedPipeMeta, pipeRuntimeException);
+ if (failedPipeMeta != null) {
+ // stopPipe intentionally returns without changing the runtime status if its
+ // task map was concurrently removed. The identified pipe still has to stop.
+ stopPipeWithRuntimeException(failedPipeMeta);
+ }
LOGGER.info(PipeMessages.STOPPED_ALL_PIPES_WITH_CRITICAL_EXCEPTION);
return;
} finally {
@@ -999,69 +1046,40 @@ public abstract class PipeTaskAgent {
});
}
- private void stopAllPipesWithCriticalExceptionInternal(final int currentNodeId) {
- // 1. track exception in all pipe tasks that share the same connector that have critical
- // exceptions.
- final Map<PipeParameters, PipeRuntimeSinkCriticalException>
- reusedConnectorParameters2ExceptionMap = new HashMap<>();
-
- pipeMetaKeeper
- .getPipeMetaList()
- .forEach(
- pipeMeta -> {
- final PipeStaticMeta staticMeta = pipeMeta.getStaticMeta();
- final PipeRuntimeMeta runtimeMeta = pipeMeta.getRuntimeMeta();
-
- runtimeMeta
- .getConsensusGroupId2TaskMetaMap()
- .values()
- .forEach(
- pipeTaskMeta -> {
- if (pipeTaskMeta.getLeaderNodeId() != currentNodeId) {
- return;
- }
-
- for (final PipeRuntimeException e : pipeTaskMeta.getExceptionMessages()) {
- if (e instanceof PipeRuntimeSinkCriticalException) {
- reusedConnectorParameters2ExceptionMap.putIfAbsent(
- staticMeta.getSinkParameters(),
- (PipeRuntimeSinkCriticalException) e);
- }
- }
- });
- });
- pipeMetaKeeper
- .getPipeMetaList()
- .forEach(
- pipeMeta -> {
- final PipeStaticMeta staticMeta = pipeMeta.getStaticMeta();
- final PipeRuntimeMeta runtimeMeta = pipeMeta.getRuntimeMeta();
+ private void stopAllPipesWithCriticalExceptionInternal(
+ final int currentNodeId,
+ final PipeMeta failedPipeMeta,
+ final PipeRuntimeException pipeRuntimeException) {
+ // 1. A sink subtask is shared only by regions of one pipe. Locate that pipe through its
+ // explicit identity, then propagate the exception only inside that pipe.
+ if (pipeRuntimeException instanceof PipeRuntimeSinkCriticalException) {
+ if (failedPipeMeta != null) {
+ final PipeRuntimeMeta runtimeMeta = failedPipeMeta.getRuntimeMeta();
+ final PipeStaticMeta staticMeta = failedPipeMeta.getStaticMeta();
+ boolean hasLocalTask = false;
+ for (final PipeTaskMeta pipeTaskMeta :
+ runtimeMeta.getConsensusGroupId2TaskMetaMap().values()) {
+ if (pipeTaskMeta.getLeaderNodeId() == currentNodeId
+ && !pipeTaskMeta.containsExceptionMessage(pipeRuntimeException)) {
+ hasLocalTask = true;
+ pipeTaskMeta.trackExceptionMessage(pipeRuntimeException);
+ PipeLogger.log(
+ LOGGER::warn,
+ PipeMessages.PIPE_STOPPED_CRITICAL_EXCEPTION,
+ staticMeta.getPipeName(),
+ staticMeta.getCreationTime(),
+ pipeRuntimeException.getTimeStamp(),
+ staticMeta.getSinkParameters());
+ }
+ }
- runtimeMeta
- .getConsensusGroupId2TaskMetaMap()
- .values()
- .forEach(
- pipeTaskMeta -> {
- if (pipeTaskMeta.getLeaderNodeId() == currentNodeId
- && reusedConnectorParameters2ExceptionMap.containsKey(
- staticMeta.getSinkParameters())
- && !pipeTaskMeta.containsExceptionMessage(
- reusedConnectorParameters2ExceptionMap.get(
- staticMeta.getSinkParameters()))) {
- final PipeRuntimeSinkCriticalException exception =
- reusedConnectorParameters2ExceptionMap.get(
- staticMeta.getSinkParameters());
- pipeTaskMeta.trackExceptionMessage(exception);
- PipeLogger.log(
- LOGGER::warn,
- PipeMessages.PIPE_STOPPED_CRITICAL_EXCEPTION,
- staticMeta.getPipeName(),
- staticMeta.getCreationTime(),
- exception.getTimeStamp(),
- staticMeta.getSinkParameters());
- }
- });
- });
+ if (!hasLocalTask) {
+ // The pipe may have no local region task when the sink callback races with task
+ // removal. The explicit pipe identity still lets us stop only this pipe.
+ stopPipeWithRuntimeException(failedPipeMeta);
+ }
+ }
+ }
// 2. stop all pipes that have critical exceptions.
pipeMetaKeeper
@@ -1094,6 +1112,104 @@ public abstract class PipeTaskAgent {
});
}
+ private void stopPipeWithRuntimeException(final PipeMeta pipeMeta) {
+ final PipeRuntimeMeta runtimeMeta = pipeMeta.getRuntimeMeta();
+ if (runtimeMeta.getStatus().get() != PipeStatus.RUNNING) {
+ return;
+ }
+
+ runtimeMeta.setIsStoppedByRuntimeException(true);
+ final PipeStaticMeta staticMeta = pipeMeta.getStaticMeta();
+ try {
+ stopPipe(staticMeta.getPipeName(), staticMeta.getCreationTime());
+ } finally {
+ // stopPipe intentionally does nothing when its task map has already been removed. Keep the
+ // runtime state consistent with the exception in that race.
+ if (runtimeMeta.getStatus().get() == PipeStatus.RUNNING) {
+ runtimeMeta.getStatus().set(PipeStatus.STOPPED);
+ }
+ }
+ }
+
+ private PipeMeta findPipeMeta(
+ final String pipeName, final long creationTime, final PipeTaskMeta pipeTaskMeta) {
+ if (pipeName != null) {
+ // An explicitly supplied identity is authoritative. Never fall back to task-meta content
+ // matching with a different pipe, since identical task metadata is valid across pipes.
+ return pipeMetaKeeper.getPipeMeta(pipeName, creationTime);
+ }
+
+ for (final PipeMeta pipeMeta : pipeMetaKeeper.getPipeMetaList()) {
+ if (pipeTaskMeta != null
+ && pipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap().values().stream()
+ .anyMatch(taskMeta -> taskMeta == pipeTaskMeta)) {
+ return pipeMeta;
+ }
+ }
+
+ // A task meta may have been deserialized before it reaches this agent. Only use content
+ // matching when it identifies one pipe uniquely; identical progress/leader metadata is valid
+ // for multiple pipes and must not be used to cross their boundaries.
+ PipeMeta matchedPipeMeta = null;
+ if (pipeTaskMeta != null) {
+ for (final PipeMeta pipeMeta : pipeMetaKeeper.getPipeMetaList()) {
+ if (pipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap().values().stream()
+ .anyMatch(pipeTaskMeta::equals)) {
+ if (matchedPipeMeta != null) {
+ return null;
+ }
+ matchedPipeMeta = pipeMeta;
+ }
+ }
+ }
+ return matchedPipeMeta;
+ }
+
+ private PipeTaskMeta findLocalPipeTaskMeta(
+ final PipeMeta pipeMeta, final PipeTaskMeta pipeTaskMeta, final int currentNodeId) {
+ if (pipeMeta == null || pipeTaskMeta == null) {
+ return null;
+ }
+
+ final Collection<PipeTaskMeta> taskMetas =
+ pipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap().values();
+
+ // Object identity is the normal in-process path and is the most precise match.
+ for (final PipeTaskMeta localTaskMeta : taskMetas) {
+ if (localTaskMeta == pipeTaskMeta) {
+ return localTaskMeta;
+ }
+ }
+
+ // Prefer a unique local leader when matching a detached task meta. This avoids recording a
+ // processor failure on a task owned by another DataNode.
+ PipeTaskMeta matchedLocalLeaderTaskMeta = null;
+ for (final PipeTaskMeta localTaskMeta : taskMetas) {
+ if (localTaskMeta.getLeaderNodeId() == currentNodeId && localTaskMeta.equals(pipeTaskMeta)) {
+ if (matchedLocalLeaderTaskMeta != null) {
+ matchedLocalLeaderTaskMeta = null;
+ break;
+ }
+ matchedLocalLeaderTaskMeta = localTaskMeta;
+ }
+ }
+ if (matchedLocalLeaderTaskMeta != null) {
+ return matchedLocalLeaderTaskMeta;
+ }
+
+ // Fall back to a unique content match if the leader metadata is stale or unavailable.
+ PipeTaskMeta matchedTaskMeta = null;
+ for (final PipeTaskMeta localTaskMeta : taskMetas) {
+ if (localTaskMeta.equals(pipeTaskMeta)) {
+ if (matchedTaskMeta != null) {
+ return null;
+ }
+ matchedTaskMeta = localTaskMeta;
+ }
+ }
+ return matchedTaskMeta;
+ }
+
public void collectPipeMetaList(final TPipeHeartbeatReq req, final TPipeHeartbeatResp resp)
throws TException {
if (!tryReadLockWithTimeOutInMs(
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeRuntimeMeta.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeRuntimeMeta.java
index 223eb69c346..1f2e92254f3 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeRuntimeMeta.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeRuntimeMeta.java
@@ -89,8 +89,8 @@ public class PipeRuntimeMeta {
* <p>1. {@link PipeRuntimeCriticalException}, to record the failure of pushing {@link PipeMeta},
* and will result in the halt of pipe execution.
*
- * <p>2. {@link PipeRuntimeSinkCriticalException}, to record the exception reported by other pipes
- * sharing the same connector, and will stop the pipe likewise.
+ * <p>2. {@link PipeRuntimeSinkCriticalException}, retained for compatibility with runtime meta
+ * written before sink subtasks were isolated by pipe.
*/
private final ConcurrentMap<Integer, PipeRuntimeException> nodeId2PipeRuntimeExceptionMap =
new ConcurrentHashMap<>();
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeTaskMeta.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeTaskMeta.java
index e9939d7b2c6..1e860383edd 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeTaskMeta.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeTaskMeta.java
@@ -54,8 +54,8 @@ public class PipeTaskMeta {
* <p>The exceptions are instances of {@link PipeRuntimeCriticalException}, {@link
* PipeRuntimeSinkCriticalException} and {@link PipeRuntimeNonCriticalException}.
*
- * <p>The failure of them, respectively, will lead to the stop of the pipe, the stop of the pipes
- * sharing the same connector, and nothing.
+ * <p>The failure of them, respectively, will lead to the stop of the pipe, the stop of the pipe
+ * that owns the failed sink, and nothing.
*/
private final Set<PipeRuntimeException> exceptionMessages =
Collections.newSetFromMap(new ConcurrentHashMap<>());
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Tag.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Tag.java
index 5fecc6b3122..91b4307e02d 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Tag.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Tag.java
@@ -29,6 +29,7 @@ public enum Tag {
STAGE("stage"),
OPERATION("operation"),
INTERFACE("interface"),
+ PIPE("pipe"),
CREATION_TIME("creation_time"),
INDEX("index"),
MODULE("module"),