This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new 8116ad23cf2 Fix pipe sink isolation between pipes (#18549) (#18558)
8116ad23cf2 is described below
commit 8116ad23cf2898ad2b71f1a68411bedb1196f5c3
Author: Caideyipi <[email protected]>
AuthorDate: Tue Sep 1 15:05:51 2026 +0800
Fix pipe sink isolation between pipes (#18549) (#18558)
* Fix pipe sink isolation between pipes (#18549)
* spotless
---
.../runtime/heartbeat/PipeHeartbeatParser.java | 36 ---
.../runtime/heartbeat/PipeHeartbeatParserTest.java | 42 +++
.../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 | 24 ++
.../task/subtask/sink/PipeSinkSubtaskManager.java | 185 +++++++++++--
.../metric/schema/PipeSchemaRegionSinkMetrics.java | 69 +++--
.../metric/sink/PipeDataRegionSinkMetrics.java | 301 ++++++++-------------
.../protocol/airgap/IoTDBDataRegionAirGapSink.java | 5 +-
.../thrift/async/IoTDBDataRegionAsyncSink.java | 5 +-
.../thrift/sync/IoTDBDataRegionSyncSink.java | 5 +-
.../pipe/agent/task/PipeDataNodeTaskAgentTest.java | 210 ++++++++++++++
.../subtask/sink/PipeSinkSubtaskManagerTest.java | 89 ++++++
.../schema/PipeSchemaRegionSinkMetricsTest.java | 77 ++++++
.../commons/pipe/agent/task/PipeTaskAgent.java | 237 +++++++++++-----
.../pipe/agent/task/meta/PipeRuntimeMeta.java | 4 +-
.../commons/pipe/agent/task/meta/PipeTaskMeta.java | 4 +-
.../iotdb/commons/service/metric/enums/Tag.java | 1 +
19 files changed, 970 insertions(+), 357 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 55bb018f6d0..19b0b73e674 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
@@ -22,7 +22,6 @@ package
org.apache.iotdb.confignode.manager.pipe.coordinator.runtime.heartbeat;
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.meta.PipeMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeRuntimeMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
@@ -30,7 +29,6 @@ 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.agent.task.meta.PipeTemporaryMetaInCoordinator;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
-import org.apache.iotdb.confignode.consensus.response.pipe.task.PipeTableResp;
import org.apache.iotdb.confignode.manager.ConfigManager;
import
org.apache.iotdb.confignode.manager.pipe.resource.PipeConfigNodeResourceManager;
import org.apache.iotdb.confignode.persistence.pipe.PipeTaskInfo;
@@ -271,40 +269,6 @@ public class PipeHeartbeatParser {
exception,
pipeName);
}
-
- if (exception instanceof PipeRuntimeSinkCriticalException) {
- ((PipeTableResp) pipeTaskInfo.get().showPipes())
- .filter(true, pipeName).getAllPipeMeta().stream()
- .filter(pipeMeta ->
!pipeMeta.getStaticMeta().getPipeName().equals(pipeName))
- .map(PipeMeta::getRuntimeMeta)
- .filter(
- runtimeMeta ->
-
!PipeStatus.PRE_DELETE.equals(runtimeMeta.getStatus().get()))
- .filter(
- runtimeMeta ->
!runtimeMeta.getStatus().get().equals(PipeStatus.STOPPED))
- .forEach(
- runtimeMeta -> {
- // 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);
-
- LOGGER.warn(
- String.format(
- "Detect
PipeRuntimeConnectorCriticalException %s "
- + "from agent, stop pipe %s.",
- 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 ebe8cc19573..ccc12244297 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
@@ -22,6 +22,7 @@ package
org.apache.iotdb.confignode.manager.pipe.coordinator.runtime.heartbeat;
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;
@@ -209,6 +210,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/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 235e7c1b0c9..fd65700e503 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
@@ -220,13 +220,25 @@ 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 {
LOGGER.warn("Attempt to report pipe exception to a null PipeTaskMeta.",
pipeRuntimeException);
}
}
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) {
LOGGER.warn(
"Report PipeRuntimeException to local PipeTaskMeta({}), exception
message: {}",
pipeTaskMeta,
@@ -237,7 +249,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 2e4b5e090eb..b4a9e78222a 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 aa87f0850c0..0046a9aded8 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
@@ -66,6 +66,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 int connectorIndex;
@@ -85,7 +86,26 @@ public class PipeSinkSubtask extends PipeAbstractSinkSubtask
{
final int connectorIndex,
final UnboundedBlockingPendingQueue<Event> inputPendingQueue,
final PipeConnector outputPipeConnector) {
+ this(
+ null,
+ taskID,
+ creationTime,
+ attributeSortedString,
+ connectorIndex,
+ inputPendingQueue,
+ outputPipeConnector);
+ }
+
+ public PipeSinkSubtask(
+ final String pipeName,
+ final String taskID,
+ final long creationTime,
+ final String attributeSortedString,
+ final int connectorIndex,
+ final UnboundedBlockingPendingQueue<Event> inputPendingQueue,
+ final PipeConnector outputPipeConnector) {
super(taskID, creationTime, outputPipeConnector);
+ this.pipeName = pipeName;
this.attributeSortedString = attributeSortedString;
this.connectorIndex = connectorIndex;
this.inputPendingQueue = inputPendingQueue;
@@ -396,6 +416,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 072b31e8f8f..dea65d86ee0 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
@@ -47,6 +47,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;
@@ -58,8 +59,8 @@ 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<>();
public synchronized String register(
final Supplier<? extends PipeSinkSubtaskExecutor> executorSupplier,
@@ -87,8 +88,11 @@ public class PipeSinkSubtaskManager {
PipeSinkConstant.CONNECTOR_REALTIME_FIRST_DEFAULT_VALUE);
}
environment.setAttributeSortedString(attributeSortedString);
+ 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);
@@ -107,7 +111,11 @@ public class PipeSinkSubtaskManager {
for (int connectorIndex = 0; connectorIndex < sinkNum; connectorIndex++)
{
final String taskID =
String.format(
- "%s_%s_%s", attributeSortedString,
environment.getCreationTime(), connectorIndex);
+ "%s_%s_%s_%s",
+ environment.getPipeName(),
+ attributeSortedString,
+ environment.getCreationTime(),
+ connectorIndex);
environment.setSinkTaskId(taskID);
final PipeConnector pipeConnector =
isDataRegionSink
@@ -139,6 +147,7 @@ public class PipeSinkSubtaskManager {
// 2. Construct PipeConnectorSubtaskLifeCycle to manage
PipeConnectorSubtask's life cycle
final PipeSinkSubtask pipeSinkSubtask =
new PipeSinkSubtask(
+ environment.getPipeName(),
taskID,
environment.getCreationTime(),
attributeSortedString,
@@ -155,12 +164,11 @@ public class PipeSinkSubtaskManager {
attributeSortedString,
executor.getWorkingThreadName(),
executor.getCallbackThreadName());
- attributeSortedString2SubtaskLifeCycleMap.put(
- attributeSortedString, pipeSinkSubtaskLifeCycleList);
+ pipeSinkSubtaskKey2SubtaskLifeCycleMap.put(pipeSinkSubtaskKey,
pipeSinkSubtaskLifeCycleList);
}
for (final PipeSinkSubtaskLifeCycle lifeCycle :
- attributeSortedString2SubtaskLifeCycleMap.get(attributeSortedString)) {
+ pipeSinkSubtaskKey2SubtaskLifeCycleMap.get(pipeSinkSubtaskKey)) {
lifeCycle.register();
}
@@ -172,12 +180,14 @@ public class PipeSinkSubtaskManager {
final long creationTime,
final int regionId,
final String attributeSortedString) {
- if
(!attributeSortedString2SubtaskLifeCycleMap.containsKey(attributeSortedString))
{
- throw new PipeException(FAILED_TO_DEREGISTER_EXCEPTION_MESSAGE +
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;
@@ -188,7 +198,7 @@ public class PipeSinkSubtaskManager {
lifeCycles.removeIf(o -> o.deregister(committerKey));
if (lifeCycles.isEmpty()) {
- attributeSortedString2SubtaskLifeCycleMap.remove(attributeSortedString);
+ pipeSinkSubtaskKey2SubtaskLifeCycleMap.remove(pipeSinkSubtaskKey);
executor.shutdown();
LOGGER.info(
"The executor {} and {} has been successfully shutdown.",
@@ -199,46 +209,119 @@ public class PipeSinkSubtaskManager {
PipeEventCommitManager.getInstance().deregister(pipeName, creationTime,
regionId);
}
+ 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))
{
- throw new PipeException(FAILED_TO_DEREGISTER_EXCEPTION_MESSAGE +
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))
{
- throw new PipeException(FAILED_TO_DEREGISTER_EXCEPTION_MESSAGE +
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 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(
+ "Failed to get PendingQueue. No such subtask: " +
attributeSortedString);
+ }
+
+ return
pipeSinkSubtaskKey2SubtaskLifeCycleMap.get(pipeSinkSubtaskKey).get(0).getPendingQueue();
+ }
+
+ /**
+ * @deprecated Use {@link #getPipeSinkPendingQueue(String, long, String)} to
identify the pipe
+ * explicitly.
+ */
+ @Deprecated
public UnboundedBlockingPendingQueue<Event> getPipeSinkPendingQueue(
final String attributeSortedString) {
- if
(!attributeSortedString2SubtaskLifeCycleMap.containsKey(attributeSortedString))
{
+ final PipeSinkSubtaskKey pipeSinkSubtaskKey =
+ getUniquePipeSinkSubtaskKey(attributeSortedString);
+ if (pipeSinkSubtaskKey == null) {
throw new PipeException(
"Failed to get PendingQueue. No such subtask: " +
attributeSortedString);
}
// All subtasks share the same pending queue
- return attributeSortedString2SubtaskLifeCycleMap
- .get(attributeSortedString)
- .get(0)
- .getPendingQueue();
+ 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(
@@ -296,6 +379,60 @@ public class PipeSinkSubtaskManager {
return sortedStringSourceMap.toString();
}
+ private void throwNoSuchSubtaskException(final PipeSinkSubtaskKey
pipeSinkSubtaskKey) {
+ throw new PipeException(
+ FAILED_TO_DEREGISTER_EXCEPTION_MESSAGE +
pipeSinkSubtaskKey.attributeSortedString);
+ }
+
+ 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(
+ "Multiple pipes match the requested sink subtask. Use the
pipe-specific "
+ + "PipeSinkSubtaskManager API.");
+ }
+ 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
/////////////////////////
private PipeSinkSubtaskManager() {
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 e8e643aaf32..5ce90a45c2f 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);
}
@@ -127,10 +148,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);
}
@@ -139,22 +157,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 23024424b92..0a01dd2ddb2 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
@@ -31,11 +31,9 @@ import org.apache.iotdb.metrics.utils.MetricLevel;
import org.apache.iotdb.metrics.utils.MetricType;
import com.google.common.collect.ImmutableSet;
-import org.checkerframework.checker.nullness.qual.NonNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
@@ -47,7 +45,7 @@ public class PipeDataRegionSinkMetrics implements IMetricSet {
@SuppressWarnings("java:S3077")
private volatile AbstractMetricService metricService;
- private final Map<String, PipeSinkSubtask> connectorMap = new HashMap<>();
+ private final Map<String, PipeSinkSubtask> connectorMap = new
ConcurrentHashMap<>();
private final Map<String, Rate> tabletRateMap = new ConcurrentHashMap<>();
@@ -75,6 +73,59 @@ public class PipeDataRegionSinkMetrics implements IMetricSet
{
createHistogram(taskID);
}
+ private static String[] getIndexedTags(final PipeSinkSubtask connector) {
+ return connector.getPipeName() == null
+ ? new String[] {
+ Tag.NAME.toString(),
+ connector.getAttributeSortedString(),
+ Tag.INDEX.toString(),
+ String.valueOf(connector.getConnectorIndex()),
+ Tag.CREATION_TIME.toString(),
+ String.valueOf(connector.getCreationTime())
+ }
+ : new String[] {
+ Tag.NAME.toString(),
+ connector.getAttributeSortedString(),
+ Tag.PIPE.toString(),
+ connector.getPipeName(),
+ Tag.INDEX.toString(),
+ String.valueOf(connector.getConnectorIndex()),
+ Tag.CREATION_TIME.toString(),
+ String.valueOf(connector.getCreationTime())
+ };
+ }
+
+ 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 createAutoGauge(final String taskID) {
final PipeSinkSubtask connector = connectorMap.get(taskID);
// Pending event count
@@ -83,80 +134,45 @@ public class PipeDataRegionSinkMetrics implements
IMetricSet {
MetricLevel.IMPORTANT,
connector,
PipeSinkSubtask::getTabletInsertionEventCount,
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(connector.getConnectorIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime()));
+ getIndexedTags(connector));
metricService.createAutoGauge(
Metric.UNTRANSFERRED_TSFILE_COUNT.toString(),
MetricLevel.IMPORTANT,
connector,
PipeSinkSubtask::getTsFileInsertionEventCount,
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(connector.getConnectorIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime()));
+ getIndexedTags(connector));
metricService.createAutoGauge(
Metric.UNTRANSFERRED_HEARTBEAT_COUNT.toString(),
MetricLevel.IMPORTANT,
connector,
PipeSinkSubtask::getPipeHeartbeatEventCount,
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(connector.getConnectorIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime()));
+ getIndexedTags(connector));
// Metrics related to IoTDBThriftAsyncConnector
metricService.createAutoGauge(
Metric.PIPE_ASYNC_CONNECTOR_RETRY_EVENT_QUEUE_SIZE.toString(),
MetricLevel.IMPORTANT,
connector,
PipeSinkSubtask::getAsyncConnectorRetryEventQueueSize,
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(connector.getConnectorIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime()));
+ getIndexedTags(connector));
metricService.createAutoGauge(
Metric.PIPE_PENDING_HANDLERS_SIZE.toString(),
MetricLevel.IMPORTANT,
connector,
PipeSinkSubtask::getPendingHandlersSize,
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(connector.getConnectorIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime()));
+ getIndexedTags(connector));
// Metrics related to IoTDB connector
metricService.createAutoGauge(
Metric.PIPE_TOTAL_UNCOMPRESSED_SIZE.toString(),
MetricLevel.IMPORTANT,
connector,
PipeSinkSubtask::getTotalUncompressedSize,
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(connector.getConnectorIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime()));
+ getIndexedTags(connector));
metricService.createAutoGauge(
Metric.PIPE_TOTAL_COMPRESSED_SIZE.toString(),
MetricLevel.IMPORTANT,
connector,
PipeSinkSubtask::getTotalCompressedSize,
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(connector.getConnectorIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime()));
+ getIndexedTags(connector));
}
private void createRate(final String taskID) {
@@ -167,47 +183,29 @@ public class PipeDataRegionSinkMetrics implements
IMetricSet {
metricService.getOrCreateRate(
Metric.PIPE_CONNECTOR_TABLET_TRANSFER.toString(),
MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(connector.getConnectorIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime())));
+ getIndexedTags(connector)));
tsFileRateMap.put(
taskID,
metricService.getOrCreateRate(
Metric.PIPE_CONNECTOR_TSFILE_TRANSFER.toString(),
MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(connector.getConnectorIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime())));
+ getIndexedTags(connector)));
pipeHeartbeatRateMap.put(
taskID,
metricService.getOrCreateRate(
Metric.PIPE_CONNECTOR_HEARTBEAT_TRANSFER.toString(),
MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(connector.getConnectorIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime())));
+ getIndexedTags(connector)));
}
private void createTimer(final String taskID) {
final PipeSinkSubtask connector = connectorMap.get(taskID);
- compressionTimerMap.putIfAbsent(
- connector.getAttributeSortedString(),
+ compressionTimerMap.put(
+ taskID,
metricService.getOrCreateTimer(
Metric.PIPE_COMPRESSION_TIME.toString(),
MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime())));
+ getCreationTimeTags(connector)));
}
private void createHistogram(final String taskID) {
@@ -217,48 +215,35 @@ public class PipeDataRegionSinkMetrics implements
IMetricSet {
metricService.getOrCreateHistogram(
Metric.PIPE_INSERT_NODE_BATCH_SIZE.toString(),
MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime()));
+ getCreationTimeTags(connector));
connector.setTabletBatchSizeHistogram(tabletBatchSizeHistogram);
final Histogram tsFileBatchSizeHistogram =
metricService.getOrCreateHistogram(
Metric.PIPE_TSFILE_BATCH_SIZE.toString(),
MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime()));
+ getCreationTimeTags(connector));
connector.setTsFileBatchSizeHistogram(tsFileBatchSizeHistogram);
final Histogram tabletBatchTimeIntervalHistogram =
metricService.getOrCreateHistogram(
Metric.PIPE_INSERT_NODE_BATCH_TIME_COST.toString(),
MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime()));
+ getCreationTimeTags(connector));
connector.setTabletBatchTimeIntervalHistogram(tabletBatchTimeIntervalHistogram);
final Histogram tsFileBatchTimeIntervalHistogram =
metricService.getOrCreateHistogram(
Metric.PIPE_TSFILE_BATCH_TIME_COST.toString(),
MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime()));
+ getCreationTimeTags(connector));
connector.setTsFileBatchTimeIntervalHistogram(tsFileBatchTimeIntervalHistogram);
- Histogram eventSizeHistogram =
+ final Histogram eventSizeHistogram =
metricService.getOrCreateHistogram(
Metric.PIPE_CONNECTOR_BATCH_SIZE.toString(),
MetricLevel.IMPORTANT,
- Tag.NAME.toString(),
- connector.getAttributeSortedString());
+ getNameTags(connector));
connector.setEventSizeHistogram(eventSizeHistogram);
}
@@ -287,68 +272,33 @@ public class PipeDataRegionSinkMetrics implements
IMetricSet {
metricService.remove(
MetricType.AUTO_GAUGE,
Metric.UNTRANSFERRED_TABLET_COUNT.toString(),
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(connector.getConnectorIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime()));
+ getIndexedTags(connector));
metricService.remove(
MetricType.AUTO_GAUGE,
Metric.UNTRANSFERRED_TSFILE_COUNT.toString(),
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(connector.getConnectorIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime()));
+ getIndexedTags(connector));
metricService.remove(
MetricType.AUTO_GAUGE,
Metric.UNTRANSFERRED_HEARTBEAT_COUNT.toString(),
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(connector.getConnectorIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime()));
+ getIndexedTags(connector));
// Metrics related to IoTDBThriftAsyncConnector
metricService.remove(
MetricType.AUTO_GAUGE,
Metric.PIPE_ASYNC_CONNECTOR_RETRY_EVENT_QUEUE_SIZE.toString(),
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(connector.getConnectorIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime()));
+ getIndexedTags(connector));
metricService.remove(
MetricType.AUTO_GAUGE,
Metric.PIPE_PENDING_HANDLERS_SIZE.toString(),
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(connector.getConnectorIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime()));
+ getIndexedTags(connector));
// Metrics related to IoTDB connector
metricService.remove(
MetricType.AUTO_GAUGE,
Metric.PIPE_TOTAL_UNCOMPRESSED_SIZE.toString(),
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(connector.getConnectorIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime()));
+ getIndexedTags(connector));
metricService.remove(
MetricType.AUTO_GAUGE,
Metric.PIPE_TOTAL_COMPRESSED_SIZE.toString(),
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(connector.getConnectorIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime()));
+ getIndexedTags(connector));
}
private void removeRate(final String taskID) {
@@ -357,30 +307,15 @@ public class PipeDataRegionSinkMetrics implements
IMetricSet {
metricService.remove(
MetricType.RATE,
Metric.PIPE_CONNECTOR_TABLET_TRANSFER.toString(),
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(connector.getConnectorIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime()));
+ getIndexedTags(connector));
metricService.remove(
MetricType.RATE,
Metric.PIPE_CONNECTOR_TSFILE_TRANSFER.toString(),
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(connector.getConnectorIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime()));
+ getIndexedTags(connector));
metricService.remove(
MetricType.RATE,
Metric.PIPE_CONNECTOR_HEARTBEAT_TRANSFER.toString(),
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.INDEX.toString(),
- String.valueOf(connector.getConnectorIndex()),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime()));
+ getIndexedTags(connector));
tabletRateMap.remove(taskID);
tsFileRateMap.remove(taskID);
pipeHeartbeatRateMap.remove(taskID);
@@ -389,13 +324,8 @@ public class PipeDataRegionSinkMetrics implements
IMetricSet {
private void removeTimer(final String taskID) {
final PipeSinkSubtask connector = connectorMap.get(taskID);
metricService.remove(
- MetricType.TIMER,
- Metric.PIPE_COMPRESSION_TIME.toString(),
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime()));
- compressionTimerMap.remove(connector.getAttributeSortedString());
+ MetricType.TIMER, Metric.PIPE_COMPRESSION_TIME.toString(),
getCreationTimeTags(connector));
+ compressionTimerMap.remove(taskID);
}
private void removeHistogram(final String taskID) {
@@ -403,42 +333,27 @@ public class PipeDataRegionSinkMetrics implements
IMetricSet {
metricService.remove(
MetricType.HISTOGRAM,
Metric.PIPE_INSERT_NODE_BATCH_SIZE.toString(),
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime()));
+ getCreationTimeTags(connector));
metricService.remove(
MetricType.HISTOGRAM,
Metric.PIPE_TSFILE_BATCH_SIZE.toString(),
- Tag.NAME.toString(),
- connector.getAttributeSortedString(),
- Tag.CREATION_TIME.toString(),
- String.valueOf(connector.getCreationTime()));
+ getCreationTimeTags(connector));
metricService.remove(
MetricType.HISTOGRAM,
Metric.PIPE_INSERT_NODE_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_TSFILE_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)
////////////////////////////
- public void register(@NonNull final PipeSinkSubtask pipeSinkSubtask) {
+ public void register(final PipeSinkSubtask pipeSinkSubtask) {
final String taskID = pipeSinkSubtask.getTaskID();
connectorMap.putIfAbsent(taskID, pipeSinkSubtask);
if (Objects.nonNull(metricService)) {
@@ -449,7 +364,7 @@ public class PipeDataRegionSinkMetrics implements
IMetricSet {
public void deregister(final String taskID) {
if (!connectorMap.containsKey(taskID)) {
LOGGER.warn(
- "Failed to deregister pipe data region connector metrics,
PipeConnectorSubtask({}) does not exist",
+ "Failed to deregister pipe data region sink metrics,
PipeSinkSubtask({}) does not exist",
taskID);
return;
}
@@ -466,7 +381,7 @@ public class PipeDataRegionSinkMetrics implements
IMetricSet {
final Rate rate = tabletRateMap.get(taskID);
if (rate == null) {
LOGGER.info(
- "Failed to mark pipe data region connector tablet event,
PipeConnectorSubtask({}) does not exist",
+ "Failed to mark pipe data region sink tablet event,
PipeSinkSubtask({}) does not exist",
taskID);
return;
}
@@ -480,7 +395,7 @@ public class PipeDataRegionSinkMetrics implements
IMetricSet {
final Rate rate = tsFileRateMap.get(taskID);
if (rate == null) {
LOGGER.info(
- "Failed to mark pipe data region connector tsfile event,
PipeConnectorSubtask({}) does not exist",
+ "Failed to mark pipe data region sink tsfile event,
PipeSinkSubtask({}) does not exist",
taskID);
return;
}
@@ -499,8 +414,32 @@ public class PipeDataRegionSinkMetrics implements
IMetricSet {
rate.mark();
}
- public Timer getCompressionTimer(final String attributeSortedString) {
- return Objects.isNull(metricService) ? null :
compressionTimerMap.get(attributeSortedString);
+ public Timer getCompressionTimer(final String taskID) {
+ if (Objects.isNull(metricService)) {
+ return null;
+ }
+ final Timer timer = compressionTimerMap.get(taskID);
+ if (timer != null) {
+ return timer;
+ }
+
+ // Keep compatibility with older sinks that pass the attribute string.
Only return a
+ // fallback when it identifies one subtask; never leak a timer across
pipes with equal sinks.
+ Timer matchedTimer = null;
+ for (final Map.Entry<String, PipeSinkSubtask> entry :
connectorMap.entrySet()) {
+ if (!Objects.equals(entry.getValue().getAttributeSortedString(),
taskID)) {
+ continue;
+ }
+ final Timer candidate = compressionTimerMap.get(entry.getKey());
+ if (candidate == null) {
+ continue;
+ }
+ if (matchedTimer != null) {
+ return null;
+ }
+ matchedTimer = candidate;
+ }
+ return matchedTimer;
}
//////////////////////////// singleton ////////////////////////////
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java
index 535f2b2ed41..ade00324084 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java
@@ -581,9 +581,8 @@ public class IoTDBDataRegionAirGapSink extends
IoTDBDataNodeAirGapSink {
@Override
protected byte[] compressIfNeeded(final byte[] reqInBytes) throws
IOException {
- if (Objects.isNull(compressionTimer) &&
Objects.nonNull(attributeSortedString)) {
- compressionTimer =
-
PipeDataRegionSinkMetrics.getInstance().getCompressionTimer(attributeSortedString);
+ if (Objects.isNull(compressionTimer) && Objects.nonNull(sinkTaskId)) {
+ compressionTimer =
PipeDataRegionSinkMetrics.getInstance().getCompressionTimer(sinkTaskId);
}
return super.compressIfNeeded(reqInBytes);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
index cd780d963d0..4772fdfe2bc 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
@@ -513,9 +513,8 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink {
@Override
public TPipeTransferReq compressIfNeeded(final TPipeTransferReq req) throws
IOException {
- if (Objects.isNull(compressionTimer) &&
Objects.nonNull(attributeSortedString)) {
- compressionTimer =
-
PipeDataRegionSinkMetrics.getInstance().getCompressionTimer(attributeSortedString);
+ if (Objects.isNull(compressionTimer) && Objects.nonNull(sinkTaskId)) {
+ compressionTimer =
PipeDataRegionSinkMetrics.getInstance().getCompressionTimer(sinkTaskId);
}
return super.compressIfNeeded(req);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java
index e5e1aca2245..728824b4cd9 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java
@@ -682,9 +682,8 @@ public class IoTDBDataRegionSyncSink extends
IoTDBDataNodeSyncSink {
@Override
public TPipeTransferReq compressIfNeeded(final TPipeTransferReq req) throws
IOException {
- if (Objects.isNull(compressionTimer) &&
Objects.nonNull(attributeSortedString)) {
- compressionTimer =
-
PipeDataRegionSinkMetrics.getInstance().getCompressionTimer(attributeSortedString);
+ if (Objects.isNull(compressionTimer) && Objects.nonNull(sinkTaskId)) {
+ compressionTimer =
PipeDataRegionSinkMetrics.getInstance().getCompressionTimer(sinkTaskId);
}
return super.compressIfNeeded(req);
}
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 3dd93e87789..a919810cc04 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,21 +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 {
@@ -80,6 +91,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("failedPipe", failedPipeMeta);
+ pipeMetaKeeper.addPipeMeta("unaffectedPipe", 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("existingPipe", 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("firstPipe", firstPipeMeta);
+ pipeMetaKeeper.addPipeMeta("secondPipe", 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("detachedPipe", 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("nullTaskMetaPipe", failedPipeMeta);
+ pipeMetaKeeper.addPipeMeta("stillRunningPipe", 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);
@@ -227,4 +429,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 21cd8369b01..e2324f5c95b 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
@@ -51,7 +51,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;
@@ -1009,6 +1008,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 associated with a critical exception. The explicit pipe
identity is important
+ * for events whose task meta was deserialized and is therefore not the same
object as the local
+ * task meta.
+ */
+ 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(
() -> {
@@ -1017,8 +1031,40 @@ public abstract class PipeTaskAgent {
while (true) {
if (tryWriteLockWithTimeOut(5)) {
try {
- pipeTaskMeta.trackExceptionMessage(pipeRuntimeException);
- stopAllPipesWithCriticalExceptionInternal(currentNodeId);
+ final PipeMeta failedPipeMeta =
+ findPipeMeta(pipeName, creationTime, pipeTaskMeta);
+
+ // An explicit identity is authoritative. A stale callback
must not be mapped to
+ // another pipe merely because its serialized task metadata
happens to match.
+ if (pipeName != null && failedPipeMeta == null) {
+ return;
+ }
+
+ final PipeTaskMeta localFailedPipeTaskMeta =
+ findLocalPipeTaskMeta(failedPipeMeta, pipeTaskMeta,
currentNodeId);
+ if (failedPipeMeta == null) {
+ // Preserve the legacy behavior for a caller that supplied
only a task meta.
+ // If it is detached and ambiguous, it is deliberately not
attached to any
+ // local pipe.
+ if (pipeTaskMeta != null) {
+ pipeTaskMeta.trackExceptionMessage(pipeRuntimeException);
+ }
+ } else if (localFailedPipeTaskMeta != null) {
+
localFailedPipeTaskMeta.trackExceptionMessage(pipeRuntimeException);
+ }
+
+ if (failedPipeMeta != null
+ && failedPipeMeta.getRuntimeMeta().getStatus().get() ==
PipeStatus.RUNNING) {
+
failedPipeMeta.getRuntimeMeta().setIsStoppedByRuntimeException(true);
+ }
+
+ stopAllPipesWithCriticalExceptionInternal(
+ currentNodeId, failedPipeMeta, pipeRuntimeException);
+ if (failedPipeMeta != null) {
+ // stopPipe() can race with task removal and return
without changing the
+ // runtime status. The identified pipe still needs to be
marked stopped.
+ stopPipeWithRuntimeException(failedPipeMeta);
+ }
LOGGER.info("Stopped all pipes with critical exception.");
return;
} finally {
@@ -1046,72 +1092,34 @@ 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.getConnectorParameters(),
- (PipeRuntimeSinkCriticalException) e);
- }
- }
- });
- });
- pipeMetaKeeper
- .getPipeMetaList()
- .forEach(
- pipeMeta -> {
- final PipeStaticMeta staticMeta = pipeMeta.getStaticMeta();
- final PipeRuntimeMeta runtimeMeta = pipeMeta.getRuntimeMeta();
-
- runtimeMeta
- .getConsensusGroupId2TaskMetaMap()
- .values()
- .forEach(
- pipeTaskMeta -> {
- if (pipeTaskMeta.getLeaderNodeId() == currentNodeId
- &&
reusedConnectorParameters2ExceptionMap.containsKey(
- staticMeta.getConnectorParameters())
- && !pipeTaskMeta.containsExceptionMessage(
- reusedConnectorParameters2ExceptionMap.get(
- staticMeta.getConnectorParameters()))) {
- final PipeRuntimeSinkCriticalException exception =
- reusedConnectorParameters2ExceptionMap.get(
- staticMeta.getConnectorParameters());
- pipeTaskMeta.trackExceptionMessage(exception);
- PipeLogger.log(
- LOGGER::warn,
- "Pipe %s (creation time = %s) will be stopped
because of critical exception "
- + "(occurred time %s) in connector %s.",
- staticMeta.getPipeName(),
- staticMeta.getCreationTime(),
- exception.getTimeStamp(),
- staticMeta.getConnectorParameters());
- }
- });
- });
+ private void stopAllPipesWithCriticalExceptionInternal(
+ final int currentNodeId,
+ final PipeMeta failedPipeMeta,
+ final PipeRuntimeException pipeRuntimeException) {
+ // A sink subtask is shared only by regions belonging to one pipe.
Propagate its exception only
+ // to the other local leader tasks of that same pipe.
+ if (pipeRuntimeException instanceof PipeRuntimeSinkCriticalException
+ && failedPipeMeta != null) {
+ final PipeStaticMeta staticMeta = failedPipeMeta.getStaticMeta();
+ for (final PipeTaskMeta pipeTaskMeta :
+
failedPipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap().values()) {
+ if (pipeTaskMeta.getLeaderNodeId() == currentNodeId
+ && !pipeTaskMeta.containsExceptionMessage(pipeRuntimeException)) {
+ pipeTaskMeta.trackExceptionMessage(pipeRuntimeException);
+ PipeLogger.log(
+ LOGGER::warn,
+ "Pipe %s (creation time = %s) will be stopped because of
critical exception "
+ + "(occurred time %s) in connector %s.",
+ staticMeta.getPipeName(),
+ staticMeta.getCreationTime(),
+ pipeRuntimeException.getTimeStamp(),
+ staticMeta.getConnectorParameters());
+ }
+ }
+ }
- // 2. stop all pipes that have critical exceptions.
+ // Stop every pipe that already has a critical exception. Sink exceptions
added above are
+ // intentionally visible only inside failedPipeMeta.
pipeMetaKeeper
.getPipeMetaList()
.forEach(
@@ -1143,6 +1151,99 @@ 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() can find no local task map during a drop race. Keep the
runtime metadata
+ // consistent with the critical exception in that case.
+ 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) {
+ final PipeMeta pipeMeta = pipeMetaKeeper.getPipeMeta(pipeName);
+ return pipeMeta != null && pipeMeta.getStaticMeta().getCreationTime() ==
creationTime
+ ? pipeMeta
+ : null;
+ }
+
+ // Object identity is the precise in-process path.
+ for (final PipeMeta pipeMeta : pipeMetaKeeper.getPipeMetaList()) {
+ if (pipeTaskMeta != null
+ &&
pipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap().values().stream()
+ .anyMatch(taskMeta -> taskMeta == pipeTaskMeta)) {
+ return pipeMeta;
+ }
+ }
+
+ // A serialized task meta is safe only when it identifies one pipe
uniquely. Equal task metas
+ // can legitimately occur in multiple pipes, so ambiguous matches are
rejected.
+ 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();
+ for (final PipeTaskMeta localTaskMeta : taskMetas) {
+ if (localTaskMeta == pipeTaskMeta) {
+ return localTaskMeta;
+ }
+ }
+
+ PipeTaskMeta matchedLocalLeaderTaskMeta = null;
+ for (final PipeTaskMeta localTaskMeta : taskMetas) {
+ if (localTaskMeta.getLeaderNodeId() == currentNodeId &&
localTaskMeta.equals(pipeTaskMeta)) {
+ if (matchedLocalLeaderTaskMeta != null) {
+ return null;
+ }
+ matchedLocalLeaderTaskMeta = localTaskMeta;
+ }
+ }
+ if (matchedLocalLeaderTaskMeta != null) {
+ return matchedLocalLeaderTaskMeta;
+ }
+
+ 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 63aa7b86c43..76849d86ef1 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
@@ -88,8 +88,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 f3062737d92..0bc05e2f301 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");
