From c98168f435544cf732a7fe6c721788cbc1320768 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 30 Oct 2026 09:46:33 +0800 Subject: [PATCH 1/3] Fix write-back sink history failure state handling --- .../auto/basic/IoTDBPipeAutoSplitIT.java | 56 +++++++ .../parameter/PipeParameterValidator.java | 24 ++- .../parameter/PipeParameterValidatorTest.java | 126 +++++++++++++++ .../confignode/i18n/ProcedureMessages.java | 2 + .../confignode/i18n/ProcedureMessages.java | 2 + .../persistence/pipe/PipeTaskInfo.java | 99 ++++++++---- .../pipe/AbstractOperatePipeProcedureV2.java | 73 ++++++--- .../pipe/runtime/PipeMetaSyncProcedure.java | 1 + .../impl/pipe/task/AlterPipeProcedureV2.java | 7 +- .../impl/pipe/task/CreatePipeProcedureV2.java | 10 +- .../impl/pipe/task/StartPipeProcedureV2.java | 37 +---- .../response/pipe/PipeTableRespTest.java | 20 +++ .../pipe/PipeTaskInfoAutoRestartTest.java | 54 +++++++ .../AbstractOperatePipeProcedureV2Test.java | 24 +++ .../runtime/PipeMetaSyncProcedureTest.java | 104 ++++++++++++ .../pipe/task/CreatePipeProcedureV2Test.java | 87 ++++++++++ .../pipe/task/StartPipeProcedureV2Test.java | 152 ++++++++++++++++++ .../iotdb/db/pipe/sink/PipeSinkTest.java | 31 ++++ .../config/sys/pipe/ShowPipeTaskTest.java | 7 + 19 files changed, 827 insertions(+), 89 deletions(-) create mode 100644 iotdb-api/pipe-api/src/test/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameterValidatorTest.java diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeAutoSplitIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeAutoSplitIT.java index 4b9012b88ab34..1df7afb2f1b2b 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeAutoSplitIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeAutoSplitIT.java @@ -20,6 +20,7 @@ package org.apache.iotdb.pipe.it.dual.treemodel.auto.basic; import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus; import org.apache.iotdb.confignode.rpc.thrift.TShowPipeInfo; import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq; import org.apache.iotdb.consensus.ConsensusFactory; @@ -42,6 +43,7 @@ import java.sql.Statement; import java.util.Arrays; import java.util.Collections; +import java.util.HashSet; import java.util.List; import java.util.Objects; @@ -82,6 +84,60 @@ public 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')"), + 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().setUserName(SessionConfig.DEFAULT_USER)).pipeInfoList; + final TShowPipeInfo realtimePipe = + pipeInfoList.stream() + .filter(pipeInfo -> "writeback_realtime".equals(pipeInfo.id)) + .findFirst() + .orElseThrow(AssertionError::new); + Assert.assertEquals(PipeStatus.RUNNING.name(), realtimePipe.getState()); + Assert.assertEquals("", realtimePipe.getExceptionMessage()); + } + + TestUtils.executeNonQueries( + senderEnv, + Collections.singletonList( + "insert into root.writeback_source.d1(time,s1) aligned values " + + "(6,6),(7,7),(8,8),(9,9)"), + null); + + TestUtils.assertDataEventuallyOnEnv( + senderEnv, + "select s1 from root.writeback_target.d1", + "Time,root.writeback_target.d1.s1,", + new HashSet<>( + Arrays.asList("1,1,", "2,2,", "3,3,", "4,4,", "5,5,", "6,6,", "7,7,", "8,8,", "9,9,")), + 90); + } + @Test public void testSingleEnv() throws Exception { final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0); diff --git a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameterValidator.java b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameterValidator.java index 57894ed244b13..3e0562bf45373 100644 --- a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameterValidator.java +++ b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameterValidator.java @@ -26,6 +26,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.Set; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -58,9 +59,14 @@ public 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( PipeApiMessages.EXCEPTION_CANNOT_SPECIFY_BOTH_ARG_AND_ARG_AT_THE_SAME_TIME_7DA8858B, @@ -77,6 +83,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/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/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java index 488e7cc759841..86ca6cc356858 100644 --- a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java +++ b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java @@ -454,6 +454,8 @@ public final class ProcedureMessages { "Failed to push pipe meta list to data nodes, will retry later."; public static final String FAILED_TO_PUSH_PIPE_META_TO_DATANODES_DETAILS = "Failed to push pipe meta to dataNodes, details: %s"; + public static final String EXCEPTION_INTERNAL_ERROR_WHILE_PROCESSING_PUSHPIPEMETA_ON_DATANODE_5983BD5F = + "Internal error while processing pushPipeMeta on DataNode."; public static final String FAILED_TO_PUSH_TOPIC_META_TO_DATANODES_DETAILS = "Failed to push topic meta to dataNodes, details: %s"; public static final String FAILED_TO_REMOVE_DATA_NODE_BECAUSE_IT_IS_NOT_IN = diff --git a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java index d65ee58ec53a7..7ad8e2dca8dd8 100644 --- a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java +++ b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java @@ -439,6 +439,8 @@ public final class ProcedureMessages { "向 DataNode 推送 pipe 元数据列表失败，将稍后重试。"; public static final String FAILED_TO_PUSH_PIPE_META_TO_DATANODES_DETAILS = "向 DataNode 推送 pipe 元数据失败，详情：%s"; + public static final String EXCEPTION_INTERNAL_ERROR_WHILE_PROCESSING_PUSHPIPEMETA_ON_DATANODE_5983BD5F = + "处理 DataNode 上的 pushPipeMeta 时发生内部错误。"; public static final String FAILED_TO_PUSH_TOPIC_META_TO_DATANODES_DETAILS = "向 DataNode 推送 topic 元数据失败，详情：%s"; public static final String FAILED_TO_REMOVE_DATA_NODE_BECAUSE_IT_IS_NOT_IN = diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java index 2427e7f3329af..66f37b9e8bce0 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java @@ -55,6 +55,7 @@ import org.apache.iotdb.confignode.consensus.request.write.pipe.task.SetPipeStatusWithStoppedByRuntimeExceptionPlanV2; import org.apache.iotdb.confignode.consensus.response.pipe.task.PipeTableResp; import org.apache.iotdb.confignode.i18n.ConfigNodeMessages; +import org.apache.iotdb.confignode.i18n.ProcedureMessages; import org.apache.iotdb.confignode.manager.pipe.resource.PipeConfigNodeResourceManager; import org.apache.iotdb.confignode.procedure.impl.pipe.runtime.PipeHandleMetaChangeProcedure; import org.apache.iotdb.confignode.rpc.thrift.TAlterPipeReq; @@ -1144,16 +1145,25 @@ private void setIsStoppedByRuntimeExceptionToFalseInternal(final String pipeName */ 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); + return recordDataNodePushPipeMetaExceptionsInternal(respMap, fallbackPipeStaticMeta); } finally { releaseWriteLock(); } } private boolean recordDataNodePushPipeMetaExceptionsInternal( - final Map respMap) { + final Map respMap, final PipeStaticMeta fallbackPipeStaticMeta) { boolean hasException = false; for (final Map.Entry respEntry : respMap.entrySet()) { @@ -1163,46 +1173,44 @@ 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()) { + if (fallbackPipeStaticMeta != null) { + final PipeMeta pipeMeta = pipeMetaKeeper.getPipeMeta(fallbackPipeStaticMeta); + if (pipeMeta != null) { + final String statusMessage = resp.getStatus().getMessage(); + recordDataNodePushPipeMetaException( + pipeMeta, + dataNodeId, + statusMessage == null || statusMessage.isEmpty() + ? ProcedureMessages + .EXCEPTION_INTERNAL_ERROR_WHILE_PROCESSING_PUSHPIPEMETA_ON_DATANODE_5983BD5F + : statusMessage, + System.currentTimeMillis()); + } + } continue; } for (final TPushPipeMetaRespExceptionMessage message : resp.getExceptionMessages()) { - final PipeMeta pipeMeta = - message.isSetCreationTime() - ? pipeMetaKeeper.getPipeMeta(message.getPipeName(), message.getCreationTime()) - : pipeMetaKeeper.getPipeMeta(message.getPipeName()); - if (pipeMeta == null) { - continue; - } - - final PipeRuntimeMeta runtimeMeta = pipeMeta.getRuntimeMeta(); - - if (PipeStatus.PRE_DELETE.equals(runtimeMeta.getStatus().get())) { - continue; + final String messagePipeName = message.getPipeName(); + final PipeMeta pipeMeta; + if (messagePipeName == null || messagePipeName.isEmpty()) { + pipeMeta = + fallbackPipeStaticMeta == null + ? null + : pipeMetaKeeper.getPipeMeta(fallbackPipeStaticMeta); + } else { + pipeMeta = + message.isSetCreationTime() + ? pipeMetaKeeper.getPipeMeta(messagePipeName, message.getCreationTime()) + : pipeMetaKeeper.getPipeMeta(messagePipeName); } - - // 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()) { + if (pipeMeta == null) { continue; } - // 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())); - } + recordDataNodePushPipeMetaException( + pipeMeta, dataNodeId, message.getMessage(), message.getTimeStamp()); } } } @@ -1210,6 +1218,29 @@ private boolean recordDataNodePushPipeMetaExceptionsInternal( return hasException; } + 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 0710fe345b85d..01d7e2c03b77c 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java @@ -21,9 +21,12 @@ import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId; import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType; +import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta; import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta; import org.apache.iotdb.commons.pipe.config.constant.SystemConstant; +import org.apache.iotdb.confignode.consensus.request.write.pipe.runtime.PipeHandleMetaChangePlan; +import org.apache.iotdb.confignode.i18n.ConfigNodeMessages; import org.apache.iotdb.confignode.i18n.ProcedureMessages; import org.apache.iotdb.confignode.manager.pipe.metric.overview.PipeProcedureMetrics; import org.apache.iotdb.confignode.persistence.pipe.PipeTaskInfo; @@ -36,6 +39,7 @@ import org.apache.iotdb.confignode.procedure.state.pipe.task.OperatePipeTaskState; import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema; import org.apache.iotdb.confignode.service.ConfigNode; +import org.apache.iotdb.consensus.exception.ConsensusException; import org.apache.iotdb.db.pipe.source.dataregion.DataRegionListeningFilter; import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaResp; import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters; @@ -655,14 +659,16 @@ public 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("."); @@ -678,7 +684,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( @@ -693,16 +699,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())); + } } }); @@ -722,6 +729,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 @@ -741,6 +756,28 @@ 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(ConfigNodeMessages.FAILED_IN_THE_WRITE_API_EXECUTING_THE_CONSENSUS_LAYER_DUE, e); + response = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode()); + response.setMessage(e.getMessage()); + } + if (response.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + throw new PipeException(response.getMessage()); + } + } + 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 f18cfb6d5453f..7aceccdcccb94 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 @@ -213,6 +213,7 @@ public void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) Map respMap = pushPipeMetaToDataNodesBestEffortAndGetResponse(env); if (pipeTaskInfo.get().recordDataNodePushPipeMetaExceptions(respMap)) { + writePipeMetaChangesToConfigNodeConsensus(env); throw new PipeException( String.format( ProcedureMessages.FAILED_TO_PUSH_PIPE_META_TO_DATANODES_DETAILS, diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AlterPipeProcedureV2.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AlterPipeProcedureV2.java index f9c3ec2294632..94512de619d06 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AlterPipeProcedureV2.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AlterPipeProcedureV2.java @@ -377,8 +377,11 @@ public void executeFromOperateOnDataNodes(final ConfigNodeProcedureEnv env) thro final String pipeName = alterPipeRequest.getPipeName(); LOGGER.info(ProcedureMessages.ALTERPIPEPROCEDUREV2_EXECUTEFROMOPERATEONDATANODES, pipeName); - final String exceptionMessage = - parsePushPipeMetaExceptionForPipe(pipeName, pushAlteredPipeMetaToDataNodes(env)); + final Map respMap = pushAlteredPipeMetaToDataNodes(env); + if (pipeTaskInfo.get().recordDataNodePushPipeMetaExceptions(respMap, updatedPipeStaticMeta)) { + writePipeMetaChangesToConfigNodeConsensus(env); + } + final String exceptionMessage = parsePushPipeMetaExceptionForPipe(pipeName, respMap); if (!exceptionMessage.isEmpty()) { LOGGER.warn( ProcedureMessages.FAILED_TO_ALTER_PIPE_DETAILS_METADATA_WILL_BE_SYNCHRONIZED_LATER, diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java index f62402db84e5e..6a0d33d25d015 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java @@ -51,6 +51,7 @@ import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq; import org.apache.iotdb.confignode.rpc.thrift.TPermissionInfoResp; import org.apache.iotdb.consensus.exception.ConsensusException; +import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaResp; import org.apache.iotdb.pipe.api.PipePlugin; import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters; import org.apache.iotdb.pipe.api.exception.PipeException; @@ -411,9 +412,12 @@ public void executeFromOperateOnDataNodes(final ConfigNodeProcedureEnv env) thro final String pipeName = createPipeRequest.getPipeName(); LOGGER.info(ProcedureMessages.CREATEPIPEPROCEDUREV2_EXECUTEFROMOPERATEONDATANODES, pipeName); - final String exceptionMessage = - parsePushPipeMetaExceptionForPipe( - pipeName, pushSinglePipeMetaToDataNodes(pipeStaticMeta, env)); + final Map respMap = + pushSinglePipeMetaToDataNodes(pipeStaticMeta, env); + if (pipeTaskInfo.get().recordDataNodePushPipeMetaExceptions(respMap, pipeStaticMeta)) { + writePipeMetaChangesToConfigNodeConsensus(env); + } + final String exceptionMessage = parsePushPipeMetaExceptionForPipe(pipeName, respMap); if (!exceptionMessage.isEmpty()) { LOGGER.warn( ProcedureMessages.FAILED_TO_CREATE_PIPE_DETAILS_METADATA_WILL_BE_SYNCHRONIZED_LATER, diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/StartPipeProcedureV2.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/StartPipeProcedureV2.java index a3d0fb01d6967..c91d41b69c907 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 @@ -20,10 +20,8 @@ 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.i18n.ConfigNodeMessages; import org.apache.iotdb.confignode.i18n.ProcedureMessages; @@ -32,6 +30,7 @@ import org.apache.iotdb.confignode.procedure.impl.pipe.PipeTaskOperation; import org.apache.iotdb.confignode.procedure.store.ProcedureType; import org.apache.iotdb.consensus.exception.ConsensusException; +import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaResp; import org.apache.iotdb.pipe.api.exception.PipeException; import org.apache.iotdb.rpc.TSStatusCode; @@ -42,8 +41,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 { @@ -134,9 +132,12 @@ public void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws IOE ? pipeTaskInfo.get().getPipeMetaByPipeName(pipeName, isTableModel) : pipeTaskInfo.get().getPipeMetaByPipeName(pipeName)) .getStaticMeta(); - final String exceptionMessage = - parsePushPipeMetaExceptionForPipe( - pipeName, pushSinglePipeMetaToDataNodes(pipeStaticMeta, env)); + final Map respMap = + pushSinglePipeMetaToDataNodes(pipeStaticMeta, env); + if (pipeTaskInfo.get().recordDataNodePushPipeMetaExceptions(respMap, pipeStaticMeta)) { + writePipeMetaChangesToConfigNodeConsensus(env); + } + final String exceptionMessage = parsePushPipeMetaExceptionForPipe(pipeName, respMap); if (!exceptionMessage.isEmpty()) { LOGGER.warn( ProcedureMessages.FAILED_TO_START_PIPE_DETAILS_METADATA_WILL_BE_SYNCHRONIZED_LATER, @@ -163,28 +164,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(ConfigNodeMessages.FAILED_IN_THE_WRITE_API_EXECUTING_THE_CONSENSUS_LAYER_DUE, e); - response = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode()); - response.setMessage(e.getMessage()); - } - if (response.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { - throw new PipeException(response.getMessage()); - } - } - @Override public void rollbackFromValidateTask(ConfigNodeProcedureEnv env) { LOGGER.info(ProcedureMessages.STARTPIPEPROCEDUREV2_ROLLBACKFROMVALIDATETASK, pipeName); diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/response/pipe/PipeTableRespTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/response/pipe/PipeTableRespTest.java index b3ae595a5caa6..66bc79f3ceab3 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; @@ -199,6 +200,25 @@ public void testConvertToTShowPipeRespIncludesPreDeleteStatus() { Assert.assertEquals(PipeStatus.PRE_DELETE.name(), showPipeResult.get(0).getState()); } + @Test + public void testConvertToTShowPipeRespIncludesStoppedStateAndExceptionMessage() { + final PipeTableResp pipeTableResp = constructPipeTableResp(); + final PipeMeta pipeMeta = pipeTableResp.getAllPipeMeta().get(0); + pipeMeta.getRuntimeMeta().getStatus().set(PipeStatus.STOPPED); + pipeMeta + .getRuntimeMeta() + .getNodeId2PipeRuntimeExceptionMap() + .put( + 1, new PipeRuntimeCriticalException("construction failed", System.currentTimeMillis())); + + final TShowPipeInfo showPipeInfo = + pipeTableResp.convertToTShowPipeResp().getPipeInfoList().get(0); + + Assert.assertEquals(PipeStatus.STOPPED.name(), showPipeInfo.getState()); + Assert.assertTrue(showPipeInfo.getExceptionMessage().contains("construction failed")); + Assert.assertTrue(showPipeInfo.getExceptionMessage().contains("nodeIds: [1]")); + } + @Test public void testFilterByModelBeforeWhereClause() { TSStatus status = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfoAutoRestartTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfoAutoRestartTest.java index fd9540aa7f624..b5e443c891b68 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfoAutoRestartTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfoAutoRestartTest.java @@ -34,6 +34,7 @@ import org.apache.iotdb.confignode.consensus.request.write.pipe.task.DropPipePlanV2; import org.apache.iotdb.confignode.consensus.request.write.pipe.task.SetPipeStatusPlanV2; import org.apache.iotdb.confignode.i18n.ConfigNodeMessages; +import org.apache.iotdb.confignode.i18n.ProcedureMessages; import org.apache.iotdb.confignode.rpc.thrift.TAlterPipeReq; import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaResp; import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaRespExceptionMessage; @@ -80,6 +81,58 @@ public void testRecordDataNodePushPipeMetaExceptionsMarksRunningPipeForAutoResta Assert.assertEquals(PipeStatus.RUNNING, runtimeMeta.getStatus().get()); } + @Test + public void testRecordDataNodePushPipeMetaExceptionsUsesFallbackForEmptyDetails() { + final String pipeName = "emptyDetailsPipe"; + final PipeStaticMeta staticMeta = createPipe(pipeName, PipeStatus.RUNNING); + final TPushPipeMetaResp response = + new TPushPipeMetaResp() + .setStatus(new TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode())) + .setExceptionMessages(Collections.emptyList()); + + Assert.assertTrue( + pipeTaskInfo.recordDataNodePushPipeMetaExceptions( + Collections.singletonMap(DATA_NODE_ID, response), staticMeta)); + + final PipeRuntimeMeta runtimeMeta = + pipeTaskInfo.getPipeMetaByPipeName(pipeName).getRuntimeMeta(); + Assert.assertEquals(PipeStatus.STOPPED, runtimeMeta.getStatus().get()); + Assert.assertTrue(runtimeMeta.getIsStoppedByRuntimeException()); + Assert.assertTrue( + runtimeMeta + .getNodeId2PipeRuntimeExceptionMap() + .get(DATA_NODE_ID) + .getMessage() + .contains( + ProcedureMessages + .EXCEPTION_INTERNAL_ERROR_WHILE_PROCESSING_PUSHPIPEMETA_ON_DATANODE_5983BD5F)); + } + + @Test + public void testRecordDataNodePushPipeMetaExceptionsUsesFallbackForEmptyPipeName() { + final String pipeName = "emptyPipeNamePipe"; + final PipeStaticMeta staticMeta = createPipe(pipeName, PipeStatus.RUNNING); + final TPushPipeMetaRespExceptionMessage exceptionMessage = + new TPushPipeMetaRespExceptionMessage( + "", "failed to close schema listening queue", System.currentTimeMillis()); + final TPushPipeMetaResp response = + new TPushPipeMetaResp() + .setStatus(new TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode())) + .setExceptionMessages(Collections.singletonList(exceptionMessage)); + + Assert.assertTrue( + pipeTaskInfo.recordDataNodePushPipeMetaExceptions( + Collections.singletonMap(DATA_NODE_ID, response), staticMeta)); + + final PipeRuntimeMeta runtimeMeta = + pipeTaskInfo.getPipeMetaByPipeName(pipeName).getRuntimeMeta(); + Assert.assertEquals(PipeStatus.STOPPED, runtimeMeta.getStatus().get()); + Assert.assertTrue(runtimeMeta.getIsStoppedByRuntimeException()); + Assert.assertEquals( + "failed to close schema listening queue", + runtimeMeta.getNodeId2PipeRuntimeExceptionMap().get(DATA_NODE_ID).getMessage()); + } + @Test public void testRecordDataNodePushPipeMetaExceptionsKeepsUserStoppedPipeOutOfAutoRestart() { final String pipeName = "stoppedPipe"; @@ -94,6 +147,7 @@ public 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 ae0feec0f7250..a4a75d60dc747 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2Test.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2Test.java @@ -19,21 +19,45 @@ package org.apache.iotdb.confignode.procedure.impl.pipe; +import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.confignode.persistence.pipe.PipeTaskInfo; import org.apache.iotdb.confignode.procedure.Procedure; import org.apache.iotdb.confignode.procedure.impl.StateMachineProcedure; import org.apache.iotdb.confignode.procedure.state.pipe.task.OperatePipeTaskState; +import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaResp; +import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaRespExceptionMessage; import org.apache.iotdb.pipe.api.exception.PipeException; +import org.apache.iotdb.rpc.TSStatusCode; import org.junit.Assert; import org.junit.Test; import java.io.IOException; +import java.util.Collections; +import java.util.Map; import java.util.Set; import java.util.concurrent.atomic.AtomicReference; public class AbstractOperatePipeProcedureV2Test { + @Test + public void testParsePushPipeMetaExceptionForPipeIncludesUnscopedException() { + final TPushPipeMetaRespExceptionMessage exceptionMessage = + new TPushPipeMetaRespExceptionMessage( + "", "failed to close schema listening queue", System.currentTimeMillis()); + final TPushPipeMetaResp response = + new TPushPipeMetaResp() + .setStatus(new TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode())) + .setExceptionMessages(Collections.singletonList(exceptionMessage)); + + final String message = + AbstractOperatePipeProcedureV2.parsePushPipeMetaExceptionForPipe( + "targetPipe", Map.of(1, response)); + + Assert.assertTrue(message.contains("DataNodeId: 1")); + Assert.assertTrue(message.contains("Message: failed to close schema listening queue")); + } + @Test public void testSuccessfulStateDoesNotYield() throws Exception { final TestOperatePipeProcedure procedure = new TestOperatePipeProcedure(); diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeMetaSyncProcedureTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeMetaSyncProcedureTest.java index 44b96284a15c1..2daaed5945988 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,41 @@ 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.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 +78,82 @@ 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) + .setCreationTime(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()); + } + + 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 013376571ec2e..4d14ea30cb01d 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,115 @@ package org.apache.iotdb.confignode.procedure.impl.pipe.task; import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeRuntimeMeta; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus; import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant; +import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan; +import org.apache.iotdb.confignode.consensus.request.write.pipe.runtime.PipeHandleMetaChangePlan; import org.apache.iotdb.confignode.manager.ConfigManager; import org.apache.iotdb.confignode.manager.PermissionManager; +import org.apache.iotdb.confignode.manager.consensus.ConsensusManager; +import org.apache.iotdb.confignode.persistence.pipe.PipeTaskInfo; import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv; import org.apache.iotdb.confignode.procedure.store.ProcedureFactory; import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq; import org.apache.iotdb.confignode.rpc.thrift.TPermissionInfoResp; +import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaResp; +import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaRespExceptionMessage; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.tsfile.utils.PublicBAOS; import org.junit.Test; +import org.mockito.ArgumentCaptor; import org.mockito.Mockito; import java.io.DataOutputStream; import java.nio.ByteBuffer; +import java.util.Collections; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.atomic.AtomicReference; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; public class CreatePipeProcedureV2Test { + @Test + public void testDataNodeConstructionFailureIsRecorded() throws Exception { + final String pipeName = "failedHistoryPipe"; + final TCreatePipeReq request = createPipeRequest(pipeName); + final PipeTaskInfo taskInfo = new PipeTaskInfo(); + final CreatePipeProcedureV2 procedure = + new CreatePipeProcedureV2(request, new AtomicReference<>(taskInfo)); + procedure.executeFromCalculateInfoForTask(Mockito.mock(ConfigNodeProcedureEnv.class)); + taskInfo.createPipe(procedure.constructPlan()); + + final String error = "Failed to construct PipeConnector"; + final TPushPipeMetaRespExceptionMessage exceptionMessage = + new TPushPipeMetaRespExceptionMessage(request.getPipeName(), error, 1L) + .setCreationTime(procedure.getPipeStaticMeta().getCreationTime()); + final ConfigNodeProcedureEnv env = Mockito.mock(ConfigNodeProcedureEnv.class); + final ConsensusManager consensusManager = mockSuccessfulConsensusWrite(env); + Mockito.when(env.pushSinglePipeMetaToDataNodes(Mockito.any(ByteBuffer.class), Mockito.any())) + .thenReturn( + Collections.singletonMap( + 1, + new TPushPipeMetaResp( + new TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode())) + .setExceptionMessages(Collections.singletonList(exceptionMessage)))); + + procedure.executeFromOperateOnDataNodes(env); + + final PipeRuntimeMeta runtimeMeta = + taskInfo.getPipeMetaByPipeName(request.getPipeName()).getRuntimeMeta(); + assertEquals(PipeStatus.STOPPED, runtimeMeta.getStatus().get()); + assertTrue(runtimeMeta.getIsStoppedByRuntimeException()); + assertEquals(error, runtimeMeta.getNodeId2PipeRuntimeExceptionMap().get(1).getMessage()); + assertPipeMetaChangeWasWritten(consensusManager, request.getPipeName(), error); + } + + 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 a623442174230..82c708d866c9f 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,162 @@ package org.apache.iotdb.confignode.procedure.impl.pipe.task; +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeRuntimeMeta; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta; +import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan; +import org.apache.iotdb.confignode.consensus.request.write.pipe.runtime.PipeHandleMetaChangePlan; +import org.apache.iotdb.confignode.consensus.request.write.pipe.task.CreatePipePlanV2; +import org.apache.iotdb.confignode.consensus.request.write.pipe.task.SetPipeStatusPlanV2; +import org.apache.iotdb.confignode.manager.ConfigManager; +import org.apache.iotdb.confignode.manager.consensus.ConsensusManager; +import org.apache.iotdb.confignode.persistence.pipe.PipeTaskInfo; +import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv; import org.apache.iotdb.confignode.procedure.store.ProcedureFactory; +import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaResp; +import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaRespExceptionMessage; +import org.apache.iotdb.rpc.TSStatusCode; import org.apache.tsfile.utils.PublicBAOS; import org.junit.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.Mockito; import java.io.DataOutputStream; import java.nio.ByteBuffer; +import java.util.Collections; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicReference; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; public class StartPipeProcedureV2Test { + @Test + public void testDataNodeConstructionFailureIsRecorded() throws Exception { + final String pipeName = "startFailurePipe"; + final PipeTaskInfo taskInfo = new PipeTaskInfo(); + final PipeStaticMeta staticMeta = + new PipeStaticMeta( + pipeName, + System.currentTimeMillis(), + Collections.singletonMap("extractor", "iotdb-source"), + Collections.singletonMap("processor", "do-nothing-processor"), + Collections.singletonMap("connector", "write-back-sink")); + final PipeRuntimeMeta runtimeMeta = new PipeRuntimeMeta(new ConcurrentHashMap<>()); + runtimeMeta + .getConsensusGroupId2TaskMetaMap() + .put(1, new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1)); + taskInfo.createPipe(new CreatePipePlanV2(staticMeta, runtimeMeta)); + taskInfo.setPipeStatus(new SetPipeStatusPlanV2(pipeName, PipeStatus.RUNNING)); + + final StartPipeProcedureV2 procedure = new TestableStartPipeProcedureV2(pipeName, taskInfo); + final ConfigNodeProcedureEnv env = Mockito.mock(ConfigNodeProcedureEnv.class); + final ConsensusManager consensusManager = mockSuccessfulConsensusWrite(env); + final String error = "Failed to construct PipeConnector"; + final TPushPipeMetaRespExceptionMessage exceptionMessage = + new TPushPipeMetaRespExceptionMessage(pipeName, error, System.currentTimeMillis()) + .setCreationTime(staticMeta.getCreationTime()); + Mockito.when(env.pushSinglePipeMetaToDataNodes(Mockito.any(ByteBuffer.class), Mockito.any())) + .thenReturn( + Collections.singletonMap( + 1, + new TPushPipeMetaResp( + new TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode())) + .setExceptionMessages(Collections.singletonList(exceptionMessage)))); + + procedure.executeFromOperateOnDataNodes(env); + + final PipeRuntimeMeta recordedRuntimeMeta = + taskInfo.getPipeMetaByPipeName(pipeName).getRuntimeMeta(); + assertEquals(PipeStatus.STOPPED, recordedRuntimeMeta.getStatus().get()); + assertEquals( + error, recordedRuntimeMeta.getNodeId2PipeRuntimeExceptionMap().get(1).getMessage()); + assertPipeMetaChangeWasWritten(consensusManager, pipeName, error); + } + + @Test + public void testFailureWithoutExceptionDetailsIsNotTreatedAsSuccessfulStart() throws Exception { + final String pipeName = "startFailureWithoutDetailsPipe"; + final PipeTaskInfo taskInfo = new PipeTaskInfo(); + final PipeStaticMeta staticMeta = + new PipeStaticMeta( + pipeName, + System.currentTimeMillis(), + Collections.singletonMap("extractor", "iotdb-source"), + Collections.singletonMap("processor", "do-nothing-processor"), + Collections.singletonMap("connector", "write-back-sink")); + final PipeRuntimeMeta runtimeMeta = new PipeRuntimeMeta(new ConcurrentHashMap<>()); + runtimeMeta + .getConsensusGroupId2TaskMetaMap() + .put(1, new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1)); + taskInfo.createPipe(new CreatePipePlanV2(staticMeta, runtimeMeta)); + taskInfo.setPipeStatus(new SetPipeStatusPlanV2(pipeName, PipeStatus.RUNNING)); + + final StartPipeProcedureV2 procedure = new TestableStartPipeProcedureV2(pipeName, taskInfo); + final ConfigNodeProcedureEnv env = Mockito.mock(ConfigNodeProcedureEnv.class); + final ConsensusManager consensusManager = mockSuccessfulConsensusWrite(env); + final String error = "DataNode failed before returning pipe-specific details"; + Mockito.when(env.pushSinglePipeMetaToDataNodes(Mockito.any(ByteBuffer.class), Mockito.any())) + .thenReturn( + Collections.singletonMap( + 1, + new TPushPipeMetaResp( + new TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode()) + .setMessage(error)) + .setExceptionMessages(Collections.emptyList()))); + + procedure.executeFromOperateOnDataNodes(env); + + final PipeRuntimeMeta recordedRuntimeMeta = + taskInfo.getPipeMetaByPipeName(pipeName).getRuntimeMeta(); + assertEquals(PipeStatus.STOPPED, recordedRuntimeMeta.getStatus().get()); + assertTrue(recordedRuntimeMeta.getIsStoppedByRuntimeException()); + assertTrue( + recordedRuntimeMeta + .getNodeId2PipeRuntimeExceptionMap() + .get(1) + .getMessage() + .contains(error)); + assertPipeMetaChangeWasWritten(consensusManager, pipeName, error); + } + + 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(); @@ -70,4 +214,12 @@ public void serializeDeserializeLegacyFormatTest() { 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 ca7ac8d2ba032..5f65556398d55 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java @@ -460,6 +460,37 @@ public void testWriteBackSinkTargetDatabaseValidation() throws Exception { () -> assertWriteBackSinkTargetDatabaseValid("root." + "a".repeat(60))); } + @Test + public void testWriteBackSinkDatabaseValidationAfterCloningParameters() throws Exception { + for (final String databaseKey : Arrays.asList("sink.database", "connector.database")) { + final PipeParameters parameters = + new PipeParameters(Collections.singletonMap(databaseKey, "root.target")) + .addOrReplaceEquivalentAttributesWithClone( + new PipeParameters( + Collections.singletonMap( + PipeSinkConstant.SINK_ENABLE_SEND_TSFILE_LIMIT, "true"))); + + try (final WriteBackSink sink = new WriteBackSink()) { + sink.validate(new PipeParameterValidator(parameters)); + sink.customize( + parameters, + new PipeTaskRuntimeConfiguration(new PipeTaskSinkRuntimeEnvironment("pipe", 1L, 1))); + Assert.assertEquals( + "root.target", getWriteBackSinkDatabaseName(sink, "targetTreeModelDatabaseName")); + } + } + + final HashMap conflictingAttributes = new HashMap<>(); + conflictingAttributes.put("connector.database", "root.connectorTarget"); + conflictingAttributes.put("sink.database", "root.sinkTarget"); + try (final WriteBackSink sink = new WriteBackSink()) { + Assert.assertThrows( + PipeException.class, + () -> + sink.validate(new PipeParameterValidator(new PipeParameters(conflictingAttributes)))); + } + } + @Test public void testWriteBackSinkTargetDatabaseCustomization() throws Exception { try (final WriteBackSink sink = createCustomizedWriteBackSink("TestTarget")) { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/pipe/ShowPipeTaskTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/pipe/ShowPipeTaskTest.java index 0822b51f1cbeb..a591627e58816 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/pipe/ShowPipeTaskTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/pipe/ShowPipeTaskTest.java @@ -19,6 +19,7 @@ package org.apache.iotdb.db.queryengine.plan.execution.config.sys.pipe; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus; import org.apache.iotdb.commons.schema.column.ColumnHeaderConstant; import org.apache.iotdb.confignode.rpc.thrift.TShowPipeInfo; import org.apache.iotdb.db.queryengine.plan.execution.config.ConfigTaskResult; @@ -41,6 +42,8 @@ public class ShowPipeTaskTest { @Test public void testBuildTSBlockWritesRuntimeColumns() throws Exception { final TShowPipeInfo degradedPipe = createPipeInfo("degraded_pipe"); + degradedPipe.setState(PipeStatus.STOPPED.name()); + degradedPipe.setExceptionMessage("nodeIds: [1], construction failed"); degradedPipe.setIsDegraded(true); final Map recentFailures = new HashMap<>(); recentFailures.put("network_timeout", 10L); @@ -62,6 +65,10 @@ public void testBuildTSBlockWritesRuntimeColumns() throws Exception { assertEquals( ColumnHeaderConstant.RECENT_FAILURES, result.getResultSetHeader().getRespColumns().get(10)); assertEquals(3, resultSet.getPositionCount()); + assertEquals(PipeStatus.STOPPED.name(), resultSet.getColumn(2).getBinary(0).toString()); + assertEquals( + "nodeIds: [1], construction failed", resultSet.getColumn(6).getBinary(0).toString()); + assertEquals("", resultSet.getColumn(6).getBinary(1).toString()); assertTrue(resultSet.getColumn(9).getBoolean(0)); assertFalse(resultSet.getColumn(9).getBoolean(1)); assertTrue(resultSet.getColumn(9).isNull(2)); From e759aab86b3b9a7e3c0eb5f82199c28a7fd987ba Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 30 Oct 2026 11:47:13 +0800 Subject: [PATCH 2/3] Fix write-back sink auto-split test credentials --- .../pipe/it/dual/treemodel/auto/basic/IoTDBPipeAutoSplitIT.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeAutoSplitIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeAutoSplitIT.java index 1df7afb2f1b2b..c737ce32ec704 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeAutoSplitIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeAutoSplitIT.java @@ -99,7 +99,7 @@ public void testWriteBackSinkWithTargetDatabase() throws Exception { + "'forwarding-pipe-requests'='false'," + "'history.enable'='true','realtime.enable'='true') " + "with sink ('sink'='write-back-sink','sink.database'='root.writeback_target'," - + "'user'='root')"), + + "'user'='root','password'='root')"), null); TestUtils.assertDataEventuallyOnEnv( From 31d9b949d2947f09fca8ab0cf7346e7ac330f2f8 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 30 Oct 2026 15:04:54 +0800 Subject: [PATCH 3/3] Fix pipe metadata retry persistence --- .../persistence/pipe/PipeTaskInfo.java | 76 ++++++---- .../pipe/AbstractOperatePipeProcedureV2.java | 7 + .../pipe/runtime/PipeMetaSyncProcedure.java | 3 + .../impl/pipe/task/AlterPipeProcedureV2.java | 2 + .../impl/pipe/task/CreatePipeProcedureV2.java | 2 + .../impl/pipe/task/StartPipeProcedureV2.java | 24 +++- .../runtime/PipeMetaSyncProcedureTest.java | 136 ++++++++++++++++++ .../pipe/task/CreatePipeProcedureV2Test.java | 59 ++++++++ .../pipe/task/StartPipeProcedureV2Test.java | 65 +++++++++ 9 files changed, 344 insertions(+), 30 deletions(-) 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 66f37b9e8bce0..78040be82d53e 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 @@ -98,6 +98,9 @@ public class PipeTaskInfo implements SnapshotProcessor { private final PipeMetaKeeper pipeMetaKeeper; + // Tracks runtime exception changes that have not yet been persisted through consensus. + private volatile boolean pipeMetaExceptionPersistencePending; + // Pure in-memory object, not involved in snapshot serialization and deserialization. private final PipeTaskInfoVersion pipeTaskInfoVersion; // Accepts a username and returns its current stored password for pipe authentication. @@ -1139,6 +1142,7 @@ 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 @@ -1156,12 +1160,23 @@ public boolean recordDataNodePushPipeMetaExceptions( final Map respMap, final PipeStaticMeta fallbackPipeStaticMeta) { acquireWriteLock(); try { - return recordDataNodePushPipeMetaExceptionsInternal(respMap, fallbackPipeStaticMeta); + 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 PipeStaticMeta fallbackPipeStaticMeta) { boolean hasException = false; @@ -1174,37 +1189,29 @@ private boolean recordDataNodePushPipeMetaExceptionsInternal( hasException = true; if (!resp.isSetExceptionMessages() || resp.getExceptionMessages().isEmpty()) { - if (fallbackPipeStaticMeta != null) { - final PipeMeta pipeMeta = pipeMetaKeeper.getPipeMeta(fallbackPipeStaticMeta); - if (pipeMeta != null) { - final String statusMessage = resp.getStatus().getMessage(); - recordDataNodePushPipeMetaException( - pipeMeta, - dataNodeId, - statusMessage == null || statusMessage.isEmpty() - ? ProcedureMessages - .EXCEPTION_INTERNAL_ERROR_WHILE_PROCESSING_PUSHPIPEMETA_ON_DATANODE_5983BD5F - : statusMessage, - System.currentTimeMillis()); - } - } + final String statusMessage = resp.getStatus().getMessage(); + recordUnscopedDataNodePushPipeMetaException( + fallbackPipeStaticMeta, + dataNodeId, + statusMessage == null || statusMessage.isEmpty() + ? ProcedureMessages + .EXCEPTION_INTERNAL_ERROR_WHILE_PROCESSING_PUSHPIPEMETA_ON_DATANODE_5983BD5F + : statusMessage, + System.currentTimeMillis()); continue; } for (final TPushPipeMetaRespExceptionMessage message : resp.getExceptionMessages()) { final String messagePipeName = message.getPipeName(); - final PipeMeta pipeMeta; if (messagePipeName == null || messagePipeName.isEmpty()) { - pipeMeta = - fallbackPipeStaticMeta == null - ? null - : pipeMetaKeeper.getPipeMeta(fallbackPipeStaticMeta); - } else { - pipeMeta = - message.isSetCreationTime() - ? pipeMetaKeeper.getPipeMeta(messagePipeName, message.getCreationTime()) - : pipeMetaKeeper.getPipeMeta(messagePipeName); + recordUnscopedDataNodePushPipeMetaException( + fallbackPipeStaticMeta, dataNodeId, message.getMessage(), message.getTimeStamp()); + continue; } + final PipeMeta pipeMeta = + message.isSetCreationTime() + ? pipeMetaKeeper.getPipeMeta(messagePipeName, message.getCreationTime()) + : pipeMetaKeeper.getPipeMeta(messagePipeName); if (pipeMeta == null) { continue; } @@ -1218,6 +1225,25 @@ private boolean recordDataNodePushPipeMetaExceptionsInternal( return hasException; } + private void recordUnscopedDataNodePushPipeMetaException( + final PipeStaticMeta fallbackPipeStaticMeta, + final int dataNodeId, + final String message, + final long timeStamp) { + if (fallbackPipeStaticMeta == null) { + pipeMetaKeeper + .getPipeMetaList() + .forEach( + pipeMeta -> + recordDataNodePushPipeMetaException(pipeMeta, dataNodeId, message, timeStamp)); + } else { + final PipeMeta pipeMeta = pipeMetaKeeper.getPipeMeta(fallbackPipeStaticMeta); + if (pipeMeta != null) { + recordDataNodePushPipeMetaException(pipeMeta, dataNodeId, message, timeStamp); + } + } + } + private void recordDataNodePushPipeMetaException( final PipeMeta pipeMeta, final int dataNodeId, final String message, final long timeStamp) { final PipeRuntimeMeta runtimeMeta = pipeMeta.getRuntimeMeta(); 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 01d7e2c03b77c..c52864f43286a 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 @@ -776,6 +776,13 @@ protected void writePipeMetaChangesToConfigNodeConsensus(final ConfigNodeProcedu 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) { 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 7aceccdcccb94..0c578cf1b6b77 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeMetaSyncProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeMetaSyncProcedure.java @@ -204,6 +204,7 @@ public void executeFromWriteConfigNodeConsensus(ConfigNodeProcedureEnv env) { if (response.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { throw new PipeException(response.getMessage()); } + pipeTaskInfo.get().markPipeMetaExceptionsPersisted(); } @Override @@ -211,6 +212,8 @@ public void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws PipeException, IOException { LOGGER.debug(ProcedureMessages.PIPEMETASYNCPROCEDURE_EXECUTEFROMOPERATEONDATANODES); + persistPendingPipeMetaExceptions(env); + Map respMap = pushPipeMetaToDataNodesBestEffortAndGetResponse(env); if (pipeTaskInfo.get().recordDataNodePushPipeMetaExceptions(respMap)) { writePipeMetaChangesToConfigNodeConsensus(env); 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 94512de619d06..cf380ac08539a 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AlterPipeProcedureV2.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AlterPipeProcedureV2.java @@ -377,6 +377,8 @@ public void executeFromOperateOnDataNodes(final ConfigNodeProcedureEnv env) thro final String pipeName = alterPipeRequest.getPipeName(); LOGGER.info(ProcedureMessages.ALTERPIPEPROCEDUREV2_EXECUTEFROMOPERATEONDATANODES, pipeName); + persistPendingPipeMetaExceptions(env); + final Map respMap = pushAlteredPipeMetaToDataNodes(env); if (pipeTaskInfo.get().recordDataNodePushPipeMetaExceptions(respMap, updatedPipeStaticMeta)) { writePipeMetaChangesToConfigNodeConsensus(env); 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 6a0d33d25d015..40134cf6dd9f2 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 @@ -412,6 +412,8 @@ public void executeFromOperateOnDataNodes(final ConfigNodeProcedureEnv env) thro final String pipeName = createPipeRequest.getPipeName(); LOGGER.info(ProcedureMessages.CREATEPIPEPROCEDUREV2_EXECUTEFROMOPERATEONDATANODES, pipeName); + persistPendingPipeMetaExceptions(env); + final Map respMap = pushSinglePipeMetaToDataNodes(pipeStaticMeta, env); if (pipeTaskInfo.get().recordDataNodePushPipeMetaExceptions(respMap, pipeStaticMeta)) { 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 c91d41b69c907..31be648dcbd63 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 @@ -20,6 +20,7 @@ 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.task.SetPipeStatusPlanV2; @@ -127,11 +128,24 @@ public void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws IOE isTableModelSet ? pipeTaskInfo.get().isStoppedByRuntimeException(pipeName, isTableModel) : pipeTaskInfo.get().isStoppedByRuntimeException(pipeName); - final PipeStaticMeta pipeStaticMeta = - (isTableModelSet - ? pipeTaskInfo.get().getPipeMetaByPipeName(pipeName, isTableModel) - : pipeTaskInfo.get().getPipeMetaByPipeName(pipeName)) - .getStaticMeta(); + final boolean hadPendingPipeMetaExceptions = + pipeTaskInfo.get().hasPendingPipeMetaExceptionPersistence(); + if (hadPendingPipeMetaExceptions) { + persistPendingPipeMetaExceptions(env); + } + final PipeMeta pipeMeta = + isTableModelSet + ? pipeTaskInfo.get().getPipeMetaByPipeName(pipeName, isTableModel) + : pipeTaskInfo.get().getPipeMetaByPipeName(pipeName); + // A previous attempt may already have recorded a runtime stop before a failed consensus write + // or procedure recovery. Persist that state without treating a STOPPED meta push as a start. + if (pipeMeta.getRuntimeMeta().getStatus().get() != PipeStatus.RUNNING) { + if (!hadPendingPipeMetaExceptions) { + writePipeMetaChangesToConfigNodeConsensus(env); + } + return; + } + final PipeStaticMeta pipeStaticMeta = pipeMeta.getStaticMeta(); final Map respMap = pushSinglePipeMetaToDataNodes(pipeStaticMeta, env); if (pipeTaskInfo.get().recordDataNodePushPipeMetaExceptions(respMap, pipeStaticMeta)) { 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 2daaed5945988..9b41b7e8dbb09 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 @@ -48,6 +48,7 @@ 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; @@ -141,6 +142,141 @@ public void testDataNodePushFailureIsPersistedToConsensus() throws Exception { 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; 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 4d14ea30cb01d..ac90a2ddec15b 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 @@ -89,6 +89,65 @@ public void testDataNodeConstructionFailureIsRecorded() throws Exception { assertPipeMetaChangeWasWritten(consensusManager, request.getPipeName(), error); } + @Test + public void testRetriesFailurePersistenceBeforeSuccessfulMetaPush() throws Exception { + final String pipeName = "failedHistoryPipePersistenceRetry"; + final TCreatePipeReq request = createPipeRequest(pipeName); + final PipeTaskInfo taskInfo = new PipeTaskInfo(); + final CreatePipeProcedureV2 procedure = + new CreatePipeProcedureV2(request, new AtomicReference<>(taskInfo)); + procedure.executeFromCalculateInfoForTask(Mockito.mock(ConfigNodeProcedureEnv.class)); + taskInfo.createPipe(procedure.constructPlan()); + + final String error = "DataNode failed before returning pipe-specific details"; + final TPushPipeMetaRespExceptionMessage exceptionMessage = + new TPushPipeMetaRespExceptionMessage(request.getPipeName(), error, 1L) + .setCreationTime(procedure.getPipeStaticMeta().getCreationTime()); + final Map 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); 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 82c708d866c9f..8a1796d4fd320 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 @@ -143,6 +143,71 @@ public void testFailureWithoutExceptionDetailsIsNotTreatedAsSuccessfulStart() th assertPipeMetaChangeWasWritten(consensusManager, pipeName, error); } + @Test + public void testRetryAfterFailureToPersistRuntimeStopDoesNotClearFailure() throws Exception { + final String pipeName = "startFailurePersistRetryPipe"; + final PipeTaskInfo taskInfo = new PipeTaskInfo(); + final PipeStaticMeta staticMeta = + new PipeStaticMeta( + pipeName, + System.currentTimeMillis(), + Collections.singletonMap("extractor", "iotdb-source"), + Collections.singletonMap("processor", "do-nothing-processor"), + Collections.singletonMap("connector", "write-back-sink")); + final PipeRuntimeMeta runtimeMeta = new PipeRuntimeMeta(new ConcurrentHashMap<>()); + runtimeMeta + .getConsensusGroupId2TaskMetaMap() + .put(1, new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1)); + taskInfo.createPipe(new CreatePipePlanV2(staticMeta, runtimeMeta)); + taskInfo.setPipeStatus(new SetPipeStatusPlanV2(pipeName, PipeStatus.RUNNING)); + + final StartPipeProcedureV2 procedure = new TestableStartPipeProcedureV2(pipeName, taskInfo); + final ConfigNodeProcedureEnv env = Mockito.mock(ConfigNodeProcedureEnv.class); + final ConfigManager configManager = Mockito.mock(ConfigManager.class); + final ConsensusManager consensusManager = Mockito.mock(ConsensusManager.class); + Mockito.when(env.getConfigManager()).thenReturn(configManager); + Mockito.when(configManager.getConsensusManager()).thenReturn(consensusManager); + Mockito.when(consensusManager.write(Mockito.any(ConfigPhysicalPlan.class))) + .thenReturn(new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode())) + .thenReturn(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())); + + final String error = "Failed to construct PipeConnector"; + final TPushPipeMetaRespExceptionMessage exceptionMessage = + new TPushPipeMetaRespExceptionMessage(pipeName, error, System.currentTimeMillis()) + .setCreationTime(staticMeta.getCreationTime()); + Mockito.when(env.pushSinglePipeMetaToDataNodes(Mockito.any(ByteBuffer.class), Mockito.any())) + .thenReturn( + Collections.singletonMap( + 1, + new TPushPipeMetaResp( + new TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode())) + .setExceptionMessages(Collections.singletonList(exceptionMessage)))); + + try { + procedure.executeFromOperateOnDataNodes(env); + fail("Expected the first attempt to fail its ConfigNode consensus write"); + } catch (final org.apache.iotdb.pipe.api.exception.PipeException expected) { + // The procedure retries this state after the persistence failure. + } + + final PipeRuntimeMeta runtimeMetaAfterFirstAttempt = + taskInfo.getPipeMetaByPipeName(pipeName).getRuntimeMeta(); + assertEquals(PipeStatus.STOPPED, runtimeMetaAfterFirstAttempt.getStatus().get()); + assertTrue(runtimeMetaAfterFirstAttempt.getIsStoppedByRuntimeException()); + + procedure.executeFromOperateOnDataNodes(env); + + final PipeRuntimeMeta runtimeMetaAfterRetry = + taskInfo.getPipeMetaByPipeName(pipeName).getRuntimeMeta(); + assertEquals(PipeStatus.STOPPED, runtimeMetaAfterRetry.getStatus().get()); + assertTrue(runtimeMetaAfterRetry.getIsStoppedByRuntimeException()); + assertEquals( + error, runtimeMetaAfterRetry.getNodeId2PipeRuntimeExceptionMap().get(1).getMessage()); + Mockito.verify(env, Mockito.times(1)) + .pushSinglePipeMetaToDataNodes(Mockito.any(ByteBuffer.class), Mockito.any()); + Mockito.verify(consensusManager, Mockito.times(2)).write(Mockito.any(ConfigPhysicalPlan.class)); + } + private static ConsensusManager mockSuccessfulConsensusWrite(final ConfigNodeProcedureEnv env) throws Exception { final ConfigManager configManager = Mockito.mock(ConfigManager.class);