Made createOrUpdateDevice retriable

integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/TableModelUtils.java

@@ -21,6 +21,7 @@
 
 import org.apache.iotdb.db.it.utils.TestUtils;
 import org.apache.iotdb.isession.ITableSession;
+import org.apache.iotdb.isession.SessionConfig;
 import org.apache.iotdb.isession.pool.ITableSessionPool;
 import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
 import org.apache.iotdb.itbase.env.BaseEnv;
@@ -98,11 +99,21 @@ public static void createDatabase(final BaseEnv baseEnv, final String database,
     }
   }
 
-  public static void insertData(
+  public static boolean insertData(
+      final String dataBaseName,
+      final String tableName,
+      final int startInclusive,
+      final int endExclusive,
+      final BaseEnv baseEnv) {
+    return insertData(dataBaseName, tableName, startInclusive, endExclusive, false, baseEnv);
+  }
+
+  public static boolean insertData(
       final String dataBaseName,
       final String tableName,
       final int startInclusive,
       final int endExclusive,
+      final boolean trial,
       final BaseEnv baseEnv) {
     List<String> list = new ArrayList<>(endExclusive - startInclusive + 1);
     for (int i = startInclusive; i < endExclusive; ++i) {
@@ -112,7 +123,19 @@ public static void insertData(
               tableName, i, i, i, i, i, i, i, i, i, i, getDateStr(i), i, i));
     }
     list.add("flush");
-    TestUtils.executeNonQueries(dataBaseName, BaseEnv.TABLE_SQL_DIALECT, baseEnv, list, null);
+    if (!trial) {
+      TestUtils.executeNonQueries(dataBaseName, BaseEnv.TABLE_SQL_DIALECT, baseEnv, list, null);
+      return true;
+    } else {
+      return TestUtils.tryExecuteNonQueriesWithRetry(
+          baseEnv,
+          list,
+          SessionConfig.DEFAULT_USER,
+          SessionConfig.DEFAULT_PASSWORD,
+          dataBaseName,
+          BaseEnv.TABLE_SQL_DIALECT,
+          null);
+    }
   }
 
   public static void insertData(

integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeClusterIT.java

@@ -335,9 +335,12 @@ public void testPipeAfterDataRegionLeaderStop() throws Exception {
             fail();
           }
 
-          TableModelUtils.insertData("test", "test", 200, 300, senderEnv);
-
-          TableModelUtils.insertData("test1", "test1", 200, 300, senderEnv);
+          // The without unknown check may randomly fail, and may cause insertion failure
+          // Directly return here because the problem has nothing to do with pipe
+          if (!TableModelUtils.insertData("test", "test", 200, 300, true, senderEnv)
+              || !TableModelUtils.insertData("test1", "test1", 200, 300, true, senderEnv)) {
+            return;
+          }
 
           TableModelUtils.assertData("test", "test", 0, 300, receiverEnv, handleFailure);
         }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableDeviceSchemaValidator.java

@@ -19,7 +19,7 @@
 
 package org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher;
 
-import org.apache.iotdb.commons.exception.IoTDBException;
+import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
 import org.apache.iotdb.db.conf.IoTDBConfig;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
 import org.apache.iotdb.db.protocol.session.SessionManager;
@@ -247,9 +247,8 @@ private void autoCreateOrUpdateDeviceSchema(
             Long.MAX_VALUE,
             false);
     if (executionResult.status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
-      throw new RuntimeException(
-          new IoTDBException(
-              executionResult.status.getMessage(), executionResult.status.getCode()));
+      throw new IoTDBRuntimeException(
+          executionResult.status.getMessage(), executionResult.status.getCode());
     }
   }