[To dev/1.3] Fix write-back sink history failure state handling

integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAutoSplitIT.java

@@ -20,6 +20,7 @@
 package org.apache.iotdb.pipe.it.autocreate;
 
 import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus;
 import org.apache.iotdb.confignode.rpc.thrift.TShowPipeInfo;
 import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq;
 import org.apache.iotdb.consensus.ConsensusFactory;
@@ -40,6 +41,7 @@
 import java.sql.Statement;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Objects;
 
@@ -77,6 +79,59 @@ public void setUp() {
     receiverEnv.initClusterEnvironment();
   }
 
+  @Test
+  public void testWriteBackSinkWithTargetDatabase() throws Exception {
+    TestUtils.executeNonQueries(
+        senderEnv,
+        Arrays.asList(
+            "create database root.writeback_source",
+            "create aligned timeseries root.writeback_source.d1(s1 INT32)",
+            "insert into root.writeback_source.d1(time,s1) aligned values "
+                + "(1,1),(2,2),(3,3),(4,4),(5,5)",
+            "flush",
+            "create pipe writeback with source ("
+                + "'path'='root.writeback_source.**','inclusion'='data.insert',"
+                + "'forwarding-pipe-requests'='false',"
+                + "'history.enable'='true','realtime.enable'='true') "
+                + "with sink ('sink'='write-back-sink','sink.database'='root.writeback_target',"
+                + "'user'='root','password'='root')"),
+        null);
+
+    TestUtils.assertDataEventuallyOnEnv(
+        senderEnv,
+        "select s1 from root.writeback_target.d1",
+        "Time,root.writeback_target.d1.s1,",
+        new HashSet<>(Arrays.asList("1,1,", "2,2,", "3,3,", "4,4,", "5,5,")),
+        90);
+
+    try (final SyncConfigNodeIServiceClient client =
+        (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+      final List<TShowPipeInfo> pipeInfoList = client.showPipe(new TShowPipeReq()).pipeInfoList;
+      final TShowPipeInfo realtimePipe =
+          pipeInfoList.stream()
+              .filter(pipeInfo -> "writeback_realtime".equals(pipeInfo.id))
+              .findFirst()
+              .orElseThrow(AssertionError::new);
+      Assert.assertEquals(PipeStatus.RUNNING.name(), realtimePipe.getState());
+      Assert.assertEquals("", realtimePipe.getExceptionMessage());
+    }
+
+    TestUtils.executeNonQueries(
+        senderEnv,
+        Collections.singletonList(
+            "insert into root.writeback_source.d1(time,s1) aligned values "
+                + "(6,6),(7,7),(8,8),(9,9)"),
+        null);
+
+    TestUtils.assertDataEventuallyOnEnv(
+        senderEnv,
+        "select s1 from root.writeback_target.d1",
+        "Time,root.writeback_target.d1.s1,",
+        new HashSet<>(
+            Arrays.asList("1,1,", "2,2,", "3,3,", "4,4,", "5,5,", "6,6,", "7,7,", "8,8,", "9,9,")),
+        90);
+  }
+
   @Test
   public void testSingleEnv() throws Exception {
     final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);

iotdb-api/pipe-api/pom.xml

@@ -35,6 +35,11 @@
             <version>${tsfile.version}</version>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
     <profiles>
         <profile>

iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameterValidator.java

@@ -25,6 +25,7 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -57,9 +58,14 @@ public PipeParameterValidator validateSynonymAttributes(
       final List<String> lhsAttributes,
       final List<String> rhsAttributes,
       final boolean isRequired) {
-    final boolean lhsExistence = lhsAttributes.stream().anyMatch(parameters::hasAttribute);
-    final boolean rhsExistence = rhsAttributes.stream().anyMatch(parameters::hasAttribute);
-    if (lhsExistence && rhsExistence) {
+    final Set<String> lhsExistingKeys = getExistingAttributeKeys(lhsAttributes);
+    final Set<String> rhsExistingKeys = getExistingAttributeKeys(rhsAttributes);
+    final boolean lhsExistence = !lhsExistingKeys.isEmpty();
+    final boolean rhsExistence = !rhsExistingKeys.isEmpty();
+    // One unprefixed key may match both sides after prefix reduction. It is still one attribute.
+    if (lhsExistence
+        && rhsExistence
+        && !(lhsExistingKeys.size() == 1 && lhsExistingKeys.equals(rhsExistingKeys))) {
       throw new PipeParameterNotValidException(
           String.format(
               "Cannot specify both %s and %s at the same time", lhsAttributes, rhsAttributes));
@@ -74,6 +80,18 @@ public 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/main/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameters.java

@@ -355,7 +355,7 @@ public PipeParameters addOrReplaceEquivalentAttributesWithClone(final PipeParame
     return new PipeParameters(thisMap);
   }
 
-  private static class KeyReducer {
+  static class KeyReducer {
     private static final Set<String> FIRST_PREFIXES = new HashSet<>();
     private static final Set<String> SECOND_PREFIXES = new HashSet<>();
 

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/pom.xml

@@ -32,6 +32,7 @@
         <confignode.it.skip>${confignode.test.skip}</confignode.it.skip>
         <confignode.test.skip>false</confignode.test.skip>
         <confignode.ut.skip>${confignode.test.skip}</confignode.ut.skip>
+        <i18n.locale>en</i18n.locale>
     </properties>
     <dependencies>
         <dependency>
@@ -168,6 +169,37 @@
     </dependencies>
     <build>
         <plugins>
+            <plugin>
+                <groupId>com.diffplug.spotless</groupId>
+                <artifactId>spotless-maven-plugin</artifactId>
+                <configuration>
+                    <java>
+                        <includes>
+                            <include>src/main/java/**/*.java</include>
+                            <include>src/test/java/**/*.java</include>
+                            <include>src/main/i18n/**/*.java</include>
+                        </includes>
+                    </java>
+                </configuration>
+            </plugin>
+            <plugin>
+                <groupId>org.codehaus.mojo</groupId>
+                <artifactId>build-helper-maven-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>add-i18n-source</id>
+                        <phase>generate-sources</phase>
+                        <goals>
+                            <goal>add-source</goal>
+                        </goals>
+                        <configuration>
+                            <sources>
+                                <source>${project.basedir}/src/main/i18n/${i18n.locale}</source>
+                            </sources>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-assembly-plugin</artifactId>
@@ -241,6 +273,12 @@
         </plugins>
     </build>
     <profiles>
+        <profile>
+            <id>with-zh-locale</id>
+            <properties>
+                <i18n.locale>zh</i18n.locale>
+            </properties>
+        </profile>
         <profile>
             <id>skipConfigNodeTests</id>
             <activation>

iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java

@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.confignode.i18n;
+
+public final class ProcedureMessages {
+
+  private ProcedureMessages() {}
+
+  public static final String
+      EXCEPTION_INTERNAL_ERROR_WHILE_PROCESSING_PUSHPIPEMETA_ON_DATANODE_5983BD5F =
+          "Internal error while processing pushPipeMeta on DataNode.";
+  public static final String
+      LOG_FAILED_IN_THE_WRITE_API_EXECUTING_THE_CONSENSUS_LAYER_DUE_TO_8DDF6FE9 =
+          "Failed in the write API executing the consensus layer due to: ";
+}

iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java

@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.confignode.i18n;
+
+public final class ProcedureMessages {
+
+  private ProcedureMessages() {}
+
+  public static final String
+      EXCEPTION_INTERNAL_ERROR_WHILE_PROCESSING_PUSHPIPEMETA_ON_DATANODE_5983BD5F =
+          "处理 DataNode 上的 pushPipeMeta 时发生内部错误。";
+  public static final String
+      LOG_FAILED_IN_THE_WRITE_API_EXECUTING_THE_CONSENSUS_LAYER_DUE_TO_8DDF6FE9 =
+          "执行共识层写入 API 时失败，原因：";
+}