Address pipe compatibility review comments

Author: Caideyipi

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

@@ -216,6 +216,9 @@ public final class ConfigNodeMessages {
       "Failed to drop trigger [%s], this trigger has not been created";
   public static final String FAILED_TO_DROP_UDF_THIS_UDF_HAS_NOT_BEEN_CREATED =
       "Failed to drop UDF [%s], this UDF has not been created";
+  public static final String
+      FAILED_TO_ENRICH_PIPE_WITH_ROOT_USER_FOR_COMPATIBILITY_BECAUSE_ROOT_USER_DOES_NOT_EXIST =
+          "Failed to enrich pipe %s with root user for compatibility because root user %s does not exist.";
   public static final String FAILED_TO_FETCH_SCHEMAENGINE_BLACK_LIST_ON_DATANODE =
       "Failed to fetch schemaengine black list on DataNode {}, {}";
   public static final String FAILED_TO_GET_FIELD = "Failed to get field {}";

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

@@ -212,6 +212,9 @@ public final class ConfigNodeMessages {
       "Failed to drop trigger [%s], this trigger has not been created";
   public static final String FAILED_TO_DROP_UDF_THIS_UDF_HAS_NOT_BEEN_CREATED =
       "Failed to drop UDF [%s], this UDF has not been created";
+  public static final String
+      FAILED_TO_ENRICH_PIPE_WITH_ROOT_USER_FOR_COMPATIBILITY_BECAUSE_ROOT_USER_DOES_NOT_EXIST =
+          "Failed to enrich pipe %s with root user for compatibility because root user %s does not exist.";
   public static final String FAILED_TO_FETCH_SCHEMAENGINE_BLACK_LIST_ON_DATANODE =
       "Failed to fetch schemaengine black list on DataNode {}, {}";
   public static final String FAILED_TO_GET_FIELD = "Failed to get field {}";

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java

@@ -376,7 +376,7 @@ public ConfigManager() throws IOException {
     CQInfo cqInfo = new CQInfo();
     ExternalServiceInfo externalServiceInfo = new ExternalServiceInfo();
     this.permissionManager = createPermissionManager(authorInfo);
-    PipeInfo pipeInfo = new PipeInfo(this.permissionManager::login4Pipe);
+    PipeInfo pipeInfo = new PipeInfo(userName -> this.permissionManager.login4Pipe(userName, null));
     QuotaInfo quotaInfo = new QuotaInfo();
     TTLInfo ttlInfo = new TTLInfo();
     SubscriptionInfo subscriptionInfo = new SubscriptionInfo();

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeInfo.java

@@ -49,7 +49,7 @@
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
-import java.util.function.BiFunction;
+import java.util.function.Function;
 
 public class PipeInfo implements SnapshotProcessor {
 
@@ -62,10 +62,10 @@ public PipeInfo() throws IOException {
     this(null);
   }
 
-  public PipeInfo(final BiFunction<String, String, String> pipeUserPasswordProvider)
+  public PipeInfo(final Function<String, String> pipeUserCurrentPasswordProvider)
       throws IOException {
     pipePluginInfo = new PipePluginInfo();
-    pipeTaskInfo = new PipeTaskInfo(pipeUserPasswordProvider);
+    pipeTaskInfo = new PipeTaskInfo(pipeUserCurrentPasswordProvider);
   }
 
   public PipePluginInfo getPipePluginInfo() {

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java

@@ -80,7 +80,7 @@
 import java.util.Optional;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
-import java.util.function.BiFunction;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 
@@ -97,16 +97,17 @@ public class PipeTaskInfo implements SnapshotProcessor {
 
   // Pure in-memory object, not involved in snapshot serialization and deserialization.
   private final PipeTaskInfoVersion pipeTaskInfoVersion;
-  private final BiFunction<String, String, String> pipeUserPasswordProvider;
+  // Accepts a username and returns its current stored password for pipe authentication.
+  private final Function<String, String> pipeUserCurrentPasswordProvider;
 
   public PipeTaskInfo() {
     this(null);
   }
 
-  public PipeTaskInfo(final BiFunction<String, String, String> pipeUserPasswordProvider) {
+  public PipeTaskInfo(final Function<String, String> pipeUserCurrentPasswordProvider) {
     this.pipeMetaKeeper = new PipeMetaKeeper();
     this.pipeTaskInfoVersion = new PipeTaskInfoVersion();
-    this.pipeUserPasswordProvider = pipeUserPasswordProvider;
+    this.pipeUserCurrentPasswordProvider = pipeUserCurrentPasswordProvider;
   }
 
   /////////////////////////////// Lock ///////////////////////////////
@@ -1022,7 +1023,7 @@ public void enrichPipeMetasWithRootUserForCompatibility() {
   }
 
   private void enrichPipeMetaWithRootUserForCompatibility(final PipeStaticMeta pipeStaticMeta) {
-    if (pipeUserPasswordProvider == null) {
+    if (pipeUserCurrentPasswordProvider == null) {
       return;
     }
     final boolean shouldEnrichSource = pipeStaticMeta.mayNeedCompatibleRootUserForIoTDBSource();
@@ -1032,12 +1033,14 @@ private void enrichPipeMetaWithRootUserForCompatibility(final PipeStaticMeta pip
     }
 
     final String rootUserName = CommonDescriptor.getInstance().getConfig().getDefaultAdminName();
-    final String password = pipeUserPasswordProvider.apply(rootUserName, null);
+    final String password = pipeUserCurrentPasswordProvider.apply(rootUserName);
     if (Objects.isNull(password)) {
       throw new PipeException(
           String.format(
-              "Failed to enrich pipe %s with root user for compatibility because root user %s does not exist.",
-              pipeStaticMeta.getPipeName(), rootUserName));
+              ConfigNodeMessages
+                  .FAILED_TO_ENRICH_PIPE_WITH_ROOT_USER_FOR_COMPATIBILITY_BECAUSE_ROOT_USER_DOES_NOT_EXIST,
+              pipeStaticMeta.getPipeName(),
+              rootUserName));
     }
 
     if (shouldEnrichSource) {

iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfoAutoRestartTest.java

@@ -96,7 +96,7 @@ public void testRecordDataNodePushPipeMetaExceptionsKeepsUserStoppedPipeOutOfAut
   public void testEnrichOldUserPipeWithRootUserForCompatibility() {
     final String rootUserName = CommonDescriptor.getInstance().getConfig().getDefaultAdminName();
     final String rootPassword = "root-current-password";
-    pipeTaskInfo = new PipeTaskInfo((username, password) -> rootPassword);
+    pipeTaskInfo = new PipeTaskInfo(username -> rootPassword);
 
     createPipe("oldPipe", PipeStatus.STOPPED);
 
@@ -124,7 +124,7 @@ public void testEnrichOldUserPipeWithRootUserForCompatibility() {
 
   @Test
   public void testDoNotOverwritePipeWithUserForCompatibility() {
-    pipeTaskInfo = new PipeTaskInfo((username, password) -> "root-current-password");
+    pipeTaskInfo = new PipeTaskInfo(username -> "root-current-password");
 
     createPipeWithSourceAttributes(
         "newPipe",
@@ -149,7 +149,7 @@ public void testDoNotOverwritePipeWithUserForCompatibility() {
 
   @Test
   public void testDoNotEnrichSystemPipeForCompatibility() {
-    pipeTaskInfo = new PipeTaskInfo((username, password) -> "root-current-password");
+    pipeTaskInfo = new PipeTaskInfo(username -> "root-current-password");
 
     createPipeWithSourceAttributes(
         PipeStaticMeta.generateSubscriptionPipeName("topic", "group"),
@@ -173,7 +173,7 @@ public void testDoNotEnrichSystemPipeForCompatibility() {
   public void testEnrichOldWriteBackSinkWithRootUserForCompatibility() {
     final String rootUserName = CommonDescriptor.getInstance().getConfig().getDefaultAdminName();
     final String rootPassword = "root-current-password";
-    pipeTaskInfo = new PipeTaskInfo((username, password) -> rootPassword);
+    pipeTaskInfo = new PipeTaskInfo(username -> rootPassword);
 
     createPipeWithAttributes(
         "oldWriteBackPipe",
@@ -206,7 +206,7 @@ public void testEnrichOldWriteBackSinkWithRootUserForCompatibility() {
   @Test
   public void testEnrichLoadedPipeMetasWithRootUserForCompatibility() {
     final String rootPassword = "root-current-password";
-    pipeTaskInfo = new PipeTaskInfo((username, password) -> rootPassword);
+    pipeTaskInfo = new PipeTaskInfo(username -> rootPassword);
 
     createPipeWithSourceAttributes(
         "loadedPipe",