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 ced1e9ee13ce5..afa30ada69601 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.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 void setUp() { 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 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 bae1e57a95856..97580ffa4524b 100644 --- a/iotdb-api/pipe-api/pom.xml +++ b/iotdb-api/pipe-api/pom.xml @@ -35,6 +35,11 @@ ${tsfile.version} provided + + junit + junit + test + 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 99d547a23c6d2..d4e6c866d455e 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 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 PipeParameterValidator validateSynonymAttributes( final List lhsAttributes, final List 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 lhsExistingKeys = getExistingAttributeKeys(lhsAttributes); + final Set 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 PipeParameterValidator validateSynonymAttributes( return this; } + private Set getExistingAttributeKeys(final List 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 b6d59c827f000..bda09613c3d05 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 PipeParameters addOrReplaceEquivalentAttributesWithClone(final PipeParame return new PipeParameters(thisMap); } - private static class KeyReducer { + static class KeyReducer { private static final Set FIRST_PREFIXES = new HashSet<>(); private static final Set 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 0000000000000..5c5e6a1a07aab --- /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 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 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 a7cd4b6298b5f..f7fd6c5046b96 100644 --- a/iotdb-core/confignode/pom.xml +++ b/iotdb-core/confignode/pom.xml @@ -32,6 +32,7 @@ ${confignode.test.skip} false ${confignode.test.skip} + en @@ -168,6 +169,37 @@ + + com.diffplug.spotless + spotless-maven-plugin + + + + src/main/java/**/*.java + src/test/java/**/*.java + src/main/i18n/**/*.java + + + + + + org.codehaus.mojo + build-helper-maven-plugin + + + add-i18n-source + generate-sources + + add-source + + + + ${project.basedir}/src/main/i18n/${i18n.locale} + + + + + org.apache.maven.plugins maven-assembly-plugin @@ -241,6 +273,12 @@ + + with-zh-locale + + zh + + skipConfigNodeTests 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 0000000000000..c80cb1f7cc465 --- /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 0000000000000..7ede13f18dcbf --- /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 3e7dcbdfa01c5..02e7aafb79c31 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.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.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 @@ private void setIsStoppedByRuntimeExceptionToFalseInternal(final String pipeName * *

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 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 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 respMap) { + final Map respMap, final PipeStaticMeta fallbackPipeStaticMeta) { boolean hasException = false; for (final Map.Entry respEntry : respMap.entrySet()) { @@ -854,50 +880,84 @@ private boolean recordDataNodePushPipeMetaExceptionsInternal( 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 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 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 5587b51af8d65..22b75a9067a6c 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.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 static String parsePushPipeMetaExceptionForPipe( .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 static String parsePushPipeMetaExceptionForPipe( } 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 static String parsePushPipeMetaExceptionForPipe( .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 static String parsePushPipeMetaExceptionForPipe( 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 @@ protected Map pushPipeMetaToDataNodesBestEffortAndGe pipeMetaBinaryList, this::setPendingDataNodeIds); } + protected void writePipeMetaChangesToConfigNodeConsensus(final ConfigNodeProcedureEnv env) { + final List 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 9764a3af3a5e2..d5e49878d20e3 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 void executeFromWriteConfigNodeConsensus(ConfigNodeProcedureEnv env) { if (response.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { throw new PipeException(response.getMessage()); } + pipeTaskInfo.get().markPipeMetaExceptionsPersisted(); } @Override @@ -136,8 +137,11 @@ public void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws PipeException, IOException { LOGGER.debug("PipeMetaSyncProcedure: executeFromOperateOnDataNodes"); + persistPendingPipeMetaExceptions(env); + Map 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 f7ed7496f773e..c6369364d0d4c 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.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 void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws IOE 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 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 ad18cc95bb050..d809c1668ed7d 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.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 void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws IOE final String pipeName = createPipeRequest.getPipeName(); LOGGER.info("CreatePipeProcedureV2: executeFromOperateOnDataNodes({})", pipeName); - final String exceptionMessage = - parsePushPipeMetaExceptionForPipe(pipeName, pushSinglePipeMetaToDataNodes(pipeName, env)); + persistPendingPipeMetaExceptions(env); + + final Map 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 962aa341fc750..a6f17e4aecfd9 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 @@ 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 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 void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws IOE 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 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 void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws IOE } } - private void writePipeMetaChangesToConfigNodeConsensus(final ConfigNodeProcedureEnv env) { - final List 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 068f85a0a3d10..72fa84df6906b 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 @@ 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 void testConvertToTShowPipeRespIncludesPreDeleteStatus() { 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 6d933dbc41311..80b7fd988781f 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.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 void testRecordDataNodePushPipeMetaExceptionsMarksRunningPipeForAutoResta 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 void testRecordDataNodePushPipeMetaExceptionsKeepsUserStoppedPipeOutOfAut 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 39da89ffe76a3..f9011fc1cb0d7 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 44b96284a15c1..7175308c555eb 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 void serializeDeserializeTest() { 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 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 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 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 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 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 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 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 responses; + + private TestablePipeMetaSyncProcedure( + final PipeTaskInfo taskInfo, final Map responses) { + this.pipeTaskInfo = new AtomicReference<>(taskInfo); + this.responses = responses; + } + + @Override + protected Map 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 ac6ca8c740c36..d34b050b61e85 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 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 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 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 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 4584d75753ac0..5f9509363e45f 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 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 void serializeDeserializeTest() { 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 62dca5a762ebf..a3db60fce14a9 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.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.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 @@ private PipeRawTabletInsertionEvent createPipeRawTabletInsertionEvent( 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 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 6bd8676199c17..47c0f88b26441 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 recentFailures = new HashMap<>(); recentFailures.put("network_timeout", 10L); recentFailures.put("memory_timeout", 15L); @@ -55,6 +58,10 @@ public void testBuildTSBlockWritesRecentFailures() throws Exception { 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());