Fix write-back sink history failure state handling

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','password'='root')"),
+        null);
+
+    TestUtils.assertDataEventuallyOnEnv(
+        senderEnv,
+        "select s1 from root.writeback_target.d1",
+        "Time,root.writeback_target.d1.s1,",
+        new HashSet<>(Arrays.asList("1,1,", "2,2,", "3,3,", "4,4,", "5,5,")),
+        90);
+
+    try (final SyncConfigNodeIServiceClient client =
+        (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+      final List<TShowPipeInfo> pipeInfoList =
+          client.showPipe(new TShowPipeReq().setUserName(SessionConfig.DEFAULT_USER)).pipeInfoList;
+      final TShowPipeInfo realtimePipe =
+          pipeInfoList.stream()
+              .filter(pipeInfo -> "writeback_realtime".equals(pipeInfo.id))
+              .findFirst()
+              .orElseThrow(AssertionError::new);
+      Assert.assertEquals(PipeStatus.RUNNING.name(), realtimePipe.getState());
+      Assert.assertEquals("", realtimePipe.getExceptionMessage());
+    }
+
+    TestUtils.executeNonQueries(
+        senderEnv,
+        Collections.singletonList(
+            "insert into root.writeback_source.d1(time,s1) aligned values "
+                + "(6,6),(7,7),(8,8),(9,9)"),
+        null);
+
+    TestUtils.assertDataEventuallyOnEnv(
+        senderEnv,
+        "select s1 from root.writeback_target.d1",
+        "Time,root.writeback_target.d1.s1,",
+        new HashSet<>(
+            Arrays.asList("1,1,", "2,2,", "3,3,", "4,4,", "5,5,", "6,6,", "7,7,", "8,8,", "9,9,")),
+        90);
+  }
+
   @Test
   public void testSingleEnv() throws Exception {
     final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);

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<String> lhsAttributes,
       final List<String> rhsAttributes,
       final boolean isRequired) {
-    final boolean lhsExistence = lhsAttributes.stream().anyMatch(parameters::hasAttribute);
-    final boolean rhsExistence = rhsAttributes.stream().anyMatch(parameters::hasAttribute);
-    if (lhsExistence && rhsExistence) {
+    final Set<String> lhsExistingKeys = getExistingAttributeKeys(lhsAttributes);
+    final Set<String> rhsExistingKeys = getExistingAttributeKeys(rhsAttributes);
+    final boolean lhsExistence = !lhsExistingKeys.isEmpty();
+    final boolean rhsExistence = !rhsExistingKeys.isEmpty();
+    // One unprefixed key may match both sides after prefix reduction. It is still one attribute.
+    if (lhsExistence
+        && rhsExistence
+        && !(lhsExistingKeys.size() == 1 && lhsExistingKeys.equals(rhsExistingKeys))) {
       throw new PipeParameterNotValidException(
           String.format(
               PipeApiMessages.EXCEPTION_CANNOT_SPECIFY_BOTH_ARG_AND_ARG_AT_THE_SAME_TIME_7DA8858B,
@@ -77,6 +83,18 @@ public PipeParameterValidator validateSynonymAttributes(
     return this;
   }
 
+  private Set<String> getExistingAttributeKeys(final List<String> attributeKeys) {
+    return attributeKeys.stream()
+        .flatMap(
+            key ->
+                Stream.of(
+                    key,
+                    PipeParameters.KeyReducer.shallowReduce(key),
+                    PipeParameters.KeyReducer.reduce(key)))
+        .filter(parameters.getAttribute()::containsKey)
+        .collect(Collectors.toSet());
+  }
+
   /**
    * Validates whether the attributes entered by the user contain an attribute whose key is
    * attributeKey.

iotdb-api/pipe-api/src/test/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameterValidatorTest.java

@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.pipe.api.customizer.parameter;
+
+import org.apache.iotdb.pipe.api.exception.PipeAttributeNotProvidedException;
+import org.apache.iotdb.pipe.api.exception.PipeParameterNotValidException;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+public class PipeParameterValidatorTest {
+
+  @Test
+  public void testDatabaseAliasesMatchOneAttribute() {
+    for (final String key : Arrays.asList("connector.database", "sink.database", "database")) {
+      final PipeParameterValidator validator =
+          new PipeParameterValidator(
+              new PipeParameters(Collections.singletonMap(key, "root.target")));
+      for (final boolean required : new boolean[] {false, true}) {
+        Assert.assertSame(
+            validator,
+            validator.validateSynonymAttributes(
+                Collections.singletonList("connector.database"),
+                Collections.singletonList("sink.database"),
+                required));
+      }
+    }
+  }
+
+  @Test
+  public void testDatabaseAliasesRejectDistinctAttributes() {
+    final String[] keys = {"connector.database", "sink.database", "database"};
+    for (int i = 0; i < keys.length; i++) {
+      for (int j = i + 1; j < keys.length; j++) {
+        final Map<String, String> attributes = new HashMap<>();
+        attributes.put(keys[i], "root.target");
+        attributes.put(keys[j], "root.target");
+        Assert.assertThrows(
+            PipeParameterNotValidException.class,
+            () ->
+                new PipeParameterValidator(new PipeParameters(attributes))
+                    .validateSynonymAttributes(
+                        Collections.singletonList("connector.database"),
+                        Collections.singletonList("sink.database"),
+                        false));
+      }
+    }
+  }
+
+  @Test
+  public void testDistinctSynonymsRemainMutuallyExclusive() {
+    for (final String userKey : Arrays.asList("connector.user", "sink.user", "user")) {
+      for (final String usernameKey :
+          Arrays.asList("connector.username", "sink.username", "username")) {
+        final Map<String, String> attributes = new HashMap<>();
+        attributes.put(userKey, "user");
+        attributes.put(usernameKey, "user");
+        Assert.assertThrows(
+            PipeParameterNotValidException.class,
+            () ->
+                new PipeParameterValidator(new PipeParameters(attributes))
+                    .validateSynonymAttributes(
+                        Arrays.asList("connector.user", "sink.user"),
+                        Arrays.asList("connector.username", "sink.username"),
+                        false));
+      }
+    }
+  }
+
+  @Test
+  public void testShallowPrefixesMatchOneAttribute() {
+    new PipeParameterValidator(
+            new PipeParameters(Collections.singletonMap("opcua.database", "root.target")))
+        .validateSynonymAttributes(
+            Collections.singletonList("connector.opcua.database"),
+            Collections.singletonList("sink.opcua.database"),
+            true);
+  }
+
+  @Test
+  public void testRequiredSynonymsSupportUnprefixedAttributes() {
+    for (final String key : Arrays.asList("output.series", "output-series")) {
+      new PipeParameterValidator(new PipeParameters(Collections.singletonMap(key, "root.db.s")))
+          .validateSynonymAttributes(
+              Collections.singletonList("processor.output.series"),
+              Collections.singletonList("processor.output-series"),
+              true);
+    }
+
+    final PipeParameterValidator validator =
+        new PipeParameterValidator(new PipeParameters(Collections.emptyMap()));
+    validator.validateSynonymAttributes(
+        Collections.singletonList("connector.database"),
+        Collections.singletonList("sink.database"),
+        false);
+    Assert.assertThrows(
+        PipeAttributeNotProvidedException.class,
+        () ->
+            validator.validateSynonymAttributes(
+                Collections.singletonList("connector.database"),
+                Collections.singletonList("sink.database"),
+                true));
+  }
+}

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 =

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 =