Merge branch 'master' into strong_password

# Conflicts:
#	integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java
#	integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java

Author: jt2594838

integration-test/src/main/java/org/apache/iotdb/it/env/cluster/EnvUtils.java

@@ -24,9 +24,10 @@
 import org.apache.commons.lang3.SystemUtils;
 import org.apache.tsfile.utils.Pair;
 
-import java.io.File;
-import java.io.IOException;
+import java.io.*;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
@@ -98,15 +99,71 @@ public static int[] searchAvailablePorts() {
   }
 
   private static boolean checkPortsAvailable(final List<Integer> ports) {
-    final String cmd = getSearchAvailablePortCmd(ports);
     try {
-      return Runtime.getRuntime().exec(cmd).waitFor() == 1;
-    } catch (final IOException ignore) {
-      // ignore
-    } catch (final InterruptedException e) {
-      Thread.currentThread().interrupt();
+      return listPortOccupation(ports).isEmpty();
+    } catch (IOException e) {
+      IoTDBTestLogger.logger.error("Cannot check available ports", e);
+      return false;
     }
-    return false;
+  }
+
+  public static Map<Integer, Long> listPortOccupation(final List<Integer> ports)
+      throws IOException {
+    return SystemUtils.IS_OS_WINDOWS
+        ? listPortOccupationWindows(ports)
+        : listPortOccupationUnix(ports);
+  }
+
+  public static Map<Integer, Long> listPortOccupation(
+      final List<Integer> ports,
+      String cmd,
+      int targetColumnLength,
+      int addressColumnIndex,
+      int pidColumnIndex)
+      throws IOException {
+    Process process = Runtime.getRuntime().exec(cmd);
+    Map<Integer, Long> result = new HashMap<>();
+    try (BufferedReader reader =
+        new BufferedReader(new InputStreamReader(process.getInputStream()))) {
+      String line;
+      while ((line = reader.readLine()) != null) {
+        String[] split = line.trim().split("\\s+");
+        if (split.length != targetColumnLength) {
+          continue;
+        }
+        String localAddress = split[addressColumnIndex];
+        for (Integer port : ports) {
+          if (localAddress.endsWith(":" + port)) {
+            result.put(port, Long.parseLong(split[pidColumnIndex]));
+            break;
+          }
+        }
+      }
+    } catch (EOFException ignored) {
+    }
+    return result;
+  }
+
+  /**
+   * List occupied port and the associated pid on windows.
+   *
+   * @param ports ports to be checked
+   * @return (occupiedPort, pid) pairs
+   */
+  public static Map<Integer, Long> listPortOccupationWindows(final List<Integer> ports)
+      throws IOException {
+    return listPortOccupation(ports, "netstat -aon -p tcp", 5, 1, 4);
+  }
+
+  /**
+   * List occupied port and the associated pid on Unix.
+   *
+   * @param ports ports to be checked
+   * @return (occupiedPort, pid) pairs
+   */
+  public static Map<Integer, Long> listPortOccupationUnix(final List<Integer> ports)
+      throws IOException {
+    return listPortOccupation(ports, "lsof -iTCP -sTCP:LISTEN -P -n", 10, 9, 1);
   }
 
   private static String getSearchAvailablePortCmd(final List<Integer> ports) {
@@ -115,7 +172,7 @@ private static String getSearchAvailablePortCmd(final List<Integer> ports) {
 
   private static String getWindowsSearchPortCmd(final List<Integer> ports) {
     return "netstat -aon -p tcp | findStr "
-        + ports.stream().map(v -> "/C:'127.0.0.1:" + v + "'").collect(Collectors.joining(" "));
+        + ports.stream().map(v -> "/C:\"127.0.0.1:" + v + "\"").collect(Collectors.joining(" "));
   }
 
   private static String getUnixSearchPortCmd(final List<Integer> ports) {

integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java

@@ -526,6 +526,12 @@ public CommonConfig setQueryMemoryProportion(String queryMemoryProportion) {
     return this;
   }
 
+  @Override
+  public CommonConfig setSubscriptionEnabled(boolean subscriptionEnabled) {
+    setProperty("subscription_enabled", String.valueOf(subscriptionEnabled));
+    return this;
+  }
+
   @Override
   public CommonConfig setDefaultStorageGroupLevel(int defaultStorageGroupLevel) {
     setProperty("default_storage_group_level", String.valueOf(defaultStorageGroupLevel));

integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java

@@ -541,6 +541,13 @@ public CommonConfig setQueryMemoryProportion(String queryMemoryProportion) {
     return this;
   }
 
+  @Override
+  public CommonConfig setSubscriptionEnabled(boolean subscriptionEnabled) {
+    dnConfig.setSubscriptionEnabled(subscriptionEnabled);
+    cnConfig.setSubscriptionEnabled(subscriptionEnabled);
+    return this;
+  }
+
   @Override
   public CommonConfig setDefaultStorageGroupLevel(int defaultStorageGroupLevel) {
     dnConfig.setDefaultStorageGroupLevel(defaultStorageGroupLevel);