[apache#1675][FOLLOWUP] fix(test): Fix various flaky tests (apache#1730)

### What changes were proposed in this pull request?

Fix various flaky tests:
1. After using `SimpleClusterManager`, release resources at once, which might cause tests to fail sometimes.
2. The compiler machine may not necessarily have a valid broadcast address (maybe in a pod created by K8S, and the broadcast address could be 0.0.0.0). Unit tests should running successfully on such machines. We need to support this scenario.
3. Refactor the `URI.create` logic in `SimpleClusterManagerTest` to support cross-platform operation, such as running tests on Windows.
4. Use `IntegrationTestBase#shutdownServers` to reduce duplicated codes.


### Why are the changes needed?

For apache#1675.

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

Unnecessary.

Author: rickyma

common/src/main/java/org/apache/uniffle/common/util/RssUtils.java

@@ -123,6 +123,10 @@ public static String getHostIp() throws Exception {
       }
       return ip;
     }
+    // Unit tests are executed on a single machine and do not interact with other machines.
+    // Therefore, unit tests should not require a valid broadcast address.
+    // When running UTs, we will still return the IP address whose broadcast address is invalid.
+    boolean isTestMode = Boolean.parseBoolean(System.getProperty("test.mode", "false"));
     Enumeration<NetworkInterface> nif = NetworkInterface.getNetworkInterfaces();
     String siteLocalAddress = null;
     while (nif.hasMoreElements()) {
@@ -133,7 +137,7 @@ public static String getHostIp() throws Exception {
       for (InterfaceAddress ifa : ni.getInterfaceAddresses()) {
         InetAddress ia = ifa.getAddress();
         InetAddress brd = ifa.getBroadcast();
-        if (brd == null || brd.isAnyLocalAddress()) {
+        if ((brd == null || brd.isAnyLocalAddress()) && !isTestMode) {
           LOGGER.info(
               "ip {} was filtered, because it don't have effective broadcast address",
               ia.getHostAddress());

coordinator/src/test/java/org/apache/uniffle/coordinator/SimpleClusterManagerTest.java

@@ -21,6 +21,7 @@
 import java.io.FileWriter;
 import java.io.PrintWriter;
 import java.net.URI;
+import java.nio.file.Paths;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -132,91 +133,93 @@ public void getLostServerListTest() throws Exception {
     CoordinatorConf coordinatorConf = new CoordinatorConf();
     // Shorten the heartbeat time
     coordinatorConf.setLong(CoordinatorConf.COORDINATOR_HEARTBEAT_TIMEOUT, 300L);
-    SimpleClusterManager clusterManager =
-        new SimpleClusterManager(coordinatorConf, new Configuration());
-    ServerNode sn1 = new ServerNode("sn1", "ip", 0, 100L, 50L, 20, 10, grpcTags);
-    ServerNode sn2 = new ServerNode("sn2", "ip", 0, 100L, 50L, 21, 10, grpcTags);
-    ServerNode sn3 = new ServerNode("sn3", "ip", 0, 100L, 50L, 20, 11, grpcTags);
-    clusterManager.add(sn1);
-    clusterManager.add(sn2);
-    clusterManager.add(sn3);
-    Set<String> expectedIds = Sets.newHashSet("sn1", "sn2", "sn3");
-    await()
-        .atMost(1, TimeUnit.SECONDS)
-        .until(
-            () -> {
-              Set<String> lostServerList =
-                  clusterManager.getLostServerList().stream()
-                      .map(ServerNode::getId)
-                      .collect(Collectors.toSet());
-              return CollectionUtils.isEqualCollection(lostServerList, expectedIds);
-            });
-    // re-register sn3
-    sn3 = new ServerNode("sn3", "ip", 0, 100L, 50L, 20, 11, grpcTags);
-    clusterManager.add(sn3);
-    Set<String> expectedIdsre = Sets.newHashSet("sn1", "sn2");
-    await()
-        .atMost(1, TimeUnit.SECONDS)
-        .until(
-            () -> {
-              // Retrieve listed ServerNode List
-              Set<String> lostServerListre =
-                  clusterManager.getLostServerList().stream()
-                      .map(ServerNode::getId)
-                      .collect(Collectors.toSet());
-              return CollectionUtils.isEqualCollection(lostServerListre, expectedIdsre);
-            });
+    try (SimpleClusterManager clusterManager =
+        new SimpleClusterManager(coordinatorConf, new Configuration())) {
+      ServerNode sn1 = new ServerNode("sn1", "ip", 0, 100L, 50L, 20, 10, grpcTags);
+      ServerNode sn2 = new ServerNode("sn2", "ip", 0, 100L, 50L, 21, 10, grpcTags);
+      ServerNode sn3 = new ServerNode("sn3", "ip", 0, 100L, 50L, 20, 11, grpcTags);
+      clusterManager.add(sn1);
+      clusterManager.add(sn2);
+      clusterManager.add(sn3);
+      Set<String> expectedIds = Sets.newHashSet("sn1", "sn2", "sn3");
+      await()
+          .atMost(1, TimeUnit.SECONDS)
+          .until(
+              () -> {
+                Set<String> lostServerList =
+                    clusterManager.getLostServerList().stream()
+                        .map(ServerNode::getId)
+                        .collect(Collectors.toSet());
+                return CollectionUtils.isEqualCollection(lostServerList, expectedIds);
+              });
+      // re-register sn3
+      sn3 = new ServerNode("sn3", "ip", 0, 100L, 50L, 20, 11, grpcTags);
+      clusterManager.add(sn3);
+      Set<String> expectedIdsre = Sets.newHashSet("sn1", "sn2");
+      await()
+          .atMost(1, TimeUnit.SECONDS)
+          .until(
+              () -> {
+                // Retrieve listed ServerNode List
+                Set<String> lostServerListre =
+                    clusterManager.getLostServerList().stream()
+                        .map(ServerNode::getId)
+                        .collect(Collectors.toSet());
+                return CollectionUtils.isEqualCollection(lostServerListre, expectedIdsre);
+              });
+    }
   }
 
   @Test
   public void getUnhealthyServerList() throws Exception {
     CoordinatorConf coordinatorConf = new CoordinatorConf();
     // Shorten the heartbeat time
     coordinatorConf.setLong(CoordinatorConf.COORDINATOR_HEARTBEAT_TIMEOUT, 300L);
-    SimpleClusterManager clusterManager =
-        new SimpleClusterManager(coordinatorConf, new Configuration());
-    ServerNode sn1 = new ServerNode("sn1", "ip", 0, 100L, 50L, 20, 10, grpcTags);
-    ServerNode sn2 = new ServerNode("sn2", "ip", 0, 100L, 50L, 21, 10, grpcTags);
-    ServerNode sn3 =
-        new ServerNode("sn3", "ip", 0, 100L, 50L, 20, 11, grpcTags, ServerStatus.UNHEALTHY);
-    ServerNode sn4 =
-        new ServerNode("sn4", "ip", 0, 100L, 50L, 20, 11, grpcTags, ServerStatus.UNHEALTHY);
-    clusterManager.add(sn1);
-    clusterManager.add(sn2);
-    clusterManager.add(sn3);
-    clusterManager.add(sn4);
-    // Analog timeout registration
-    Set<String> expectedIds = Sets.newHashSet("sn3", "sn4");
-    await()
-        .atMost(1, TimeUnit.SECONDS)
-        .until(
-            () -> {
-              Set<String> unhealthyServerList =
-                  clusterManager.getUnhealthyServerList().stream()
-                      .map(ServerNode::getId)
-                      .collect(Collectors.toSet());
-              return CollectionUtils.isEqualCollection(unhealthyServerList, expectedIds);
-            });
-    // Register unhealthy node sn3 again
-    sn3 = new ServerNode("sn3", "ip", 0, 100L, 50L, 20, 11, grpcTags, ServerStatus.UNHEALTHY);
-    clusterManager.add(sn3);
-    Set<String> expectedIdsre = Sets.newHashSet("sn3");
-    await()
-        .atMost(1, TimeUnit.SECONDS)
-        .until(
-            () -> {
-              Set<String> unhealthyServerListre =
-                  clusterManager.getUnhealthyServerList().stream()
-                      .map(ServerNode::getId)
-                      .collect(Collectors.toSet());
-              return CollectionUtils.isEqualCollection(unhealthyServerListre, expectedIdsre);
-            });
-    // At this point verify that sn4 is in the lost list
-    List<ServerNode> lostremoveunhealthy = clusterManager.getLostServerList();
-    Set<String> expectedIdlostremoveunhealthy = Sets.newHashSet("sn1", "sn2", "sn4");
-    assertEquals(
-        expectedIdlostremoveunhealthy,
-        lostremoveunhealthy.stream().map(ServerNode::getId).collect(Collectors.toSet()));
+    try (SimpleClusterManager clusterManager =
+        new SimpleClusterManager(coordinatorConf, new Configuration())) {
+      ServerNode sn1 = new ServerNode("sn1", "ip", 0, 100L, 50L, 20, 10, grpcTags);
+      ServerNode sn2 = new ServerNode("sn2", "ip", 0, 100L, 50L, 21, 10, grpcTags);
+      ServerNode sn3 =
+          new ServerNode("sn3", "ip", 0, 100L, 50L, 20, 11, grpcTags, ServerStatus.UNHEALTHY);
+      ServerNode sn4 =
+          new ServerNode("sn4", "ip", 0, 100L, 50L, 20, 11, grpcTags, ServerStatus.UNHEALTHY);
+      clusterManager.add(sn1);
+      clusterManager.add(sn2);
+      clusterManager.add(sn3);
+      clusterManager.add(sn4);
+      // Analog timeout registration
+      Set<String> expectedIds = Sets.newHashSet("sn3", "sn4");
+      await()
+          .atMost(1, TimeUnit.SECONDS)
+          .until(
+              () -> {
+                Set<String> unhealthyServerList =
+                    clusterManager.getUnhealthyServerList().stream()
+                        .map(ServerNode::getId)
+                        .collect(Collectors.toSet());
+                return CollectionUtils.isEqualCollection(unhealthyServerList, expectedIds);
+              });
+      // Register unhealthy node sn3 again
+      sn3 = new ServerNode("sn3", "ip", 0, 100L, 50L, 20, 11, grpcTags, ServerStatus.UNHEALTHY);
+      clusterManager.add(sn3);
+      Set<String> expectedIdsre = Sets.newHashSet("sn3");
+      await()
+          .atMost(1, TimeUnit.SECONDS)
+          .until(
+              () -> {
+                Set<String> unhealthyServerListre =
+                    clusterManager.getUnhealthyServerList().stream()
+                        .map(ServerNode::getId)
+                        .collect(Collectors.toSet());
+                return CollectionUtils.isEqualCollection(unhealthyServerListre, expectedIdsre);
+              });
+      // At this point verify that sn4 is in the lost list
+      List<ServerNode> lostremoveunhealthy = clusterManager.getLostServerList();
+      Set<String> expectedIdlostremoveunhealthy = Sets.newHashSet("sn1", "sn2", "sn4");
+      assertEquals(
+          expectedIdlostremoveunhealthy,
+          lostremoveunhealthy.stream().map(ServerNode::getId).collect(Collectors.toSet()));
+    }
   }
 
   @Test
@@ -423,11 +426,12 @@ public void testGetCorrectServerNodesWhenOneNodeRemoved() throws Exception {
   public void updateExcludeNodesTest() throws Exception {
     String excludeNodesFolder =
         (new File(ClassLoader.getSystemResource("empty").getFile())).getParent();
-    String excludeNodesPath = excludeNodesFolder + "/excludeNodes";
+    String excludeNodesPath = Paths.get(excludeNodesFolder, "excludeNodes").toString();
     CoordinatorConf ssc = new CoordinatorConf();
+    File excludeNodesFile = new File(excludeNodesPath);
+    URI excludeNodesUri = excludeNodesFile.toURI();
     ssc.setString(
-        CoordinatorConf.COORDINATOR_EXCLUDE_NODES_FILE_PATH,
-        URI.create(excludeNodesPath).toString());
+        CoordinatorConf.COORDINATOR_EXCLUDE_NODES_FILE_PATH, excludeNodesUri.toURL().toString());
     ssc.setLong(CoordinatorConf.COORDINATOR_EXCLUDE_NODES_CHECK_INTERVAL, 1000);
 
     try (SimpleClusterManager scm = new SimpleClusterManager(ssc, new Configuration())) {
@@ -492,11 +496,12 @@ public void updateExcludeNodesTest() throws Exception {
   public void excludeNodesNoDelayTest() throws Exception {
     String excludeNodesFolder =
         (new File(ClassLoader.getSystemResource("empty").getFile())).getParent();
-    String excludeNodesPath = excludeNodesFolder + "/excludeNodes";
+    String excludeNodesPath = Paths.get(excludeNodesFolder, "excludeNodes").toString();
     CoordinatorConf ssc = new CoordinatorConf();
+    File excludeNodesFile = new File(excludeNodesPath);
+    URI excludeNodesUri = excludeNodesFile.toURI();
     ssc.setString(
-        CoordinatorConf.COORDINATOR_EXCLUDE_NODES_FILE_PATH,
-        URI.create(excludeNodesPath).toString());
+        CoordinatorConf.COORDINATOR_EXCLUDE_NODES_FILE_PATH, excludeNodesUri.toURL().toString());
     ssc.setLong(CoordinatorConf.COORDINATOR_EXCLUDE_NODES_CHECK_INTERVAL, 5000);
 
     final Set<String> nodes = Sets.newHashSet("node1-1999", "node2-1999");