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 <[email protected]>
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"),
