Pipe IT: Add tests for iotdb-legacy-pipe-sink & Add forced-log & file mode tests for real-time mode & Refactor ITs

integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java

@@ -411,6 +411,31 @@ public static boolean tryExecuteNonQueryWithRetry(BaseEnv env, String sql) {
     return false;
   }
 
+  // This method will not throw failure given that a failure is encountered.
+  // Instead, it return a flag to indicate the result of the execution.
+  public static boolean tryExecuteNonQueriesWithRetry(BaseEnv env, List<String> sqlList) {
+    for (int retryCountLeft = 10; retryCountLeft >= 0; retryCountLeft--) {
+      try (Connection connection = env.getConnection();
+          Statement statement = connection.createStatement()) {
+        for (String sql : sqlList) {
+          statement.execute(sql);
+        }
+        return true;
+      } catch (SQLException e) {
+        if (retryCountLeft > 0) {
+          try {
+            Thread.sleep(10000);
+          } catch (InterruptedException ignored) {
+          }
+        } else {
+          e.printStackTrace();
+          return false;
+        }
+      }
+    }
+    return false;
+  }
+
   public static void executeNonQueryOnSpecifiedDataNodeWithRetry(
       BaseEnv env, DataNodeWrapper wrapper, String sql) {
     for (int retryCountLeft = 10; retryCountLeft >= 0; retryCountLeft--) {
@@ -456,6 +481,30 @@ public static boolean tryExecuteNonQueryOnSpecifiedDataNodeWithRetry(
     return false;
   }
 
+  public static boolean tryExecuteNonQueriesOnSpecifiedDataNodeWithRetry(
+      BaseEnv env, DataNodeWrapper wrapper, List<String> sqlList) {
+    for (int retryCountLeft = 10; retryCountLeft >= 0; retryCountLeft--) {
+      try (Connection connection = env.getConnectionWithSpecifiedDataNode(wrapper);
+          Statement statement = connection.createStatement()) {
+        for (String sql : sqlList) {
+          statement.execute(sql);
+        }
+        return true;
+      } catch (SQLException e) {
+        if (retryCountLeft > 0) {
+          try {
+            Thread.sleep(10000);
+          } catch (InterruptedException ignored) {
+          }
+        } else {
+          e.printStackTrace();
+          return false;
+        }
+      }
+    }
+    return false;
+  }
+
   public static void executeQuery(String sql) {
     executeQuery(sql, "root", "root");
   }