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 595aa8442bb Fix write-back sink history failure state handling (#19056)
595aa8442bb is described below
commit 595aa8442bb50fc8bfafdb8f11d40a94e1b3c1bd
Author: Caideyipi <[email protected]>
AuthorDate: Mon Nov 2 10:08:10 2026 +0800
Fix write-back sink history failure state handling (#19056)
* Fix write-back sink history failure state handling
* Fix write-back sink auto-split test credentials
* Fix pipe metadata retry persistence
---
.../treemodel/auto/basic/IoTDBPipeAutoSplitIT.java | 56 +++++
.../parameter/PipeParameterValidator.java | 24 ++-
.../parameter/PipeParameterValidatorTest.java | 126 +++++++++++
.../iotdb/confignode/i18n/ProcedureMessages.java | 2 +
.../iotdb/confignode/i18n/ProcedureMessages.java | 2 +
.../confignode/persistence/pipe/PipeTaskInfo.java | 121 ++++++++---
.../impl/pipe/AbstractOperatePipeProcedureV2.java | 80 +++++--
.../impl/pipe/runtime/PipeMetaSyncProcedure.java | 4 +
.../impl/pipe/task/AlterPipeProcedureV2.java | 9 +-
.../impl/pipe/task/CreatePipeProcedureV2.java | 12 +-
.../impl/pipe/task/StartPipeProcedureV2.java | 59 +++--
.../consensus/response/pipe/PipeTableRespTest.java | 20 ++
.../pipe/PipeTaskInfoAutoRestartTest.java | 54 +++++
.../pipe/AbstractOperatePipeProcedureV2Test.java | 24 +++
.../pipe/runtime/PipeMetaSyncProcedureTest.java | 240 +++++++++++++++++++++
.../impl/pipe/task/CreatePipeProcedureV2Test.java | 146 +++++++++++++
.../impl/pipe/task/StartPipeProcedureV2Test.java | 217 +++++++++++++++++++
.../apache/iotdb/db/pipe/sink/PipeSinkTest.java | 31 +++
.../config/sys/pipe/ShowPipeTaskTest.java | 7 +
19 files changed, 1143 insertions(+), 91 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeAutoSplitIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeAutoSplitIT.java
index 4b9012b88ab..c737ce32ec7 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeAutoSplitIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeAutoSplitIT.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.pipe.it.dual.treemodel.auto.basic;
import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipeInfo;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq;
import org.apache.iotdb.consensus.ConsensusFactory;
@@ -42,6 +43,7 @@ import java.sql.SQLException;
import java.sql.Statement;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.Objects;
@@ -82,6 +84,60 @@ public class IoTDBPipeAutoSplitIT extends
AbstractPipeDualTreeModelAutoIT {
receiverEnv.initClusterEnvironment();
}
+ @Test
+ public void testWriteBackSinkWithTargetDatabase() throws Exception {
+ TestUtils.executeNonQueries(
+ senderEnv,
+ Arrays.asList(
+ "create database root.writeback_source",
+ "create aligned timeseries root.writeback_source.d1(s1 INT32)",
+ "insert into root.writeback_source.d1(time,s1) aligned values "
+ + "(1,1),(2,2),(3,3),(4,4),(5,5)",
+ "flush",
+ "create pipe writeback with source ("
+ +
"'path'='root.writeback_source.**','inclusion'='data.insert',"
+ + "'forwarding-pipe-requests'='false',"
+ + "'history.enable'='true','realtime.enable'='true') "
+ + "with sink
('sink'='write-back-sink','sink.database'='root.writeback_target',"
+ + "'user'='root','password'='root')"),
+ null);
+
+ TestUtils.assertDataEventuallyOnEnv(
+ senderEnv,
+ "select s1 from root.writeback_target.d1",
+ "Time,root.writeback_target.d1.s1,",
+ new HashSet<>(Arrays.asList("1,1,", "2,2,", "3,3,", "4,4,", "5,5,")),
+ 90);
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient)
senderEnv.getLeaderConfigNodeConnection()) {
+ final List<TShowPipeInfo> pipeInfoList =
+ client.showPipe(new
TShowPipeReq().setUserName(SessionConfig.DEFAULT_USER)).pipeInfoList;
+ final TShowPipeInfo realtimePipe =
+ pipeInfoList.stream()
+ .filter(pipeInfo -> "writeback_realtime".equals(pipeInfo.id))
+ .findFirst()
+ .orElseThrow(AssertionError::new);
+ Assert.assertEquals(PipeStatus.RUNNING.name(), realtimePipe.getState());
+ Assert.assertEquals("", realtimePipe.getExceptionMessage());
+ }
+
+ TestUtils.executeNonQueries(
+ senderEnv,
+ Collections.singletonList(
+ "insert into root.writeback_source.d1(time,s1) aligned values "
+ + "(6,6),(7,7),(8,8),(9,9)"),
+ null);
+
+ TestUtils.assertDataEventuallyOnEnv(
+ senderEnv,
+ "select s1 from root.writeback_target.d1",
+ "Time,root.writeback_target.d1.s1,",
+ new HashSet<>(
+ Arrays.asList("1,1,", "2,2,", "3,3,", "4,4,", "5,5,", "6,6,",
"7,7,", "8,8,", "9,9,")),
+ 90);
+ }
+
@Test
public void testSingleEnv() throws Exception {
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
diff --git
a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameterValidator.java
b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameterValidator.java
index 57894ed244b..3e0562bf453 100644
---
a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameterValidator.java
+++
b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameterValidator.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.pipe.api.i18n.PipeApiMessages;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -58,9 +59,14 @@ public class PipeParameterValidator {
final List<String> lhsAttributes,
final List<String> rhsAttributes,
final boolean isRequired) {
- final boolean lhsExistence =
lhsAttributes.stream().anyMatch(parameters::hasAttribute);
- final boolean rhsExistence =
rhsAttributes.stream().anyMatch(parameters::hasAttribute);
- if (lhsExistence && rhsExistence) {
+ final Set<String> lhsExistingKeys =
getExistingAttributeKeys(lhsAttributes);
+ final Set<String> rhsExistingKeys =
getExistingAttributeKeys(rhsAttributes);
+ final boolean lhsExistence = !lhsExistingKeys.isEmpty();
+ final boolean rhsExistence = !rhsExistingKeys.isEmpty();
+ // One unprefixed key may match both sides after prefix reduction. It is
still one attribute.
+ if (lhsExistence
+ && rhsExistence
+ && !(lhsExistingKeys.size() == 1 &&
lhsExistingKeys.equals(rhsExistingKeys))) {
throw new PipeParameterNotValidException(
String.format(
PipeApiMessages.EXCEPTION_CANNOT_SPECIFY_BOTH_ARG_AND_ARG_AT_THE_SAME_TIME_7DA8858B,
@@ -77,6 +83,18 @@ public class PipeParameterValidator {
return this;
}
+ private Set<String> getExistingAttributeKeys(final List<String>
attributeKeys) {
+ return attributeKeys.stream()
+ .flatMap(
+ key ->
+ Stream.of(
+ key,
+ PipeParameters.KeyReducer.shallowReduce(key),
+ PipeParameters.KeyReducer.reduce(key)))
+ .filter(parameters.getAttribute()::containsKey)
+ .collect(Collectors.toSet());
+ }
+
/**
* Validates whether the attributes entered by the user contain an attribute
whose key is
* attributeKey.
diff --git
a/iotdb-api/pipe-api/src/test/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameterValidatorTest.java
b/iotdb-api/pipe-api/src/test/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameterValidatorTest.java
new file mode 100644
index 00000000000..5c5e6a1a07a
--- /dev/null
+++
b/iotdb-api/pipe-api/src/test/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameterValidatorTest.java
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.pipe.api.customizer.parameter;
+
+import org.apache.iotdb.pipe.api.exception.PipeAttributeNotProvidedException;
+import org.apache.iotdb.pipe.api.exception.PipeParameterNotValidException;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+public class PipeParameterValidatorTest {
+
+ @Test
+ public void testDatabaseAliasesMatchOneAttribute() {
+ for (final String key : Arrays.asList("connector.database",
"sink.database", "database")) {
+ final PipeParameterValidator validator =
+ new PipeParameterValidator(
+ new PipeParameters(Collections.singletonMap(key,
"root.target")));
+ for (final boolean required : new boolean[] {false, true}) {
+ Assert.assertSame(
+ validator,
+ validator.validateSynonymAttributes(
+ Collections.singletonList("connector.database"),
+ Collections.singletonList("sink.database"),
+ required));
+ }
+ }
+ }
+
+ @Test
+ public void testDatabaseAliasesRejectDistinctAttributes() {
+ final String[] keys = {"connector.database", "sink.database", "database"};
+ for (int i = 0; i < keys.length; i++) {
+ for (int j = i + 1; j < keys.length; j++) {
+ final Map<String, String> attributes = new HashMap<>();
+ attributes.put(keys[i], "root.target");
+ attributes.put(keys[j], "root.target");
+ Assert.assertThrows(
+ PipeParameterNotValidException.class,
+ () ->
+ new PipeParameterValidator(new PipeParameters(attributes))
+ .validateSynonymAttributes(
+ Collections.singletonList("connector.database"),
+ Collections.singletonList("sink.database"),
+ false));
+ }
+ }
+ }
+
+ @Test
+ public void testDistinctSynonymsRemainMutuallyExclusive() {
+ for (final String userKey : Arrays.asList("connector.user", "sink.user",
"user")) {
+ for (final String usernameKey :
+ Arrays.asList("connector.username", "sink.username", "username")) {
+ final Map<String, String> attributes = new HashMap<>();
+ attributes.put(userKey, "user");
+ attributes.put(usernameKey, "user");
+ Assert.assertThrows(
+ PipeParameterNotValidException.class,
+ () ->
+ new PipeParameterValidator(new PipeParameters(attributes))
+ .validateSynonymAttributes(
+ Arrays.asList("connector.user", "sink.user"),
+ Arrays.asList("connector.username", "sink.username"),
+ false));
+ }
+ }
+ }
+
+ @Test
+ public void testShallowPrefixesMatchOneAttribute() {
+ new PipeParameterValidator(
+ new PipeParameters(Collections.singletonMap("opcua.database",
"root.target")))
+ .validateSynonymAttributes(
+ Collections.singletonList("connector.opcua.database"),
+ Collections.singletonList("sink.opcua.database"),
+ true);
+ }
+
+ @Test
+ public void testRequiredSynonymsSupportUnprefixedAttributes() {
+ for (final String key : Arrays.asList("output.series", "output-series")) {
+ new PipeParameterValidator(new
PipeParameters(Collections.singletonMap(key, "root.db.s")))
+ .validateSynonymAttributes(
+ Collections.singletonList("processor.output.series"),
+ Collections.singletonList("processor.output-series"),
+ true);
+ }
+
+ final PipeParameterValidator validator =
+ new PipeParameterValidator(new PipeParameters(Collections.emptyMap()));
+ validator.validateSynonymAttributes(
+ Collections.singletonList("connector.database"),
+ Collections.singletonList("sink.database"),
+ false);
+ Assert.assertThrows(
+ PipeAttributeNotProvidedException.class,
+ () ->
+ validator.validateSynonymAttributes(
+ Collections.singletonList("connector.database"),
+ Collections.singletonList("sink.database"),
+ true));
+ }
+}
diff --git
a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
index 46a278848c9..79b02325d7a 100644
---
a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
+++
b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
@@ -454,6 +454,8 @@ public final class ProcedureMessages {
"Failed to push pipe meta list to data nodes, will retry later.";
public static final String FAILED_TO_PUSH_PIPE_META_TO_DATANODES_DETAILS =
"Failed to push pipe meta to dataNodes, details: %s";
+ public static final String
EXCEPTION_INTERNAL_ERROR_WHILE_PROCESSING_PUSHPIPEMETA_ON_DATANODE_5983BD5F =
+ "Internal error while processing pushPipeMeta on DataNode.";
public static final String FAILED_TO_PUSH_TOPIC_META_TO_DATANODES_DETAILS =
"Failed to push topic meta to dataNodes, details: %s";
public static final String FAILED_TO_REMOVE_DATA_NODE_BECAUSE_IT_IS_NOT_IN =
diff --git
a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
index e972e688ae6..c10a30c65f5 100644
---
a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
+++
b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
@@ -439,6 +439,8 @@ public final class ProcedureMessages {
"向 DataNode 推送 pipe 元数据列表失败，将稍后重试。";
public static final String FAILED_TO_PUSH_PIPE_META_TO_DATANODES_DETAILS =
"向 DataNode 推送 pipe 元数据失败，详情：%s";
+ public static final String
EXCEPTION_INTERNAL_ERROR_WHILE_PROCESSING_PUSHPIPEMETA_ON_DATANODE_5983BD5F =
+ "处理 DataNode 上的 pushPipeMeta 时发生内部错误。";
public static final String FAILED_TO_PUSH_TOPIC_META_TO_DATANODES_DETAILS =
"向 DataNode 推送 topic 元数据失败，详情：%s";
public static final String FAILED_TO_REMOVE_DATA_NODE_BECAUSE_IT_IS_NOT_IN =
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
index 2427e7f3329..78040be82d5 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
@@ -55,6 +55,7 @@ import
org.apache.iotdb.confignode.consensus.request.write.pipe.task.SetPipeStat
import
org.apache.iotdb.confignode.consensus.request.write.pipe.task.SetPipeStatusWithStoppedByRuntimeExceptionPlanV2;
import org.apache.iotdb.confignode.consensus.response.pipe.task.PipeTableResp;
import org.apache.iotdb.confignode.i18n.ConfigNodeMessages;
+import org.apache.iotdb.confignode.i18n.ProcedureMessages;
import
org.apache.iotdb.confignode.manager.pipe.resource.PipeConfigNodeResourceManager;
import
org.apache.iotdb.confignode.procedure.impl.pipe.runtime.PipeHandleMetaChangeProcedure;
import org.apache.iotdb.confignode.rpc.thrift.TAlterPipeReq;
@@ -97,6 +98,9 @@ public class PipeTaskInfo implements SnapshotProcessor {
private final PipeMetaKeeper pipeMetaKeeper;
+ // Tracks runtime exception changes that have not yet been persisted through
consensus.
+ private volatile boolean pipeMetaExceptionPersistencePending;
+
// Pure in-memory object, not involved in snapshot serialization and
deserialization.
private final PipeTaskInfoVersion pipeTaskInfoVersion;
// Accepts a username and returns its current stored password for pipe
authentication.
@@ -1138,22 +1142,43 @@ public class PipeTaskInfo implements SnapshotProcessor {
*
* <p>If there are exceptions recorded, the related pipes will be stopped,
and the exception
* messages will then be updated to all the nodes through {@link
PipeHandleMetaChangeProcedure}.
+ * Failures without a pipe identity apply to every pipe in the full metadata
sync.
*
* @param respMap The responseMap after pushing pipe meta
* @return {@code true} if there are exceptions encountered
*/
public boolean recordDataNodePushPipeMetaExceptions(
final Map<Integer, TPushPipeMetaResp> respMap) {
+ return recordDataNodePushPipeMetaExceptions(respMap, null);
+ }
+
+ /**
+ * Record push failures locally, using the given pipe meta when a DataNode
cannot identify the
+ * pipe in its response.
+ */
+ public boolean recordDataNodePushPipeMetaExceptions(
+ final Map<Integer, TPushPipeMetaResp> respMap, final PipeStaticMeta
fallbackPipeStaticMeta) {
acquireWriteLock();
try {
- return recordDataNodePushPipeMetaExceptionsInternal(respMap);
+ final boolean hasException =
+ recordDataNodePushPipeMetaExceptionsInternal(respMap,
fallbackPipeStaticMeta);
+ pipeMetaExceptionPersistencePending |= hasException;
+ return hasException;
} finally {
releaseWriteLock();
}
}
+ public boolean hasPendingPipeMetaExceptionPersistence() {
+ return pipeMetaExceptionPersistencePending;
+ }
+
+ public void markPipeMetaExceptionsPersisted() {
+ pipeMetaExceptionPersistencePending = false;
+ }
+
private boolean recordDataNodePushPipeMetaExceptionsInternal(
- final Map<Integer, TPushPipeMetaResp> respMap) {
+ final Map<Integer, TPushPipeMetaResp> respMap, final PipeStaticMeta
fallbackPipeStaticMeta) {
boolean hasException = false;
for (final Map.Entry<Integer, TPushPipeMetaResp> respEntry :
respMap.entrySet()) {
@@ -1163,46 +1188,36 @@ public class PipeTaskInfo implements SnapshotProcessor {
if (resp.getStatus().getCode() ==
TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode()) {
hasException = true;
- if (!resp.isSetExceptionMessages()) {
- // The pushPipeMeta process on dataNode encountered internal errors
+ if (!resp.isSetExceptionMessages() ||
resp.getExceptionMessages().isEmpty()) {
+ final String statusMessage = resp.getStatus().getMessage();
+ recordUnscopedDataNodePushPipeMetaException(
+ fallbackPipeStaticMeta,
+ dataNodeId,
+ statusMessage == null || statusMessage.isEmpty()
+ ? ProcedureMessages
+
.EXCEPTION_INTERNAL_ERROR_WHILE_PROCESSING_PUSHPIPEMETA_ON_DATANODE_5983BD5F
+ : statusMessage,
+ System.currentTimeMillis());
continue;
}
for (final TPushPipeMetaRespExceptionMessage message :
resp.getExceptionMessages()) {
+ final String messagePipeName = message.getPipeName();
+ if (messagePipeName == null || messagePipeName.isEmpty()) {
+ recordUnscopedDataNodePushPipeMetaException(
+ fallbackPipeStaticMeta, dataNodeId, message.getMessage(),
message.getTimeStamp());
+ continue;
+ }
final PipeMeta pipeMeta =
message.isSetCreationTime()
- ? pipeMetaKeeper.getPipeMeta(message.getPipeName(),
message.getCreationTime())
- : pipeMetaKeeper.getPipeMeta(message.getPipeName());
+ ? pipeMetaKeeper.getPipeMeta(messagePipeName,
message.getCreationTime())
+ : pipeMetaKeeper.getPipeMeta(messagePipeName);
if (pipeMeta == null) {
continue;
}
- final PipeRuntimeMeta runtimeMeta = pipeMeta.getRuntimeMeta();
-
- if (PipeStatus.PRE_DELETE.equals(runtimeMeta.getStatus().get())) {
- continue;
- }
-
- // Keep user-stopped pipes out of the auto-restart flow. Otherwise,
a failed STOPPED meta
- // sync can turn a manually stopped pipe into a runtime-stopped one
and the next
- // PipeMetaSyncer round will restart it automatically.
- if (PipeStatus.STOPPED.equals(runtimeMeta.getStatus().get())
- && !runtimeMeta.getIsStoppedByRuntimeException()) {
- continue;
- }
-
- // Mark the status of the pipe with exception as stopped
- runtimeMeta.getStatus().set(PipeStatus.STOPPED);
- runtimeMeta.setIsStoppedByRuntimeException(true);
-
- final Map<Integer, PipeRuntimeException> exceptionMap =
- runtimeMeta.getNodeId2PipeRuntimeExceptionMap();
- if (!exceptionMap.containsKey(dataNodeId)
- || exceptionMap.get(dataNodeId).getTimeStamp() <
message.getTimeStamp()) {
- exceptionMap.put(
- dataNodeId,
- new PipeRuntimeCriticalException(message.getMessage(),
message.getTimeStamp()));
- }
+ recordDataNodePushPipeMetaException(
+ pipeMeta, dataNodeId, message.getMessage(),
message.getTimeStamp());
}
}
}
@@ -1210,6 +1225,48 @@ public class PipeTaskInfo implements SnapshotProcessor {
return hasException;
}
+ private void recordUnscopedDataNodePushPipeMetaException(
+ final PipeStaticMeta fallbackPipeStaticMeta,
+ final int dataNodeId,
+ final String message,
+ final long timeStamp) {
+ if (fallbackPipeStaticMeta == null) {
+ pipeMetaKeeper
+ .getPipeMetaList()
+ .forEach(
+ pipeMeta ->
+ recordDataNodePushPipeMetaException(pipeMeta, dataNodeId,
message, timeStamp));
+ } else {
+ final PipeMeta pipeMeta =
pipeMetaKeeper.getPipeMeta(fallbackPipeStaticMeta);
+ if (pipeMeta != null) {
+ recordDataNodePushPipeMetaException(pipeMeta, dataNodeId, message,
timeStamp);
+ }
+ }
+ }
+
+ private void recordDataNodePushPipeMetaException(
+ final PipeMeta pipeMeta, final int dataNodeId, final String message,
final long timeStamp) {
+ final PipeRuntimeMeta runtimeMeta = pipeMeta.getRuntimeMeta();
+
+ if (PipeStatus.PRE_DELETE.equals(runtimeMeta.getStatus().get())) {
+ return;
+ }
+
+ // Keep user-stopped pipes out of the auto-restart flow, but retain the
failure for SHOW PIPES.
+ if (!PipeStatus.STOPPED.equals(runtimeMeta.getStatus().get())
+ || runtimeMeta.getIsStoppedByRuntimeException()) {
+ runtimeMeta.getStatus().set(PipeStatus.STOPPED);
+ runtimeMeta.setIsStoppedByRuntimeException(true);
+ }
+
+ final Map<Integer, PipeRuntimeException> exceptionMap =
+ runtimeMeta.getNodeId2PipeRuntimeExceptionMap();
+ if (!exceptionMap.containsKey(dataNodeId)
+ || exceptionMap.get(dataNodeId).getTimeStamp() < timeStamp) {
+ exceptionMap.put(dataNodeId, new PipeRuntimeCriticalException(message,
timeStamp));
+ }
+ }
+
public boolean autoRestart() {
acquireWriteLock();
try {
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java
index 0710fe345b8..c52864f4328 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java
@@ -21,9 +21,12 @@ package org.apache.iotdb.confignode.procedure.impl.pipe;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
+import
org.apache.iotdb.confignode.consensus.request.write.pipe.runtime.PipeHandleMetaChangePlan;
+import org.apache.iotdb.confignode.i18n.ConfigNodeMessages;
import org.apache.iotdb.confignode.i18n.ProcedureMessages;
import
org.apache.iotdb.confignode.manager.pipe.metric.overview.PipeProcedureMetrics;
import org.apache.iotdb.confignode.persistence.pipe.PipeTaskInfo;
@@ -36,6 +39,7 @@ import
org.apache.iotdb.confignode.procedure.state.ProcedureLockState;
import
org.apache.iotdb.confignode.procedure.state.pipe.task.OperatePipeTaskState;
import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema;
import org.apache.iotdb.confignode.service.ConfigNode;
+import org.apache.iotdb.consensus.exception.ConsensusException;
import org.apache.iotdb.db.pipe.source.dataregion.DataRegionListeningFilter;
import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaResp;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
@@ -655,14 +659,16 @@ public abstract class AbstractOperatePipeProcedureV2
.forEach(
message -> {
// Ignore the timeStamp for simplicity
- if (pipeName == null) {
- enoughMemoryMessageBuilder.append(
- String.format(
- "PipeName: %s, Message: %s",
- message.getPipeName(), message.getMessage()));
- } else if (pipeName.equals(message.getPipeName())) {
- enoughMemoryMessageBuilder.append(
- String.format("Message: %s", message.getMessage()));
+ if (isPipeMetaExceptionForPipe(pipeName,
message.getPipeName())) {
+ if (pipeName == null) {
+ enoughMemoryMessageBuilder.append(
+ String.format(
+ "PipeName: %s, Message: %s",
+ message.getPipeName(), message.getMessage()));
+ } else {
+ enoughMemoryMessageBuilder.append(
+ String.format("Message: %s", message.getMessage()));
+ }
}
});
enoughMemoryMessageBuilder.append(".");
@@ -678,7 +684,7 @@ public abstract class AbstractOperatePipeProcedureV2
}
if (resp.getStatus().getCode() ==
TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode()) {
- if (!resp.isSetExceptionMessages()) {
+ if (!resp.isSetExceptionMessages() ||
resp.getExceptionMessages().isEmpty()) {
final String statusMessage = resp.getStatus().getMessage();
exceptionMessageBuilder.append(
String.format(
@@ -693,16 +699,17 @@ public abstract class AbstractOperatePipeProcedureV2
.forEach(
message -> {
// Ignore the timeStamp for simplicity
- if (pipeName == null) {
+ if (isPipeMetaExceptionForPipe(pipeName,
message.getPipeName())) {
hasException.set(true);
- exceptionMessageBuilder.append(
- String.format(
- "PipeName: %s, Message: %s",
- message.getPipeName(), message.getMessage()));
- } else if (pipeName.equals(message.getPipeName())) {
- hasException.set(true);
- exceptionMessageBuilder.append(
- String.format("Message: %s", message.getMessage()));
+ if (pipeName == null) {
+ exceptionMessageBuilder.append(
+ String.format(
+ "PipeName: %s, Message: %s",
+ message.getPipeName(), message.getMessage()));
+ } else {
+ exceptionMessageBuilder.append(
+ String.format("Message: %s", message.getMessage()));
+ }
}
});
@@ -722,6 +729,14 @@ public abstract class AbstractOperatePipeProcedureV2
return exceptionMessageBuilder.toString();
}
+ private static boolean isPipeMetaExceptionForPipe(
+ final String pipeName, final String exceptionPipeName) {
+ return pipeName == null
+ || exceptionPipeName == null
+ || exceptionPipeName.isEmpty()
+ || pipeName.equals(exceptionPipeName);
+ }
+
protected void pushPipeMetaToDataNodesIgnoreException(ConfigNodeProcedureEnv
env) {
try {
// Ignore the exceptions reported
@@ -741,6 +756,35 @@ public abstract class AbstractOperatePipeProcedureV2
pipeMetaBinaryList, this::setPendingDataNodeIds);
}
+ protected void writePipeMetaChangesToConfigNodeConsensus(final
ConfigNodeProcedureEnv env) {
+ final List<PipeMeta> pipeMetaList = new ArrayList<>();
+ for (final PipeMeta pipeMeta : pipeTaskInfo.get().getPipeMetaList()) {
+ pipeMetaList.add(pipeMeta);
+ }
+
+ TSStatus response;
+ try {
+ response =
+ env.getConfigManager()
+ .getConsensusManager()
+ .write(new PipeHandleMetaChangePlan(pipeMetaList));
+ } catch (final ConsensusException e) {
+
LOGGER.warn(ConfigNodeMessages.FAILED_IN_THE_WRITE_API_EXECUTING_THE_CONSENSUS_LAYER_DUE,
e);
+ response = new
TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
+ response.setMessage(e.getMessage());
+ }
+ if (response.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new PipeException(response.getMessage());
+ }
+ pipeTaskInfo.get().markPipeMetaExceptionsPersisted();
+ }
+
+ protected void persistPendingPipeMetaExceptions(final ConfigNodeProcedureEnv
env) {
+ if (pipeTaskInfo.get().hasPendingPipeMetaExceptionPersistence()) {
+ writePipeMetaChangesToConfigNodeConsensus(env);
+ }
+ }
+
protected void pushPipeMetaToDataNodesBestEffort(ConfigNodeProcedureEnv env)
{
try {
pushPipeMetaToDataNodesBestEffortAndGetResponse(env);
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeMetaSyncProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeMetaSyncProcedure.java
index f18cfb6d545..0c578cf1b6b 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeMetaSyncProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeMetaSyncProcedure.java
@@ -204,6 +204,7 @@ public class PipeMetaSyncProcedure extends
AbstractOperatePipeProcedureV2 {
if (response.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
throw new PipeException(response.getMessage());
}
+ pipeTaskInfo.get().markPipeMetaExceptionsPersisted();
}
@Override
@@ -211,8 +212,11 @@ public class PipeMetaSyncProcedure extends
AbstractOperatePipeProcedureV2 {
throws PipeException, IOException {
LOGGER.debug(ProcedureMessages.PIPEMETASYNCPROCEDURE_EXECUTEFROMOPERATEONDATANODES);
+ persistPendingPipeMetaExceptions(env);
+
Map<Integer, TPushPipeMetaResp> respMap =
pushPipeMetaToDataNodesBestEffortAndGetResponse(env);
if (pipeTaskInfo.get().recordDataNodePushPipeMetaExceptions(respMap)) {
+ writePipeMetaChangesToConfigNodeConsensus(env);
throw new PipeException(
String.format(
ProcedureMessages.FAILED_TO_PUSH_PIPE_META_TO_DATANODES_DETAILS,
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AlterPipeProcedureV2.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AlterPipeProcedureV2.java
index f9c3ec22946..cf380ac0853 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AlterPipeProcedureV2.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AlterPipeProcedureV2.java
@@ -377,8 +377,13 @@ public class AlterPipeProcedureV2 extends
AbstractOperatePipeProcedureV2 {
final String pipeName = alterPipeRequest.getPipeName();
LOGGER.info(ProcedureMessages.ALTERPIPEPROCEDUREV2_EXECUTEFROMOPERATEONDATANODES,
pipeName);
- final String exceptionMessage =
- parsePushPipeMetaExceptionForPipe(pipeName,
pushAlteredPipeMetaToDataNodes(env));
+ persistPendingPipeMetaExceptions(env);
+
+ final Map<Integer, TPushPipeMetaResp> respMap =
pushAlteredPipeMetaToDataNodes(env);
+ if (pipeTaskInfo.get().recordDataNodePushPipeMetaExceptions(respMap,
updatedPipeStaticMeta)) {
+ writePipeMetaChangesToConfigNodeConsensus(env);
+ }
+ final String exceptionMessage =
parsePushPipeMetaExceptionForPipe(pipeName, respMap);
if (!exceptionMessage.isEmpty()) {
LOGGER.warn(
ProcedureMessages.FAILED_TO_ALTER_PIPE_DETAILS_METADATA_WILL_BE_SYNCHRONIZED_LATER,
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java
index f62402db84e..40134cf6dd9 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java
@@ -51,6 +51,7 @@ import
org.apache.iotdb.confignode.procedure.store.ProcedureType;
import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
import org.apache.iotdb.confignode.rpc.thrift.TPermissionInfoResp;
import org.apache.iotdb.consensus.exception.ConsensusException;
+import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaResp;
import org.apache.iotdb.pipe.api.PipePlugin;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.exception.PipeException;
@@ -411,9 +412,14 @@ public class CreatePipeProcedureV2 extends
AbstractOperatePipeProcedureV2 {
final String pipeName = createPipeRequest.getPipeName();
LOGGER.info(ProcedureMessages.CREATEPIPEPROCEDUREV2_EXECUTEFROMOPERATEONDATANODES,
pipeName);
- final String exceptionMessage =
- parsePushPipeMetaExceptionForPipe(
- pipeName, pushSinglePipeMetaToDataNodes(pipeStaticMeta, env));
+ persistPendingPipeMetaExceptions(env);
+
+ final Map<Integer, TPushPipeMetaResp> respMap =
+ pushSinglePipeMetaToDataNodes(pipeStaticMeta, env);
+ if (pipeTaskInfo.get().recordDataNodePushPipeMetaExceptions(respMap,
pipeStaticMeta)) {
+ writePipeMetaChangesToConfigNodeConsensus(env);
+ }
+ final String exceptionMessage =
parsePushPipeMetaExceptionForPipe(pipeName, respMap);
if (!exceptionMessage.isEmpty()) {
LOGGER.warn(
ProcedureMessages.FAILED_TO_CREATE_PIPE_DETAILS_METADATA_WILL_BE_SYNCHRONIZED_LATER,
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/StartPipeProcedureV2.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/StartPipeProcedureV2.java
index a3d0fb01d69..31be648dcbd 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/StartPipeProcedureV2.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/StartPipeProcedureV2.java
@@ -23,7 +23,6 @@ import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus;
-import
org.apache.iotdb.confignode.consensus.request.write.pipe.runtime.PipeHandleMetaChangePlan;
import
org.apache.iotdb.confignode.consensus.request.write.pipe.task.SetPipeStatusPlanV2;
import org.apache.iotdb.confignode.i18n.ConfigNodeMessages;
import org.apache.iotdb.confignode.i18n.ProcedureMessages;
@@ -32,6 +31,7 @@ import
org.apache.iotdb.confignode.procedure.impl.pipe.AbstractOperatePipeProced
import org.apache.iotdb.confignode.procedure.impl.pipe.PipeTaskOperation;
import org.apache.iotdb.confignode.procedure.store.ProcedureType;
import org.apache.iotdb.consensus.exception.ConsensusException;
+import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaResp;
import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -42,8 +42,7 @@ import org.slf4j.LoggerFactory;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.List;
+import java.util.Map;
import java.util.Objects;
public class StartPipeProcedureV2 extends AbstractOperatePipeProcedureV2 {
@@ -129,14 +128,30 @@ public class StartPipeProcedureV2 extends
AbstractOperatePipeProcedureV2 {
isTableModelSet
? pipeTaskInfo.get().isStoppedByRuntimeException(pipeName,
isTableModel)
: pipeTaskInfo.get().isStoppedByRuntimeException(pipeName);
- final PipeStaticMeta pipeStaticMeta =
- (isTableModelSet
- ? pipeTaskInfo.get().getPipeMetaByPipeName(pipeName,
isTableModel)
- : pipeTaskInfo.get().getPipeMetaByPipeName(pipeName))
- .getStaticMeta();
- final String exceptionMessage =
- parsePushPipeMetaExceptionForPipe(
- pipeName, pushSinglePipeMetaToDataNodes(pipeStaticMeta, env));
+ final boolean hadPendingPipeMetaExceptions =
+ pipeTaskInfo.get().hasPendingPipeMetaExceptionPersistence();
+ if (hadPendingPipeMetaExceptions) {
+ persistPendingPipeMetaExceptions(env);
+ }
+ final PipeMeta pipeMeta =
+ isTableModelSet
+ ? pipeTaskInfo.get().getPipeMetaByPipeName(pipeName, isTableModel)
+ : pipeTaskInfo.get().getPipeMetaByPipeName(pipeName);
+ // A previous attempt may already have recorded a runtime stop before a
failed consensus write
+ // or procedure recovery. Persist that state without treating a STOPPED
meta push as a start.
+ if (pipeMeta.getRuntimeMeta().getStatus().get() != PipeStatus.RUNNING) {
+ if (!hadPendingPipeMetaExceptions) {
+ writePipeMetaChangesToConfigNodeConsensus(env);
+ }
+ return;
+ }
+ final PipeStaticMeta pipeStaticMeta = pipeMeta.getStaticMeta();
+ final Map<Integer, TPushPipeMetaResp> respMap =
+ pushSinglePipeMetaToDataNodes(pipeStaticMeta, env);
+ if (pipeTaskInfo.get().recordDataNodePushPipeMetaExceptions(respMap,
pipeStaticMeta)) {
+ writePipeMetaChangesToConfigNodeConsensus(env);
+ }
+ final String exceptionMessage =
parsePushPipeMetaExceptionForPipe(pipeName, respMap);
if (!exceptionMessage.isEmpty()) {
LOGGER.warn(
ProcedureMessages.FAILED_TO_START_PIPE_DETAILS_METADATA_WILL_BE_SYNCHRONIZED_LATER,
@@ -163,28 +178,6 @@ public class StartPipeProcedureV2 extends
AbstractOperatePipeProcedureV2 {
}
}
- private void writePipeMetaChangesToConfigNodeConsensus(final
ConfigNodeProcedureEnv env) {
- final List<PipeMeta> pipeMetaList = new ArrayList<>();
- for (final PipeMeta pipeMeta : pipeTaskInfo.get().getPipeMetaList()) {
- pipeMetaList.add(pipeMeta);
- }
-
- TSStatus response;
- try {
- response =
- env.getConfigManager()
- .getConsensusManager()
- .write(new PipeHandleMetaChangePlan(pipeMetaList));
- } catch (ConsensusException e) {
-
LOGGER.warn(ConfigNodeMessages.FAILED_IN_THE_WRITE_API_EXECUTING_THE_CONSENSUS_LAYER_DUE,
e);
- response = new
TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
- response.setMessage(e.getMessage());
- }
- if (response.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- throw new PipeException(response.getMessage());
- }
- }
-
@Override
public void rollbackFromValidateTask(ConfigNodeProcedureEnv env) {
LOGGER.info(ProcedureMessages.STARTPIPEPROCEDUREV2_ROLLBACKFROMVALIDATETASK,
pipeName);
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/response/pipe/PipeTableRespTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/response/pipe/PipeTableRespTest.java
index b3ae595a5ca..66bc79f3cea 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/response/pipe/PipeTableRespTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/response/pipe/PipeTableRespTest.java
@@ -20,6 +20,7 @@ package org.apache.iotdb.confignode.consensus.response.pipe;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
+import org.apache.iotdb.commons.exception.pipe.PipeRuntimeCriticalException;
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;
@@ -199,6 +200,25 @@ public class PipeTableRespTest {
Assert.assertEquals(PipeStatus.PRE_DELETE.name(),
showPipeResult.get(0).getState());
}
+ @Test
+ public void
testConvertToTShowPipeRespIncludesStoppedStateAndExceptionMessage() {
+ final PipeTableResp pipeTableResp = constructPipeTableResp();
+ final PipeMeta pipeMeta = pipeTableResp.getAllPipeMeta().get(0);
+ pipeMeta.getRuntimeMeta().getStatus().set(PipeStatus.STOPPED);
+ pipeMeta
+ .getRuntimeMeta()
+ .getNodeId2PipeRuntimeExceptionMap()
+ .put(
+ 1, new PipeRuntimeCriticalException("construction failed",
System.currentTimeMillis()));
+
+ final TShowPipeInfo showPipeInfo =
+ pipeTableResp.convertToTShowPipeResp().getPipeInfoList().get(0);
+
+ Assert.assertEquals(PipeStatus.STOPPED.name(), showPipeInfo.getState());
+
Assert.assertTrue(showPipeInfo.getExceptionMessage().contains("construction
failed"));
+ Assert.assertTrue(showPipeInfo.getExceptionMessage().contains("nodeIds:
[1]"));
+ }
+
@Test
public void testFilterByModelBeforeWhereClause() {
TSStatus status = new
TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfoAutoRestartTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfoAutoRestartTest.java
index fd9540aa7f6..b5e443c891b 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfoAutoRestartTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfoAutoRestartTest.java
@@ -34,6 +34,7 @@ import
org.apache.iotdb.confignode.consensus.request.write.pipe.task.CreatePipeP
import
org.apache.iotdb.confignode.consensus.request.write.pipe.task.DropPipePlanV2;
import
org.apache.iotdb.confignode.consensus.request.write.pipe.task.SetPipeStatusPlanV2;
import org.apache.iotdb.confignode.i18n.ConfigNodeMessages;
+import org.apache.iotdb.confignode.i18n.ProcedureMessages;
import org.apache.iotdb.confignode.rpc.thrift.TAlterPipeReq;
import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaResp;
import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaRespExceptionMessage;
@@ -80,6 +81,58 @@ public class PipeTaskInfoAutoRestartTest {
Assert.assertEquals(PipeStatus.RUNNING, runtimeMeta.getStatus().get());
}
+ @Test
+ public void
testRecordDataNodePushPipeMetaExceptionsUsesFallbackForEmptyDetails() {
+ final String pipeName = "emptyDetailsPipe";
+ final PipeStaticMeta staticMeta = createPipe(pipeName, PipeStatus.RUNNING);
+ final TPushPipeMetaResp response =
+ new TPushPipeMetaResp()
+ .setStatus(new
TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode()))
+ .setExceptionMessages(Collections.emptyList());
+
+ Assert.assertTrue(
+ pipeTaskInfo.recordDataNodePushPipeMetaExceptions(
+ Collections.singletonMap(DATA_NODE_ID, response), staticMeta));
+
+ final PipeRuntimeMeta runtimeMeta =
+ pipeTaskInfo.getPipeMetaByPipeName(pipeName).getRuntimeMeta();
+ Assert.assertEquals(PipeStatus.STOPPED, runtimeMeta.getStatus().get());
+ Assert.assertTrue(runtimeMeta.getIsStoppedByRuntimeException());
+ Assert.assertTrue(
+ runtimeMeta
+ .getNodeId2PipeRuntimeExceptionMap()
+ .get(DATA_NODE_ID)
+ .getMessage()
+ .contains(
+ ProcedureMessages
+
.EXCEPTION_INTERNAL_ERROR_WHILE_PROCESSING_PUSHPIPEMETA_ON_DATANODE_5983BD5F));
+ }
+
+ @Test
+ public void
testRecordDataNodePushPipeMetaExceptionsUsesFallbackForEmptyPipeName() {
+ final String pipeName = "emptyPipeNamePipe";
+ final PipeStaticMeta staticMeta = createPipe(pipeName, PipeStatus.RUNNING);
+ final TPushPipeMetaRespExceptionMessage exceptionMessage =
+ new TPushPipeMetaRespExceptionMessage(
+ "", "failed to close schema listening queue",
System.currentTimeMillis());
+ final TPushPipeMetaResp response =
+ new TPushPipeMetaResp()
+ .setStatus(new
TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode()))
+ .setExceptionMessages(Collections.singletonList(exceptionMessage));
+
+ Assert.assertTrue(
+ pipeTaskInfo.recordDataNodePushPipeMetaExceptions(
+ Collections.singletonMap(DATA_NODE_ID, response), staticMeta));
+
+ final PipeRuntimeMeta runtimeMeta =
+ pipeTaskInfo.getPipeMetaByPipeName(pipeName).getRuntimeMeta();
+ Assert.assertEquals(PipeStatus.STOPPED, runtimeMeta.getStatus().get());
+ Assert.assertTrue(runtimeMeta.getIsStoppedByRuntimeException());
+ Assert.assertEquals(
+ "failed to close schema listening queue",
+
runtimeMeta.getNodeId2PipeRuntimeExceptionMap().get(DATA_NODE_ID).getMessage());
+ }
+
@Test
public void
testRecordDataNodePushPipeMetaExceptionsKeepsUserStoppedPipeOutOfAutoRestart() {
final String pipeName = "stoppedPipe";
@@ -94,6 +147,7 @@ public class PipeTaskInfoAutoRestartTest {
Assert.assertEquals(PipeStatus.STOPPED, runtimeMeta.getStatus().get());
Assert.assertFalse(runtimeMeta.getIsStoppedByRuntimeException());
Assert.assertTrue(pipeTaskInfo.isPipeStoppedByUser(pipeName));
+
Assert.assertTrue(runtimeMeta.getNodeId2PipeRuntimeExceptionMap().containsKey(DATA_NODE_ID));
Assert.assertFalse(pipeTaskInfo.autoRestart());
Assert.assertEquals(PipeStatus.STOPPED, runtimeMeta.getStatus().get());
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2Test.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2Test.java
index ae0feec0f72..a4a75d60dc7 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2Test.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2Test.java
@@ -19,21 +19,45 @@
package org.apache.iotdb.confignode.procedure.impl.pipe;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.confignode.persistence.pipe.PipeTaskInfo;
import org.apache.iotdb.confignode.procedure.Procedure;
import org.apache.iotdb.confignode.procedure.impl.StateMachineProcedure;
import
org.apache.iotdb.confignode.procedure.state.pipe.task.OperatePipeTaskState;
+import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaResp;
+import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaRespExceptionMessage;
import org.apache.iotdb.pipe.api.exception.PipeException;
+import org.apache.iotdb.rpc.TSStatusCode;
import org.junit.Assert;
import org.junit.Test;
import java.io.IOException;
+import java.util.Collections;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
public class AbstractOperatePipeProcedureV2Test {
+ @Test
+ public void testParsePushPipeMetaExceptionForPipeIncludesUnscopedException()
{
+ final TPushPipeMetaRespExceptionMessage exceptionMessage =
+ new TPushPipeMetaRespExceptionMessage(
+ "", "failed to close schema listening queue",
System.currentTimeMillis());
+ final TPushPipeMetaResp response =
+ new TPushPipeMetaResp()
+ .setStatus(new
TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode()))
+ .setExceptionMessages(Collections.singletonList(exceptionMessage));
+
+ final String message =
+ AbstractOperatePipeProcedureV2.parsePushPipeMetaExceptionForPipe(
+ "targetPipe", Map.of(1, response));
+
+ Assert.assertTrue(message.contains("DataNodeId: 1"));
+ Assert.assertTrue(message.contains("Message: failed to close schema
listening queue"));
+ }
+
@Test
public void testSuccessfulStateDoesNotYield() throws Exception {
final TestOperatePipeProcedure procedure = new TestOperatePipeProcedure();
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeMetaSyncProcedureTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeMetaSyncProcedureTest.java
index 44b96284a15..9b41b7e8dbb 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeMetaSyncProcedureTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeMetaSyncProcedureTest.java
@@ -19,15 +19,42 @@
package org.apache.iotdb.confignode.procedure.impl.pipe.runtime;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
+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;
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus;
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
+import
org.apache.iotdb.confignode.consensus.request.write.pipe.runtime.PipeHandleMetaChangePlan;
+import
org.apache.iotdb.confignode.consensus.request.write.pipe.task.CreatePipePlanV2;
+import org.apache.iotdb.confignode.manager.ConfigManager;
+import org.apache.iotdb.confignode.manager.consensus.ConsensusManager;
+import org.apache.iotdb.confignode.persistence.pipe.PipeTaskInfo;
+import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
import org.apache.iotdb.confignode.procedure.store.ProcedureFactory;
+import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaResp;
+import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaRespExceptionMessage;
+import org.apache.iotdb.pipe.api.exception.PipeException;
+import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.tsfile.utils.PublicBAOS;
import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
import java.io.DataOutputStream;
+import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicReference;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
public class PipeMetaSyncProcedureTest {
@@ -52,4 +79,217 @@ public class PipeMetaSyncProcedureTest {
fail();
}
}
+
+ @Test
+ public void testDataNodePushFailureIsPersistedToConsensus() throws Exception
{
+ final String pipeName = "metaSyncFailurePipe";
+ final long creationTime = System.currentTimeMillis();
+ final PipeStaticMeta staticMeta =
+ new PipeStaticMeta(
+ pipeName,
+ creationTime,
+ Collections.singletonMap("extractor", "iotdb-source"),
+ Collections.singletonMap("processor", "do-nothing-processor"),
+ Collections.singletonMap("connector", "write-back-sink"));
+ final PipeRuntimeMeta runtimeMeta = new PipeRuntimeMeta(new
ConcurrentHashMap<>());
+ runtimeMeta.getStatus().set(PipeStatus.RUNNING);
+ runtimeMeta
+ .getConsensusGroupId2TaskMetaMap()
+ .put(1, new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1));
+ final PipeTaskInfo taskInfo = new PipeTaskInfo();
+ taskInfo.createPipe(new CreatePipePlanV2(staticMeta, runtimeMeta));
+
+ final String error = "Failed to construct PipeConnector";
+ final TPushPipeMetaRespExceptionMessage exceptionMessage =
+ new TPushPipeMetaRespExceptionMessage(pipeName, error, creationTime)
+ .setCreationTime(creationTime);
+ final Map<Integer, TPushPipeMetaResp> responses =
+ Collections.singletonMap(
+ 1,
+ new TPushPipeMetaResp(new
TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode()))
+
.setExceptionMessages(Collections.singletonList(exceptionMessage)));
+
+ final ConfigNodeProcedureEnv env =
Mockito.mock(ConfigNodeProcedureEnv.class);
+ final ConfigManager configManager = Mockito.mock(ConfigManager.class);
+ final ConsensusManager consensusManager =
Mockito.mock(ConsensusManager.class);
+ Mockito.when(env.getConfigManager()).thenReturn(configManager);
+
Mockito.when(configManager.getConsensusManager()).thenReturn(consensusManager);
+ Mockito.when(consensusManager.write(Mockito.any(ConfigPhysicalPlan.class)))
+ .thenReturn(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+
+ final PipeMetaSyncProcedure procedure = new
TestablePipeMetaSyncProcedure(taskInfo, responses);
+ try {
+ procedure.executeFromOperateOnDataNodes(env);
+ fail("Expected the sync procedure to report the DataNode push failure");
+ } catch (final PipeException expected) {
+ assertTrue(expected.getMessage().contains(error));
+ }
+
+ final ArgumentCaptor<ConfigPhysicalPlan> planCaptor =
+ ArgumentCaptor.forClass(ConfigPhysicalPlan.class);
+ Mockito.verify(consensusManager).write(planCaptor.capture());
+ assertTrue(planCaptor.getValue() instanceof PipeHandleMetaChangePlan);
+ final PipeMeta persistedPipeMeta =
+ ((PipeHandleMetaChangePlan) planCaptor.getValue())
+ .getPipeMetaList().stream()
+ .filter(meta ->
pipeName.equals(meta.getStaticMeta().getPipeName()))
+ .findFirst()
+ .orElseThrow(AssertionError::new);
+ assertEquals(PipeStatus.STOPPED,
persistedPipeMeta.getRuntimeMeta().getStatus().get());
+
assertTrue(persistedPipeMeta.getRuntimeMeta().getIsStoppedByRuntimeException());
+ assertEquals(
+ error,
+
persistedPipeMeta.getRuntimeMeta().getNodeId2PipeRuntimeExceptionMap().get(1).getMessage());
+ }
+
+ @Test
+ public void testUnscopedPushFailureStopsAndPersistsAllPipes() throws
Exception {
+ final PipeTaskInfo taskInfo = new PipeTaskInfo();
+ final PipeStaticMeta firstPipe = createRunningPipe(taskInfo,
"firstUnscopedPipe");
+ final PipeStaticMeta secondPipe = createRunningPipe(taskInfo,
"secondUnscopedPipe");
+ final String statusMessage = "DataNode failed before returning
pipe-specific details";
+ final Map<Integer, TPushPipeMetaResp> responses =
+ Collections.singletonMap(
+ 1,
+ new TPushPipeMetaResp(
+ new TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode())
+ .setMessage(statusMessage)));
+
+ final ConfigNodeProcedureEnv env =
Mockito.mock(ConfigNodeProcedureEnv.class);
+ final ConfigManager configManager = Mockito.mock(ConfigManager.class);
+ final ConsensusManager consensusManager =
Mockito.mock(ConsensusManager.class);
+ Mockito.when(env.getConfigManager()).thenReturn(configManager);
+
Mockito.when(configManager.getConsensusManager()).thenReturn(consensusManager);
+ Mockito.when(consensusManager.write(Mockito.any(ConfigPhysicalPlan.class)))
+ .thenReturn(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+
+ final PipeMetaSyncProcedure procedure = new
TestablePipeMetaSyncProcedure(taskInfo, responses);
+ try {
+ procedure.executeFromOperateOnDataNodes(env);
+ fail("Expected the sync procedure to report the DataNode push failure");
+ } catch (final PipeException expected) {
+ assertTrue(expected.getMessage().contains(statusMessage));
+ }
+
+ final ArgumentCaptor<ConfigPhysicalPlan> planCaptor =
+ ArgumentCaptor.forClass(ConfigPhysicalPlan.class);
+ Mockito.verify(consensusManager).write(planCaptor.capture());
+ assertTrue(planCaptor.getValue() instanceof PipeHandleMetaChangePlan);
+ final PipeHandleMetaChangePlan persistedPlan =
+ (PipeHandleMetaChangePlan)
+
ConfigPhysicalPlan.Factory.create(planCaptor.getValue().serializeToByteBuffer());
+ final Map<String, PipeMeta> persistedPipes =
+ persistedPlan.getPipeMetaList().stream()
+ .collect(
+ java.util.stream.Collectors.toMap(
+ meta -> meta.getStaticMeta().getPipeName(), meta -> meta));
+ assertEquals(2, persistedPipes.size());
+ for (final PipeStaticMeta pipeStaticMeta : new PipeStaticMeta[]
{firstPipe, secondPipe}) {
+ final PipeRuntimeMeta persistedRuntimeMeta =
+ persistedPipes.get(pipeStaticMeta.getPipeName()).getRuntimeMeta();
+ assertEquals(PipeStatus.STOPPED, persistedRuntimeMeta.getStatus().get());
+ assertTrue(persistedRuntimeMeta.getIsStoppedByRuntimeException());
+ assertTrue(
+ persistedRuntimeMeta
+ .getNodeId2PipeRuntimeExceptionMap()
+ .get(1)
+ .getMessage()
+ .contains(statusMessage));
+ }
+ assertTrue(
+ taskInfo
+ .getPipeMetaByPipeName(firstPipe.getPipeName())
+ .getRuntimeMeta()
+ .getNodeId2PipeRuntimeExceptionMap()
+ .containsKey(1));
+ }
+
+ @Test
+ public void testRetriesPersistenceWhenPushSucceedsAfterConsensusFailure()
throws Exception {
+ final PipeTaskInfo taskInfo = new PipeTaskInfo();
+ final PipeStaticMeta pipeStaticMeta = createRunningPipe(taskInfo,
"metaSyncRetryPipe");
+ final String error = "DataNode failed before returning pipe-specific
details";
+ final Map<Integer, TPushPipeMetaResp> responses = new HashMap<>();
+ responses.put(
+ 1,
+ new TPushPipeMetaResp(
+ new
TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode()).setMessage(error)));
+
+ final ConfigNodeProcedureEnv env =
Mockito.mock(ConfigNodeProcedureEnv.class);
+ final ConfigManager configManager = Mockito.mock(ConfigManager.class);
+ final ConsensusManager consensusManager =
Mockito.mock(ConsensusManager.class);
+ Mockito.when(env.getConfigManager()).thenReturn(configManager);
+
Mockito.when(configManager.getConsensusManager()).thenReturn(consensusManager);
+ Mockito.when(consensusManager.write(Mockito.any(ConfigPhysicalPlan.class)))
+ .thenReturn(new
TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode()))
+ .thenReturn(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+
+ final PipeMetaSyncProcedure procedure = new
TestablePipeMetaSyncProcedure(taskInfo, responses);
+ try {
+ procedure.executeFromOperateOnDataNodes(env);
+ fail("Expected the first attempt to fail its ConfigNode consensus
write");
+ } catch (final PipeException expected) {
+ // The procedure retries this state after the persistence failure.
+ }
+
+ responses.put(
+ 1, new TPushPipeMetaResp(new
TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())));
+ procedure.executeFromOperateOnDataNodes(env);
+
+ final ArgumentCaptor<ConfigPhysicalPlan> planCaptor =
+ ArgumentCaptor.forClass(ConfigPhysicalPlan.class);
+ Mockito.verify(consensusManager,
Mockito.times(2)).write(planCaptor.capture());
+ assertTrue(planCaptor.getAllValues().get(1) instanceof
PipeHandleMetaChangePlan);
+ final PipeMeta persistedPipeMeta =
+ ((PipeHandleMetaChangePlan) planCaptor.getAllValues().get(1))
+ .getPipeMetaList().stream()
+ .filter(
+ meta ->
pipeStaticMeta.getPipeName().equals(meta.getStaticMeta().getPipeName()))
+ .findFirst()
+ .orElseThrow(AssertionError::new);
+ assertEquals(PipeStatus.STOPPED,
persistedPipeMeta.getRuntimeMeta().getStatus().get());
+
assertTrue(persistedPipeMeta.getRuntimeMeta().getIsStoppedByRuntimeException());
+ assertTrue(
+ persistedPipeMeta
+ .getRuntimeMeta()
+ .getNodeId2PipeRuntimeExceptionMap()
+ .get(1)
+ .getMessage()
+ .contains(error));
+ }
+
+ private static PipeStaticMeta createRunningPipe(
+ final PipeTaskInfo taskInfo, final String pipeName) {
+ final long creationTime = System.nanoTime();
+ final PipeStaticMeta staticMeta =
+ new PipeStaticMeta(
+ pipeName,
+ creationTime,
+ Collections.singletonMap("extractor", "iotdb-source"),
+ Collections.singletonMap("processor", "do-nothing-processor"),
+ Collections.singletonMap("connector", "write-back-sink"));
+ final PipeRuntimeMeta runtimeMeta = new PipeRuntimeMeta(new
ConcurrentHashMap<>());
+ runtimeMeta.getStatus().set(PipeStatus.RUNNING);
+ runtimeMeta
+ .getConsensusGroupId2TaskMetaMap()
+ .put(1, new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1));
+ taskInfo.createPipe(new CreatePipePlanV2(staticMeta, runtimeMeta));
+ return staticMeta;
+ }
+
+ private static class TestablePipeMetaSyncProcedure extends
PipeMetaSyncProcedure {
+ private final Map<Integer, TPushPipeMetaResp> responses;
+
+ private TestablePipeMetaSyncProcedure(
+ final PipeTaskInfo taskInfo, final Map<Integer, TPushPipeMetaResp>
responses) {
+ this.pipeTaskInfo = new AtomicReference<>(taskInfo);
+ this.responses = responses;
+ }
+
+ @Override
+ protected Map<Integer, TPushPipeMetaResp>
pushPipeMetaToDataNodesBestEffortAndGetResponse(
+ final ConfigNodeProcedureEnv env) throws IOException {
+ return responses;
+ }
+ }
}
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2Test.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2Test.java
index 013376571ec..ac90a2ddec1 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2Test.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2Test.java
@@ -20,28 +20,174 @@
package org.apache.iotdb.confignode.procedure.impl.pipe.task;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
+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.config.constant.PipeSourceConstant;
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
+import
org.apache.iotdb.confignode.consensus.request.write.pipe.runtime.PipeHandleMetaChangePlan;
import org.apache.iotdb.confignode.manager.ConfigManager;
import org.apache.iotdb.confignode.manager.PermissionManager;
+import org.apache.iotdb.confignode.manager.consensus.ConsensusManager;
+import org.apache.iotdb.confignode.persistence.pipe.PipeTaskInfo;
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
import org.apache.iotdb.confignode.procedure.store.ProcedureFactory;
import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
import org.apache.iotdb.confignode.rpc.thrift.TPermissionInfoResp;
+import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaResp;
+import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaRespExceptionMessage;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.tsfile.utils.PublicBAOS;
import org.junit.Test;
+import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
import java.io.DataOutputStream;
import java.nio.ByteBuffer;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
public class CreatePipeProcedureV2Test {
+ @Test
+ public void testDataNodeConstructionFailureIsRecorded() throws Exception {
+ final String pipeName = "failedHistoryPipe";
+ final TCreatePipeReq request = createPipeRequest(pipeName);
+ final PipeTaskInfo taskInfo = new PipeTaskInfo();
+ final CreatePipeProcedureV2 procedure =
+ new CreatePipeProcedureV2(request, new AtomicReference<>(taskInfo));
+
procedure.executeFromCalculateInfoForTask(Mockito.mock(ConfigNodeProcedureEnv.class));
+ taskInfo.createPipe(procedure.constructPlan());
+
+ final String error = "Failed to construct PipeConnector";
+ final TPushPipeMetaRespExceptionMessage exceptionMessage =
+ new TPushPipeMetaRespExceptionMessage(request.getPipeName(), error, 1L)
+ .setCreationTime(procedure.getPipeStaticMeta().getCreationTime());
+ final ConfigNodeProcedureEnv env =
Mockito.mock(ConfigNodeProcedureEnv.class);
+ final ConsensusManager consensusManager =
mockSuccessfulConsensusWrite(env);
+
Mockito.when(env.pushSinglePipeMetaToDataNodes(Mockito.any(ByteBuffer.class),
Mockito.any()))
+ .thenReturn(
+ Collections.singletonMap(
+ 1,
+ new TPushPipeMetaResp(
+ new
TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode()))
+
.setExceptionMessages(Collections.singletonList(exceptionMessage))));
+
+ procedure.executeFromOperateOnDataNodes(env);
+
+ final PipeRuntimeMeta runtimeMeta =
+ taskInfo.getPipeMetaByPipeName(request.getPipeName()).getRuntimeMeta();
+ assertEquals(PipeStatus.STOPPED, runtimeMeta.getStatus().get());
+ assertTrue(runtimeMeta.getIsStoppedByRuntimeException());
+ assertEquals(error,
runtimeMeta.getNodeId2PipeRuntimeExceptionMap().get(1).getMessage());
+ assertPipeMetaChangeWasWritten(consensusManager, request.getPipeName(),
error);
+ }
+
+ @Test
+ public void testRetriesFailurePersistenceBeforeSuccessfulMetaPush() throws
Exception {
+ final String pipeName = "failedHistoryPipePersistenceRetry";
+ final TCreatePipeReq request = createPipeRequest(pipeName);
+ final PipeTaskInfo taskInfo = new PipeTaskInfo();
+ final CreatePipeProcedureV2 procedure =
+ new CreatePipeProcedureV2(request, new AtomicReference<>(taskInfo));
+
procedure.executeFromCalculateInfoForTask(Mockito.mock(ConfigNodeProcedureEnv.class));
+ taskInfo.createPipe(procedure.constructPlan());
+
+ final String error = "DataNode failed before returning pipe-specific
details";
+ final TPushPipeMetaRespExceptionMessage exceptionMessage =
+ new TPushPipeMetaRespExceptionMessage(request.getPipeName(), error, 1L)
+ .setCreationTime(procedure.getPipeStaticMeta().getCreationTime());
+ final Map<Integer, TPushPipeMetaResp> responses = new HashMap<>();
+ responses.put(
+ 1,
+ new TPushPipeMetaResp(new
TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode()))
+
.setExceptionMessages(Collections.singletonList(exceptionMessage)));
+
+ final ConfigNodeProcedureEnv env =
Mockito.mock(ConfigNodeProcedureEnv.class);
+ final ConfigManager configManager = Mockito.mock(ConfigManager.class);
+ final ConsensusManager consensusManager =
Mockito.mock(ConsensusManager.class);
+ Mockito.when(env.getConfigManager()).thenReturn(configManager);
+
Mockito.when(configManager.getConsensusManager()).thenReturn(consensusManager);
+ Mockito.when(consensusManager.write(Mockito.any(ConfigPhysicalPlan.class)))
+ .thenReturn(new
TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode()))
+ .thenReturn(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+
Mockito.when(env.pushSinglePipeMetaToDataNodes(Mockito.any(ByteBuffer.class),
Mockito.any()))
+ .thenReturn(responses);
+
+ try {
+ procedure.executeFromOperateOnDataNodes(env);
+ fail("Expected the first attempt to fail its ConfigNode consensus
write");
+ } catch (final org.apache.iotdb.pipe.api.exception.PipeException expected)
{
+ // The procedure retries this state after the persistence failure.
+ }
+
+ responses.put(
+ 1, new TPushPipeMetaResp(new
TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())));
+ procedure.executeFromOperateOnDataNodes(env);
+
+ final ArgumentCaptor<ConfigPhysicalPlan> planCaptor =
+ ArgumentCaptor.forClass(ConfigPhysicalPlan.class);
+ Mockito.verify(consensusManager,
Mockito.times(2)).write(planCaptor.capture());
+ assertTrue(planCaptor.getAllValues().get(1) instanceof
PipeHandleMetaChangePlan);
+ final PipeRuntimeMeta persistedRuntimeMeta =
+ ((PipeHandleMetaChangePlan) planCaptor.getAllValues().get(1))
+ .getPipeMetaList().stream()
+ .filter(meta ->
request.getPipeName().equals(meta.getStaticMeta().getPipeName()))
+ .findFirst()
+ .orElseThrow(AssertionError::new)
+ .getRuntimeMeta();
+ assertEquals(PipeStatus.STOPPED, persistedRuntimeMeta.getStatus().get());
+ assertTrue(persistedRuntimeMeta.getIsStoppedByRuntimeException());
+ assertEquals(
+ error,
persistedRuntimeMeta.getNodeId2PipeRuntimeExceptionMap().get(1).getMessage());
+ }
+
+ private static ConsensusManager mockSuccessfulConsensusWrite(final
ConfigNodeProcedureEnv env)
+ throws Exception {
+ final ConfigManager configManager = Mockito.mock(ConfigManager.class);
+ final ConsensusManager consensusManager =
Mockito.mock(ConsensusManager.class);
+ Mockito.when(env.getConfigManager()).thenReturn(configManager);
+
Mockito.when(configManager.getConsensusManager()).thenReturn(consensusManager);
+ Mockito.when(consensusManager.write(Mockito.any(ConfigPhysicalPlan.class)))
+ .thenReturn(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+ return consensusManager;
+ }
+
+ private static void assertPipeMetaChangeWasWritten(
+ final ConsensusManager consensusManager, final String pipeName, final
String error)
+ throws Exception {
+ final ArgumentCaptor<ConfigPhysicalPlan> planCaptor =
+ ArgumentCaptor.forClass(ConfigPhysicalPlan.class);
+ Mockito.verify(consensusManager).write(planCaptor.capture());
+ assertTrue(planCaptor.getValue() instanceof PipeHandleMetaChangePlan);
+ final PipeRuntimeMeta persistedRuntimeMeta =
+ ((PipeHandleMetaChangePlan) planCaptor.getValue())
+ .getPipeMetaList().stream()
+ .filter(meta ->
pipeName.equals(meta.getStaticMeta().getPipeName()))
+ .findFirst()
+ .orElseThrow(AssertionError::new)
+ .getRuntimeMeta();
+ assertEquals(PipeStatus.STOPPED, persistedRuntimeMeta.getStatus().get());
+ assertTrue(persistedRuntimeMeta.getIsStoppedByRuntimeException());
+ assertEquals(
+ error,
persistedRuntimeMeta.getNodeId2PipeRuntimeExceptionMap().get(1).getMessage());
+ }
+
+ private static TCreatePipeReq createPipeRequest(final String pipeName) {
+ final Map<String, String> extractorAttributes = new HashMap<>();
+
extractorAttributes.put(PipeSourceConstant.EXTRACTOR_CONSENSUS_GROUP_ID_KEY,
"DataRegion[1]");
+
extractorAttributes.put(PipeSourceConstant.EXTRACTOR_CONSENSUS_SENDER_DATANODE_ID_KEY,
"1");
+ return new TCreatePipeReq(PipeStaticMeta.CONSENSUS_PIPE_PREFIX + pipeName,
new HashMap<>())
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(new HashMap<>());
+ }
+
@Test
public void serializeDeserializeTest() {
PublicBAOS byteArrayOutputStream = new PublicBAOS();
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/StartPipeProcedureV2Test.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/StartPipeProcedureV2Test.java
index a6234421742..8a1796d4fd3 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/StartPipeProcedureV2Test.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/StartPipeProcedureV2Test.java
@@ -19,18 +19,227 @@
package org.apache.iotdb.confignode.procedure.impl.pipe.task;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
+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.confignode.consensus.request.ConfigPhysicalPlan;
+import
org.apache.iotdb.confignode.consensus.request.write.pipe.runtime.PipeHandleMetaChangePlan;
+import
org.apache.iotdb.confignode.consensus.request.write.pipe.task.CreatePipePlanV2;
+import
org.apache.iotdb.confignode.consensus.request.write.pipe.task.SetPipeStatusPlanV2;
+import org.apache.iotdb.confignode.manager.ConfigManager;
+import org.apache.iotdb.confignode.manager.consensus.ConsensusManager;
+import org.apache.iotdb.confignode.persistence.pipe.PipeTaskInfo;
+import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
import org.apache.iotdb.confignode.procedure.store.ProcedureFactory;
+import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaResp;
+import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaRespExceptionMessage;
+import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.tsfile.utils.PublicBAOS;
import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
import java.io.DataOutputStream;
import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicReference;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
public class StartPipeProcedureV2Test {
+ @Test
+ public void testDataNodeConstructionFailureIsRecorded() throws Exception {
+ final String pipeName = "startFailurePipe";
+ final PipeTaskInfo taskInfo = new PipeTaskInfo();
+ final PipeStaticMeta staticMeta =
+ new PipeStaticMeta(
+ pipeName,
+ System.currentTimeMillis(),
+ Collections.singletonMap("extractor", "iotdb-source"),
+ Collections.singletonMap("processor", "do-nothing-processor"),
+ Collections.singletonMap("connector", "write-back-sink"));
+ final PipeRuntimeMeta runtimeMeta = new PipeRuntimeMeta(new
ConcurrentHashMap<>());
+ runtimeMeta
+ .getConsensusGroupId2TaskMetaMap()
+ .put(1, new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1));
+ taskInfo.createPipe(new CreatePipePlanV2(staticMeta, runtimeMeta));
+ taskInfo.setPipeStatus(new SetPipeStatusPlanV2(pipeName,
PipeStatus.RUNNING));
+
+ final StartPipeProcedureV2 procedure = new
TestableStartPipeProcedureV2(pipeName, taskInfo);
+ final ConfigNodeProcedureEnv env =
Mockito.mock(ConfigNodeProcedureEnv.class);
+ final ConsensusManager consensusManager =
mockSuccessfulConsensusWrite(env);
+ final String error = "Failed to construct PipeConnector";
+ final TPushPipeMetaRespExceptionMessage exceptionMessage =
+ new TPushPipeMetaRespExceptionMessage(pipeName, error,
System.currentTimeMillis())
+ .setCreationTime(staticMeta.getCreationTime());
+
Mockito.when(env.pushSinglePipeMetaToDataNodes(Mockito.any(ByteBuffer.class),
Mockito.any()))
+ .thenReturn(
+ Collections.singletonMap(
+ 1,
+ new TPushPipeMetaResp(
+ new
TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode()))
+
.setExceptionMessages(Collections.singletonList(exceptionMessage))));
+
+ procedure.executeFromOperateOnDataNodes(env);
+
+ final PipeRuntimeMeta recordedRuntimeMeta =
+ taskInfo.getPipeMetaByPipeName(pipeName).getRuntimeMeta();
+ assertEquals(PipeStatus.STOPPED, recordedRuntimeMeta.getStatus().get());
+ assertEquals(
+ error,
recordedRuntimeMeta.getNodeId2PipeRuntimeExceptionMap().get(1).getMessage());
+ assertPipeMetaChangeWasWritten(consensusManager, pipeName, error);
+ }
+
+ @Test
+ public void
testFailureWithoutExceptionDetailsIsNotTreatedAsSuccessfulStart() throws
Exception {
+ final String pipeName = "startFailureWithoutDetailsPipe";
+ final PipeTaskInfo taskInfo = new PipeTaskInfo();
+ final PipeStaticMeta staticMeta =
+ new PipeStaticMeta(
+ pipeName,
+ System.currentTimeMillis(),
+ Collections.singletonMap("extractor", "iotdb-source"),
+ Collections.singletonMap("processor", "do-nothing-processor"),
+ Collections.singletonMap("connector", "write-back-sink"));
+ final PipeRuntimeMeta runtimeMeta = new PipeRuntimeMeta(new
ConcurrentHashMap<>());
+ runtimeMeta
+ .getConsensusGroupId2TaskMetaMap()
+ .put(1, new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1));
+ taskInfo.createPipe(new CreatePipePlanV2(staticMeta, runtimeMeta));
+ taskInfo.setPipeStatus(new SetPipeStatusPlanV2(pipeName,
PipeStatus.RUNNING));
+
+ final StartPipeProcedureV2 procedure = new
TestableStartPipeProcedureV2(pipeName, taskInfo);
+ final ConfigNodeProcedureEnv env =
Mockito.mock(ConfigNodeProcedureEnv.class);
+ final ConsensusManager consensusManager =
mockSuccessfulConsensusWrite(env);
+ final String error = "DataNode failed before returning pipe-specific
details";
+
Mockito.when(env.pushSinglePipeMetaToDataNodes(Mockito.any(ByteBuffer.class),
Mockito.any()))
+ .thenReturn(
+ Collections.singletonMap(
+ 1,
+ new TPushPipeMetaResp(
+ new
TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode())
+ .setMessage(error))
+ .setExceptionMessages(Collections.emptyList())));
+
+ procedure.executeFromOperateOnDataNodes(env);
+
+ final PipeRuntimeMeta recordedRuntimeMeta =
+ taskInfo.getPipeMetaByPipeName(pipeName).getRuntimeMeta();
+ assertEquals(PipeStatus.STOPPED, recordedRuntimeMeta.getStatus().get());
+ assertTrue(recordedRuntimeMeta.getIsStoppedByRuntimeException());
+ assertTrue(
+ recordedRuntimeMeta
+ .getNodeId2PipeRuntimeExceptionMap()
+ .get(1)
+ .getMessage()
+ .contains(error));
+ assertPipeMetaChangeWasWritten(consensusManager, pipeName, error);
+ }
+
+ @Test
+ public void testRetryAfterFailureToPersistRuntimeStopDoesNotClearFailure()
throws Exception {
+ final String pipeName = "startFailurePersistRetryPipe";
+ final PipeTaskInfo taskInfo = new PipeTaskInfo();
+ final PipeStaticMeta staticMeta =
+ new PipeStaticMeta(
+ pipeName,
+ System.currentTimeMillis(),
+ Collections.singletonMap("extractor", "iotdb-source"),
+ Collections.singletonMap("processor", "do-nothing-processor"),
+ Collections.singletonMap("connector", "write-back-sink"));
+ final PipeRuntimeMeta runtimeMeta = new PipeRuntimeMeta(new
ConcurrentHashMap<>());
+ runtimeMeta
+ .getConsensusGroupId2TaskMetaMap()
+ .put(1, new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1));
+ taskInfo.createPipe(new CreatePipePlanV2(staticMeta, runtimeMeta));
+ taskInfo.setPipeStatus(new SetPipeStatusPlanV2(pipeName,
PipeStatus.RUNNING));
+
+ final StartPipeProcedureV2 procedure = new
TestableStartPipeProcedureV2(pipeName, taskInfo);
+ final ConfigNodeProcedureEnv env =
Mockito.mock(ConfigNodeProcedureEnv.class);
+ final ConfigManager configManager = Mockito.mock(ConfigManager.class);
+ final ConsensusManager consensusManager =
Mockito.mock(ConsensusManager.class);
+ Mockito.when(env.getConfigManager()).thenReturn(configManager);
+
Mockito.when(configManager.getConsensusManager()).thenReturn(consensusManager);
+ Mockito.when(consensusManager.write(Mockito.any(ConfigPhysicalPlan.class)))
+ .thenReturn(new
TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode()))
+ .thenReturn(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+
+ final String error = "Failed to construct PipeConnector";
+ final TPushPipeMetaRespExceptionMessage exceptionMessage =
+ new TPushPipeMetaRespExceptionMessage(pipeName, error,
System.currentTimeMillis())
+ .setCreationTime(staticMeta.getCreationTime());
+
Mockito.when(env.pushSinglePipeMetaToDataNodes(Mockito.any(ByteBuffer.class),
Mockito.any()))
+ .thenReturn(
+ Collections.singletonMap(
+ 1,
+ new TPushPipeMetaResp(
+ new
TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode()))
+
.setExceptionMessages(Collections.singletonList(exceptionMessage))));
+
+ try {
+ procedure.executeFromOperateOnDataNodes(env);
+ fail("Expected the first attempt to fail its ConfigNode consensus
write");
+ } catch (final org.apache.iotdb.pipe.api.exception.PipeException expected)
{
+ // The procedure retries this state after the persistence failure.
+ }
+
+ final PipeRuntimeMeta runtimeMetaAfterFirstAttempt =
+ taskInfo.getPipeMetaByPipeName(pipeName).getRuntimeMeta();
+ assertEquals(PipeStatus.STOPPED,
runtimeMetaAfterFirstAttempt.getStatus().get());
+ assertTrue(runtimeMetaAfterFirstAttempt.getIsStoppedByRuntimeException());
+
+ procedure.executeFromOperateOnDataNodes(env);
+
+ final PipeRuntimeMeta runtimeMetaAfterRetry =
+ taskInfo.getPipeMetaByPipeName(pipeName).getRuntimeMeta();
+ assertEquals(PipeStatus.STOPPED, runtimeMetaAfterRetry.getStatus().get());
+ assertTrue(runtimeMetaAfterRetry.getIsStoppedByRuntimeException());
+ assertEquals(
+ error,
runtimeMetaAfterRetry.getNodeId2PipeRuntimeExceptionMap().get(1).getMessage());
+ Mockito.verify(env, Mockito.times(1))
+ .pushSinglePipeMetaToDataNodes(Mockito.any(ByteBuffer.class),
Mockito.any());
+ Mockito.verify(consensusManager,
Mockito.times(2)).write(Mockito.any(ConfigPhysicalPlan.class));
+ }
+
+ private static ConsensusManager mockSuccessfulConsensusWrite(final
ConfigNodeProcedureEnv env)
+ throws Exception {
+ final ConfigManager configManager = Mockito.mock(ConfigManager.class);
+ final ConsensusManager consensusManager =
Mockito.mock(ConsensusManager.class);
+ Mockito.when(env.getConfigManager()).thenReturn(configManager);
+
Mockito.when(configManager.getConsensusManager()).thenReturn(consensusManager);
+ Mockito.when(consensusManager.write(Mockito.any(ConfigPhysicalPlan.class)))
+ .thenReturn(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+ return consensusManager;
+ }
+
+ private static void assertPipeMetaChangeWasWritten(
+ final ConsensusManager consensusManager, final String pipeName, final
String error)
+ throws Exception {
+ final ArgumentCaptor<ConfigPhysicalPlan> planCaptor =
+ ArgumentCaptor.forClass(ConfigPhysicalPlan.class);
+ Mockito.verify(consensusManager).write(planCaptor.capture());
+ assertTrue(planCaptor.getValue() instanceof PipeHandleMetaChangePlan);
+
+ final PipeRuntimeMeta persistedRuntimeMeta =
+ ((PipeHandleMetaChangePlan) planCaptor.getValue())
+ .getPipeMetaList().stream()
+ .filter(meta ->
pipeName.equals(meta.getStaticMeta().getPipeName()))
+ .findFirst()
+ .orElseThrow(AssertionError::new)
+ .getRuntimeMeta();
+ assertEquals(PipeStatus.STOPPED, persistedRuntimeMeta.getStatus().get());
+ assertTrue(persistedRuntimeMeta.getIsStoppedByRuntimeException());
+ assertEquals(
+ error,
persistedRuntimeMeta.getNodeId2PipeRuntimeExceptionMap().get(1).getMessage());
+ }
+
@Test
public void serializeDeserializeTest() {
PublicBAOS byteArrayOutputStream = new PublicBAOS();
@@ -70,4 +279,12 @@ public class StartPipeProcedureV2Test {
fail();
}
}
+
+ private static class TestableStartPipeProcedureV2 extends
StartPipeProcedureV2 {
+ private TestableStartPipeProcedureV2(final String pipeName, final
PipeTaskInfo taskInfo)
+ throws Exception {
+ super(pipeName);
+ pipeTaskInfo = new AtomicReference<>(taskInfo);
+ }
+ }
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java
index ca7ac8d2ba0..5f65556398d 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java
@@ -460,6 +460,37 @@ public class PipeSinkTest {
() -> assertWriteBackSinkTargetDatabaseValid("root." +
"a".repeat(60)));
}
+ @Test
+ public void testWriteBackSinkDatabaseValidationAfterCloningParameters()
throws Exception {
+ for (final String databaseKey : Arrays.asList("sink.database",
"connector.database")) {
+ final PipeParameters parameters =
+ new PipeParameters(Collections.singletonMap(databaseKey,
"root.target"))
+ .addOrReplaceEquivalentAttributesWithClone(
+ new PipeParameters(
+ Collections.singletonMap(
+ PipeSinkConstant.SINK_ENABLE_SEND_TSFILE_LIMIT,
"true")));
+
+ try (final WriteBackSink sink = new WriteBackSink()) {
+ sink.validate(new PipeParameterValidator(parameters));
+ sink.customize(
+ parameters,
+ new PipeTaskRuntimeConfiguration(new
PipeTaskSinkRuntimeEnvironment("pipe", 1L, 1)));
+ Assert.assertEquals(
+ "root.target", getWriteBackSinkDatabaseName(sink,
"targetTreeModelDatabaseName"));
+ }
+ }
+
+ final HashMap<String, String> conflictingAttributes = new HashMap<>();
+ conflictingAttributes.put("connector.database", "root.connectorTarget");
+ conflictingAttributes.put("sink.database", "root.sinkTarget");
+ try (final WriteBackSink sink = new WriteBackSink()) {
+ Assert.assertThrows(
+ PipeException.class,
+ () ->
+ sink.validate(new PipeParameterValidator(new
PipeParameters(conflictingAttributes))));
+ }
+ }
+
@Test
public void testWriteBackSinkTargetDatabaseCustomization() throws Exception {
try (final WriteBackSink sink =
createCustomizedWriteBackSink("TestTarget")) {
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/pipe/ShowPipeTaskTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/pipe/ShowPipeTaskTest.java
index 0822b51f1cb..a591627e588 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/pipe/ShowPipeTaskTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/pipe/ShowPipeTaskTest.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.queryengine.plan.execution.config.sys.pipe;
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus;
import org.apache.iotdb.commons.schema.column.ColumnHeaderConstant;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipeInfo;
import org.apache.iotdb.db.queryengine.plan.execution.config.ConfigTaskResult;
@@ -41,6 +42,8 @@ public class ShowPipeTaskTest {
@Test
public void testBuildTSBlockWritesRuntimeColumns() throws Exception {
final TShowPipeInfo degradedPipe = createPipeInfo("degraded_pipe");
+ degradedPipe.setState(PipeStatus.STOPPED.name());
+ degradedPipe.setExceptionMessage("nodeIds: [1], construction failed");
degradedPipe.setIsDegraded(true);
final Map<String, Long> recentFailures = new HashMap<>();
recentFailures.put("network_timeout", 10L);
@@ -62,6 +65,10 @@ public class ShowPipeTaskTest {
assertEquals(
ColumnHeaderConstant.RECENT_FAILURES,
result.getResultSetHeader().getRespColumns().get(10));
assertEquals(3, resultSet.getPositionCount());
+ assertEquals(PipeStatus.STOPPED.name(),
resultSet.getColumn(2).getBinary(0).toString());
+ assertEquals(
+ "nodeIds: [1], construction failed",
resultSet.getColumn(6).getBinary(0).toString());
+ assertEquals("", resultSet.getColumn(6).getBinary(1).toString());
assertTrue(resultSet.getColumn(9).getBoolean(0));
assertFalse(resultSet.getColumn(9).getBoolean(1));
assertTrue(resultSet.getColumn(9).isNull(2));
