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 7f6e63f726f [To dev/1.3] Fix write-back sink history failure state
handling (#19072)
7f6e63f726f is described below
commit 7f6e63f726fff9cfd4a112d80a5f6930192abb27
Author: Caideyipi <[email protected]>
AuthorDate: Mon Nov 2 16:20:49 2026 +0800
[To dev/1.3] Fix write-back sink history failure state handling (#19072)
* 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
(cherry picked from commit 595aa8442bb50fc8bfafdb8f11d40a94e1b3c1bd)
* Adapt key reducer visibility for the 1.3 backport
* Add test dependency for the Pipe API backport
---
.../pipe/it/autocreate/IoTDBPipeAutoSplitIT.java | 55 +++++
iotdb-api/pipe-api/pom.xml | 5 +
.../parameter/PipeParameterValidator.java | 24 ++-
.../api/customizer/parameter/PipeParameters.java | 2 +-
.../parameter/PipeParameterValidatorTest.java | 126 +++++++++++
iotdb-core/confignode/pom.xml | 38 ++++
.../iotdb/confignode/i18n/ProcedureMessages.java | 32 +++
.../iotdb/confignode/i18n/ProcedureMessages.java | 32 +++
.../confignode/persistence/pipe/PipeTaskInfo.java | 134 ++++++++----
.../impl/pipe/AbstractOperatePipeProcedureV2.java | 83 +++++--
.../impl/pipe/runtime/PipeMetaSyncProcedure.java | 4 +
.../impl/pipe/task/AlterPipeProcedureV2.java | 20 +-
.../impl/pipe/task/CreatePipeProcedureV2.java | 10 +-
.../impl/pipe/task/StartPipeProcedureV2.java | 50 ++---
.../consensus/response/pipe/PipeTableRespTest.java | 20 ++
.../pipe/PipeTaskInfoAutoRestartTest.java | 56 +++++
.../pipe/AbstractOperatePipeProcedureV2Test.java | 24 +++
.../pipe/runtime/PipeMetaSyncProcedureTest.java | 239 +++++++++++++++++++++
.../impl/pipe/task/CreatePipeProcedureV2Test.java | 144 +++++++++++++
.../impl/pipe/task/StartPipeProcedureV2Test.java | 215 ++++++++++++++++++
.../apache/iotdb/db/pipe/sink/PipeSinkTest.java | 36 ++++
.../config/sys/pipe/ShowPipeTaskTest.java | 7 +
22 files changed, 1260 insertions(+), 96 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAutoSplitIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAutoSplitIT.java
index ced1e9ee13c..afa30ada696 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAutoSplitIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAutoSplitIT.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.pipe.it.autocreate;
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;
@@ -40,6 +41,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;
@@ -77,6 +79,59 @@ public class IoTDBPipeAutoSplitIT extends
AbstractPipeDualAutoIT {
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()).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/pom.xml b/iotdb-api/pipe-api/pom.xml
index bae1e57a958..97580ffa452 100644
--- a/iotdb-api/pipe-api/pom.xml
+++ b/iotdb-api/pipe-api/pom.xml
@@ -35,6 +35,11 @@
<version>${tsfile.version}</version>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<profiles>
<profile>
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 99d547a23c6..d4e6c866d45 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
@@ -25,6 +25,7 @@ import
org.apache.iotdb.pipe.api.exception.PipeParameterNotValidException;
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;
@@ -57,9 +58,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(
"Cannot specify both %s and %s at the same time", lhsAttributes,
rhsAttributes));
@@ -74,6 +80,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/main/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameters.java
b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameters.java
index b6d59c827f0..bda09613c3d 100644
---
a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameters.java
+++
b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameters.java
@@ -355,7 +355,7 @@ public class PipeParameters {
return new PipeParameters(thisMap);
}
- private static class KeyReducer {
+ static class KeyReducer {
private static final Set<String> FIRST_PREFIXES = new HashSet<>();
private static final Set<String> SECOND_PREFIXES = new HashSet<>();
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/pom.xml b/iotdb-core/confignode/pom.xml
index a7cd4b6298b..f7fd6c5046b 100644
--- a/iotdb-core/confignode/pom.xml
+++ b/iotdb-core/confignode/pom.xml
@@ -32,6 +32,7 @@
<confignode.it.skip>${confignode.test.skip}</confignode.it.skip>
<confignode.test.skip>false</confignode.test.skip>
<confignode.ut.skip>${confignode.test.skip}</confignode.ut.skip>
+ <i18n.locale>en</i18n.locale>
</properties>
<dependencies>
<dependency>
@@ -168,6 +169,37 @@
</dependencies>
<build>
<plugins>
+ <plugin>
+ <groupId>com.diffplug.spotless</groupId>
+ <artifactId>spotless-maven-plugin</artifactId>
+ <configuration>
+ <java>
+ <includes>
+ <include>src/main/java/**/*.java</include>
+ <include>src/test/java/**/*.java</include>
+ <include>src/main/i18n/**/*.java</include>
+ </includes>
+ </java>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>add-i18n-source</id>
+ <phase>generate-sources</phase>
+ <goals>
+ <goal>add-source</goal>
+ </goals>
+ <configuration>
+ <sources>
+
<source>${project.basedir}/src/main/i18n/${i18n.locale}</source>
+ </sources>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
@@ -241,6 +273,12 @@
</plugins>
</build>
<profiles>
+ <profile>
+ <id>with-zh-locale</id>
+ <properties>
+ <i18n.locale>zh</i18n.locale>
+ </properties>
+ </profile>
<profile>
<id>skipConfigNodeTests</id>
<activation>
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
new file mode 100644
index 00000000000..c80cb1f7cc4
--- /dev/null
+++
b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
@@ -0,0 +1,32 @@
+/*
+ * 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.confignode.i18n;
+
+public final class ProcedureMessages {
+
+ private ProcedureMessages() {}
+
+ public static final String
+
EXCEPTION_INTERNAL_ERROR_WHILE_PROCESSING_PUSHPIPEMETA_ON_DATANODE_5983BD5F =
+ "Internal error while processing pushPipeMeta on DataNode.";
+ public static final String
+
LOG_FAILED_IN_THE_WRITE_API_EXECUTING_THE_CONSENSUS_LAYER_DUE_TO_8DDF6FE9 =
+ "Failed in the write API executing the consensus layer due to: ";
+}
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
new file mode 100644
index 00000000000..7ede13f18dc
--- /dev/null
+++
b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
@@ -0,0 +1,32 @@
+/*
+ * 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.confignode.i18n;
+
+public final class ProcedureMessages {
+
+ private ProcedureMessages() {}
+
+ public static final String
+
EXCEPTION_INTERNAL_ERROR_WHILE_PROCESSING_PUSHPIPEMETA_ON_DATANODE_5983BD5F =
+ "处理 DataNode 上的 pushPipeMeta 时发生内部错误。";
+ public static final String
+
LOG_FAILED_IN_THE_WRITE_API_EXECUTING_THE_CONSENSUS_LAYER_DUE_TO_8DDF6FE9 =
+ "执行共识层写入 API 时失败，原因：";
+}
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 3e7dcbdfa01..02e7aafb79c 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
@@ -50,6 +50,7 @@ import
org.apache.iotdb.confignode.consensus.request.write.pipe.task.OperateMult
import
org.apache.iotdb.confignode.consensus.request.write.pipe.task.SetPipeStatusPlanV2;
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.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;
@@ -57,6 +58,7 @@ import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
import org.apache.iotdb.confignode.service.ConfigNode;
import org.apache.iotdb.consensus.common.DataSet;
import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaResp;
+import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaRespExceptionMessage;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -91,6 +93,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;
@@ -829,22 +834,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()) {
@@ -854,50 +880,84 @@ 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;
}
- resp.getExceptionMessages().stream()
- .filter(message ->
pipeMetaKeeper.containsPipeMeta(message.getPipeName()))
- .forEach(
- message -> {
- final PipeRuntimeMeta runtimeMeta =
-
pipeMetaKeeper.getPipeMeta(message.getPipeName()).getRuntimeMeta();
-
- if
(PipeStatus.PRE_DELETE.equals(runtimeMeta.getStatus().get())) {
- return;
- }
-
- // 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()) {
- return;
- }
-
- // 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()));
- }
- });
+ 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 =
pipeMetaKeeper.getPipeMeta(messagePipeName);
+ if (pipeMeta == null) {
+ continue;
+ }
+
+ recordDataNodePushPipeMetaException(
+ pipeMeta, dataNodeId, message.getMessage(),
message.getTimeStamp());
+ }
}
}
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.getPipeName());
+ if (pipeMeta != null
+ && pipeMeta.getStaticMeta().getCreationTime()
+ == fallbackPipeStaticMeta.getCreationTime()) {
+ 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 5587b51af8d..22b75a9067a 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
@@ -19,8 +19,11 @@
package org.apache.iotdb.confignode.procedure.impl.pipe;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta;
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.ProcedureMessages;
import
org.apache.iotdb.confignode.manager.pipe.metric.overview.PipeProcedureMetrics;
import org.apache.iotdb.confignode.persistence.pipe.PipeTaskInfo;
import org.apache.iotdb.confignode.procedure.Procedure;
@@ -30,6 +33,7 @@ import
org.apache.iotdb.confignode.procedure.impl.node.AbstractNodeProcedure;
import
org.apache.iotdb.confignode.procedure.impl.pipe.runtime.PipeMetaSyncProcedure;
import org.apache.iotdb.confignode.procedure.state.ProcedureLockState;
import
org.apache.iotdb.confignode.procedure.state.pipe.task.OperatePipeTaskState;
+import org.apache.iotdb.consensus.exception.ConsensusException;
import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaResp;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.exception.PipeException;
@@ -631,14 +635,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(".");
@@ -654,7 +660,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(
@@ -669,16 +675,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()));
+ }
}
});
@@ -698,6 +705,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
@@ -717,6 +732,38 @@ 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(
+ ProcedureMessages
+
.LOG_FAILED_IN_THE_WRITE_API_EXECUTING_THE_CONSENSUS_LAYER_DUE_TO_8DDF6FE9,
+ 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 9764a3af3a5..d5e49878d20 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
@@ -129,6 +129,7 @@ public class PipeMetaSyncProcedure extends
AbstractOperatePipeProcedureV2 {
if (response.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
throw new PipeException(response.getMessage());
}
+ pipeTaskInfo.get().markPipeMetaExceptionsPersisted();
}
@Override
@@ -136,8 +137,11 @@ public class PipeMetaSyncProcedure extends
AbstractOperatePipeProcedureV2 {
throws PipeException, IOException {
LOGGER.debug("PipeMetaSyncProcedure: executeFromOperateOnDataNodes");
+ persistPendingPipeMetaExceptions(env);
+
Map<Integer, TPushPipeMetaResp> respMap =
pushPipeMetaToDataNodesBestEffortAndGetResponse(env);
if (pipeTaskInfo.get().recordDataNodePushPipeMetaExceptions(respMap)) {
+ writePipeMetaChangesToConfigNodeConsensus(env);
throw new PipeException(
String.format(
"Failed to push pipe meta to dataNodes, details: %s",
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 f7ed7496f77..c6369364d0d 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
@@ -42,6 +42,7 @@ import
org.apache.iotdb.confignode.procedure.impl.pipe.PipeTaskOperation;
import org.apache.iotdb.confignode.procedure.store.ProcedureType;
import org.apache.iotdb.confignode.rpc.thrift.TAlterPipeReq;
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;
@@ -317,14 +318,17 @@ public class AlterPipeProcedureV2 extends
AbstractOperatePipeProcedureV2 {
final String pipeName = alterPipeRequest.getPipeName();
LOGGER.info("AlterPipeProcedureV2: executeFromOperateOnDataNodes({})",
pipeName);
- final String exceptionMessage =
- parsePushPipeMetaExceptionForPipe(
- pipeName,
-
!PipeTaskAgent.isRealtimeOnlyPipe(currentPipeStaticMeta.getExtractorParameters())
- && PipeTaskAgent.isRealtimeOnlyPipe(
- updatedPipeStaticMeta.getExtractorParameters())
- ? pushSinglePipeMetaToDataNodes4Realtime(pipeName, env)
- : pushSinglePipeMetaToDataNodes(pipeName, env));
+ persistPendingPipeMetaExceptions(env);
+
+ final Map<Integer, TPushPipeMetaResp> respMap =
+
!PipeTaskAgent.isRealtimeOnlyPipe(currentPipeStaticMeta.getExtractorParameters())
+ &&
PipeTaskAgent.isRealtimeOnlyPipe(updatedPipeStaticMeta.getExtractorParameters())
+ ? pushSinglePipeMetaToDataNodes4Realtime(pipeName, env)
+ : pushSinglePipeMetaToDataNodes(pipeName, env);
+ if (pipeTaskInfo.get().recordDataNodePushPipeMetaExceptions(respMap,
updatedPipeStaticMeta)) {
+ writePipeMetaChangesToConfigNodeConsensus(env);
+ }
+ final String exceptionMessage =
parsePushPipeMetaExceptionForPipe(pipeName, respMap);
if (!exceptionMessage.isEmpty()) {
LOGGER.warn(
"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 ad18cc95bb0..d809c1668ed 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
@@ -45,6 +45,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;
@@ -301,8 +302,13 @@ public class CreatePipeProcedureV2 extends
AbstractOperatePipeProcedureV2 {
final String pipeName = createPipeRequest.getPipeName();
LOGGER.info("CreatePipeProcedureV2: executeFromOperateOnDataNodes({})",
pipeName);
- final String exceptionMessage =
- parsePushPipeMetaExceptionForPipe(pipeName,
pushSinglePipeMetaToDataNodes(pipeName, env));
+ persistPendingPipeMetaExceptions(env);
+
+ final Map<Integer, TPushPipeMetaResp> respMap =
pushSinglePipeMetaToDataNodes(pipeName, env);
+ if (pipeTaskInfo.get().recordDataNodePushPipeMetaExceptions(respMap,
pipeStaticMeta)) {
+ writePipeMetaChangesToConfigNodeConsensus(env);
+ }
+ final String exceptionMessage =
parsePushPipeMetaExceptionForPipe(pipeName, respMap);
if (!exceptionMessage.isEmpty()) {
LOGGER.warn(
"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 962aa341fc7..a6f17e4aecf 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
@@ -21,14 +21,15 @@ 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.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.procedure.env.ConfigNodeProcedureEnv;
import
org.apache.iotdb.confignode.procedure.impl.pipe.AbstractOperatePipeProcedureV2;
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;
@@ -39,8 +40,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 {
@@ -106,8 +106,26 @@ public class StartPipeProcedureV2 extends
AbstractOperatePipeProcedureV2 {
final long exceptionsClearTime = System.currentTimeMillis();
final boolean isStoppedByRuntimeException =
pipeTaskInfo.get().isStoppedByRuntimeException(pipeName);
- final String exceptionMessage =
- parsePushPipeMetaExceptionForPipe(pipeName,
pushSinglePipeMetaToDataNodes(pipeName, env));
+ final boolean hadPendingPipeMetaExceptions =
+ pipeTaskInfo.get().hasPendingPipeMetaExceptionPersistence();
+ if (hadPendingPipeMetaExceptions) {
+ persistPendingPipeMetaExceptions(env);
+ }
+ final PipeMeta pipeMeta =
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(pipeName, env);
+ if (pipeTaskInfo.get().recordDataNodePushPipeMetaExceptions(respMap,
pipeStaticMeta)) {
+ writePipeMetaChangesToConfigNodeConsensus(env);
+ }
+ final String exceptionMessage =
parsePushPipeMetaExceptionForPipe(pipeName, respMap);
if (!exceptionMessage.isEmpty()) {
LOGGER.warn(
"Failed to start pipe {}, details: {}, metadata will be synchronized
later.",
@@ -127,28 +145,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("Failed in the write API executing the consensus layer due
to: ", 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("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 068f85a0a3d..72fa84df690 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;
@@ -176,4 +177,23 @@ public class PipeTableRespTest {
PipeStatus.PRE_DELETE.name(),
pipeTableResp.convertToTShowPipeResp().getPipeInfoList().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]"));
+ }
}
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 6d933dbc413..80b7fd98878 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
@@ -27,6 +27,7 @@ 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.write.pipe.task.CreatePipePlanV2;
import
org.apache.iotdb.confignode.consensus.request.write.pipe.task.SetPipeStatusPlanV2;
+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;
@@ -71,6 +72,60 @@ public class PipeTaskInfoAutoRestartTest {
Assert.assertEquals(PipeStatus.RUNNING, runtimeMeta.getStatus().get());
}
+ @Test
+ public void
testRecordDataNodePushPipeMetaExceptionsUsesFallbackForEmptyDetails() {
+ final String pipeName = "emptyDetailsPipe";
+ createPipe(pipeName, PipeStatus.RUNNING);
+ final PipeStaticMeta staticMeta =
pipeTaskInfo.getPipeMetaByPipeName(pipeName).getStaticMeta();
+ 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";
+ createPipe(pipeName, PipeStatus.RUNNING);
+ final PipeStaticMeta staticMeta =
pipeTaskInfo.getPipeMetaByPipeName(pipeName).getStaticMeta();
+ 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";
@@ -85,6 +140,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 39da89ffe76..f9011fc1cb0 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,23 +19,47 @@
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.Arrays;
+import java.util.Collections;
import java.util.HashSet;
+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..7175308c555 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,216 @@ 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);
+ 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 ac6ca8c740c..d34b050b61e 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,172 @@
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);
+ 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);
+ 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 4584d75753a..5f9509363e4 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,225 @@
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());
+
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());
+
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();
@@ -50,4 +257,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 62dca5a762e..a3db60fce14 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
@@ -31,6 +31,7 @@ import
org.apache.iotdb.db.pipe.sink.protocol.thrift.async.IoTDBDataRegionAsyncS
import
org.apache.iotdb.db.pipe.sink.protocol.thrift.sync.IoTDBDataRegionSyncSink;
import
org.apache.iotdb.db.pipe.sink.protocol.websocket.WebSocketConnectorServer;
import org.apache.iotdb.db.pipe.sink.protocol.websocket.WebSocketSink;
+import org.apache.iotdb.db.pipe.sink.protocol.writeback.WriteBackSink;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.exception.PipeException;
@@ -42,8 +43,10 @@ import org.junit.Assert;
import org.junit.Test;
import org.mockito.Mockito;
+import java.lang.reflect.Field;
import java.security.SecureRandom;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -279,4 +282,37 @@ public class PipeSinkTest {
return new PipeRawTabletInsertionEvent(
tablet, false, pipeName, creationTime, null, null, false);
}
+
+ @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)));
+ final Field databaseNameField =
+
WriteBackSink.class.getDeclaredField("targetTreeModelDatabaseName");
+ databaseNameField.setAccessible(true);
+ Assert.assertEquals("root.target", databaseNameField.get(sink));
+ }
+ }
+
+ 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))));
+ }
+ }
}
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 6bd8676199c..47c0f88b264 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.confignode.rpc.thrift.TShowPipeInfo;
import org.apache.iotdb.db.queryengine.common.header.ColumnHeaderConstant;
import org.apache.iotdb.db.queryengine.plan.execution.config.ConfigTaskResult;
@@ -39,6 +40,8 @@ public class ShowPipeTaskTest {
@Test
public void testBuildTSBlockWritesRecentFailures() throws Exception {
final TShowPipeInfo failedPipe = createPipeInfo("failed_pipe");
+ failedPipe.setState(PipeStatus.STOPPED.name());
+ failedPipe.setExceptionMessage("nodeIds: [1], construction failed");
final Map<String, Long> recentFailures = new HashMap<>();
recentFailures.put("network_timeout", 10L);
recentFailures.put("memory_timeout", 15L);
@@ -55,6 +58,10 @@ public class ShowPipeTaskTest {
assertEquals(
ColumnHeaderConstant.RECENT_FAILURES,
result.getResultSetHeader().getRespColumns().get(9));
assertEquals(2, 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());
assertEquals(
"{memory_timeout=15, network_timeout=10}",
resultSet.getColumn(9).getBinary(0).toString());
assertEquals("{}", resultSet.getColumn(9).getBinary(1).toString());
