[To dev/1.3] [IT] Fix port occupation check that always reported ports as free (#15794, #19070)

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

@@ -24,9 +24,14 @@
 import org.apache.commons.lang3.SystemUtils;
 import org.apache.tsfile.utils.Pair;
 
+import java.io.BufferedReader;
+import java.io.EOFException;
 import java.io.File;
 import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
@@ -96,30 +101,74 @@ 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;
   }
 
-  private static String getSearchAvailablePortCmd(final List<Integer> ports) {
-    return SystemUtils.IS_OS_WINDOWS ? getWindowsSearchPortCmd(ports) : getUnixSearchPortCmd(ports);
+  public static Map<Integer, Long> listPortOccupation(final List<Integer> ports)
+      throws IOException {
+    return SystemUtils.IS_OS_WINDOWS
+        ? listPortOccupationWindows(ports)
+        : listPortOccupationUnix(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(" "));
+  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);
   }
 
-  private static String getUnixSearchPortCmd(final List<Integer> ports) {
-    return "lsof -iTCP -sTCP:LISTEN,TIME_WAIT -P -n | awk '{print $9}' | grep -E "
-        + ports.stream().map(String::valueOf).collect(Collectors.joining("|"))
-        + "\"";
+  /**
+   * List occupied port and the associated pid on Unix.
+   *
+   * <p>A line of the lsof output splits into 10 columns, e.g. {@code java 1234 user 77u IPv6 0x1a2b
+   * 0t0 TCP 127.0.0.1:6667 (LISTEN)}. The address is in column 8 and the state in column 9.
+   *
+   * @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,TIME_WAIT -P -n", 10, 8, 1);
   }
 
   private static Pair<Integer, Integer> getClusterNodesNum(final int index) {