Fix compatibility issues when loading snapshot in pipe. (#16580)

Author: wenyanshi-123

integration-test/src/test/java/org/apache/iotdb/ainode/it/AINodeConcurrentInferenceIT.java

@@ -65,7 +65,7 @@ public static void tearDown() throws Exception {
 
   private static void prepareDataForTreeModel() throws SQLException {
     try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TREE_SQL_DIALECT);
-         Statement statement = connection.createStatement()) {
+        Statement statement = connection.createStatement()) {
       statement.execute("CREATE DATABASE root.AI");
       statement.execute("CREATE TIMESERIES root.AI.s WITH DATATYPE=DOUBLE, ENCODING=RLE");
       for (int i = 0; i < 2880; i++) {
@@ -79,7 +79,7 @@ private static void prepareDataForTreeModel() throws SQLException {
 
   private static void prepareDataForTableModel() throws SQLException {
     try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
-         Statement statement = connection.createStatement()) {
+        Statement statement = connection.createStatement()) {
       statement.execute("CREATE DATABASE root");
       statement.execute("CREATE TABLE root.AI (s DOUBLE FIELD)");
       for (int i = 0; i < 2880; i++) {
@@ -99,7 +99,7 @@ public void concurrentGPUCallInferenceTest() throws SQLException, InterruptedExc
   private void concurrentGPUCallInferenceTest(String modelId)
       throws SQLException, InterruptedException {
     try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TREE_SQL_DIALECT);
-         Statement statement = connection.createStatement()) {
+        Statement statement = connection.createStatement()) {
       final int threadCnt = 10;
       final int loop = 100;
       final int predictLength = 512;
@@ -134,7 +134,7 @@ public void concurrentGPUForecastTest() throws SQLException, InterruptedExceptio
   public void concurrentGPUForecastTest(String modelId, String selectSql)
       throws SQLException, InterruptedException {
     try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
-         Statement statement = connection.createStatement()) {
+        Statement statement = connection.createStatement()) {
       final int threadCnt = 10;
       final int loop = 100;
       final int predictLength = 512;
@@ -164,7 +164,7 @@ private void checkModelOnSpecifiedDevice(Statement statement, String modelId, St
     for (int retry = 0; retry < 200; retry++) {
       Set<String> foundDevices = new HashSet<>();
       try (final ResultSet resultSet =
-               statement.executeQuery(String.format("SHOW LOADED MODELS '%s'", device))) {
+          statement.executeQuery(String.format("SHOW LOADED MODELS '%s'", device))) {
         while (resultSet.next()) {
           String deviceId = resultSet.getString("DeviceId");
           String loadedModelId = resultSet.getString("ModelId");

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/source/ConfigRegionListeningQueue.java

@@ -157,15 +157,21 @@ public synchronized void tryListenToSnapshots(
                   : null,
               snapshotPathInfo.getRight());
       if (type == CNSnapshotFileType.USER_ROLE) {
-        long userId = Long.parseLong(snapshotPath.toFile().getName().split("_")[0]);
-        try {
-          curEvent.setAuthUserName(
-              ConfigNode.getInstance()
-                  .getConfigManager()
-                  .getPermissionManager()
-                  .getUserName(userId));
-        } catch (AuthException e) {
-          LOGGER.warn("Failed to collect user name for user id {}", userId, e);
+        String userName = snapshotPath.toFile().getName().split("_")[0];
+        long userId;
+        if (userName.matches("\\d+")) {
+          userId = Long.parseLong(userName);
+          try {
+            curEvent.setAuthUserName(
+                ConfigNode.getInstance()
+                    .getConfigManager()
+                    .getPermissionManager()
+                    .getUserName(userId));
+          } catch (AuthException e) {
+            LOGGER.warn("Failed to collect user name for user id {}", userId, e);
+          }
+        } else {
+          curEvent.setAuthUserName(userName);
         }
       }
       events.add(curEvent);

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/udf/UDTFForecast.java

@@ -125,11 +125,8 @@ public void beforeStart(UDFParameters parameters, UDTFConfigurations configurati
         Arrays.stream(
                 parameters.getStringOrDefault(OPTIONS_PARAMETER_NAME, DEFAULT_OPTIONS).split(","))
             .map(s -> s.split("="))
-            .filter(arr -> arr.length == 2 && !arr[0].isEmpty()) // 防御性检查
-            .collect(
-                Collectors.toMap(
-                    arr -> arr[0].trim(), arr -> arr[1].trim(), (v1, v2) -> v2 // 如果 key 重复，保留后一个
-                ));
+            .filter(arr -> arr.length == 2 && !arr[0].isEmpty())
+            .collect(Collectors.toMap(arr -> arr[0].trim(), arr -> arr[1].trim(), (v1, v2) -> v2));
     this.inputRows = new LinkedList<>();
     List<TSDataType> tsDataTypeList = new ArrayList<>(this.types.size() - 1);
     for (int i = 0; i < this.types.size(); i++) {

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/auth/user/BasicUserManager.java

@@ -41,7 +41,6 @@
 
 import static org.apache.iotdb.commons.auth.entity.User.INTERNAL_USER_END_ID;
 
-
 /** This class stores information of each user. */
 public abstract class BasicUserManager extends BasicRoleManager {