[#2388] improvement(test): mr, tez, spark integration test use random port (#2397)

### What changes were proposed in this pull request?
- Integration test of mr, tez, spark use random port
- Re-enable `AccessClusterTest`

### Why are the changes needed?
Fix: #2388

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

### How was this patch tested?
UT

Author: summaryzb

common/src/test/java/org/apache/uniffle/common/port/PortRegistry.java

@@ -29,15 +29,12 @@
 
 import com.google.common.collect.Sets;
 import com.google.common.util.concurrent.Uninterruptibles;
-import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 
-import org.apache.uniffle.client.api.CoordinatorClient;
-import org.apache.uniffle.client.factory.CoordinatorClientFactory;
 import org.apache.uniffle.client.request.RssAccessClusterRequest;
 import org.apache.uniffle.client.response.RssAccessClusterResponse;
-import org.apache.uniffle.common.ClientType;
 import org.apache.uniffle.common.rpc.ServerType;
 import org.apache.uniffle.common.rpc.StatusCode;
 import org.apache.uniffle.common.util.Constants;
@@ -46,13 +43,13 @@
 import org.apache.uniffle.coordinator.access.AccessCheckResult;
 import org.apache.uniffle.coordinator.access.AccessInfo;
 import org.apache.uniffle.coordinator.access.checker.AccessChecker;
+import org.apache.uniffle.coordinator.metric.CoordinatorMetrics;
 import org.apache.uniffle.server.ShuffleServer;
 import org.apache.uniffle.server.ShuffleServerConf;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
-@Disabled("flaky test")
 public class AccessClusterTest extends CoordinatorTestBase {
 
   public static class MockedAccessChecker implements AccessChecker {
@@ -83,20 +80,27 @@ public void close() throws IOException {
     }
   }
 
+  @AfterEach
+  public void afterEach() throws Exception {
+    shutdownServers();
+    CoordinatorMetrics.clear();
+  }
+
   @Test
   public void testUsingCustomExtraProperties() throws Exception {
-    CoordinatorConf coordinatorConf = getCoordinatorConf();
+    CoordinatorConf coordinatorConf = coordinatorConfWithoutPort();
     coordinatorConf.setString(
         "rss.coordinator.access.checkers",
         "org.apache.uniffle.test.AccessClusterTest$MockedAccessChecker");
-    createCoordinatorServer(coordinatorConf);
-    startServers();
+    storeCoordinatorConf(coordinatorConf);
+    startServersWithRandomPorts();
     Uninterruptibles.sleepUninterruptibly(3, TimeUnit.SECONDS);
     // case1: empty map
     String accessID = "acessid";
     RssAccessClusterRequest request =
         new RssAccessClusterRequest(
             accessID, Sets.newHashSet(Constants.SHUFFLE_SERVER_VERSION), 2000, "user");
+    createClient();
     RssAccessClusterResponse response = coordinatorClient.accessCluster(request);
     assertEquals(StatusCode.ACCESS_DENIED, response.getStatusCode());
 
@@ -125,8 +129,6 @@ public void testUsingCustomExtraProperties() throws Exception {
             "user");
     response = coordinatorClient.accessCluster(request);
     assertEquals(StatusCode.SUCCESS, response.getStatusCode());
-
-    shutdownServers();
   }
 
   @Test
@@ -140,23 +142,23 @@ public void test(@TempDir File tempDir) throws Exception {
     printWriter.flush();
     printWriter.close();
 
-    CoordinatorConf coordinatorConf = getCoordinatorConf();
+    CoordinatorConf coordinatorConf = coordinatorConfWithoutPort();
     coordinatorConf.setInteger("rss.coordinator.access.loadChecker.serverNum.threshold", 2);
     coordinatorConf.setString("rss.coordinator.access.candidates.path", cfgFile.getAbsolutePath());
     coordinatorConf.setString(
         "rss.coordinator.access.checkers",
         "org.apache.uniffle.coordinator.access.checker.AccessCandidatesChecker,"
             + "org.apache.uniffle.coordinator.access.checker.AccessClusterLoadChecker");
-    createCoordinatorServer(coordinatorConf);
+    storeCoordinatorConf(coordinatorConf);
 
-    ShuffleServerConf shuffleServerConf = getShuffleServerConf(ServerType.GRPC);
-    createShuffleServer(shuffleServerConf);
-    startServers();
+    storeShuffleServerConf(shuffleServerConfWithoutPort(0, tempDir, ServerType.GRPC));
+    startServersWithRandomPorts();
     Uninterruptibles.sleepUninterruptibly(3, TimeUnit.SECONDS);
     String accessId = "111111";
     RssAccessClusterRequest request =
         new RssAccessClusterRequest(
             accessId, Sets.newHashSet(Constants.SHUFFLE_SERVER_VERSION), 2000, "user");
+    createClient();
     RssAccessClusterResponse response = coordinatorClient.accessCluster(request);
     assertEquals(StatusCode.ACCESS_DENIED, response.getStatusCode());
     assertTrue(response.getMessage().startsWith("Denied by AccessCandidatesChecker"));
@@ -168,32 +170,19 @@ public void test(@TempDir File tempDir) throws Exception {
     response = coordinatorClient.accessCluster(request);
     assertEquals(StatusCode.ACCESS_DENIED, response.getStatusCode());
     assertTrue(response.getMessage().startsWith("Denied by AccessClusterLoadChecker"));
-
-    shuffleServerConf.setInteger(
-        "rss.rpc.server.port", shuffleServerConf.getInteger(ShuffleServerConf.RPC_SERVER_PORT) + 2);
-    shuffleServerConf.setInteger(
-        "rss.jetty.http.port", shuffleServerConf.getInteger(ShuffleServerConf.JETTY_HTTP_PORT) + 1);
+    ShuffleServerConf shuffleServerConf = shuffleServerConfWithoutPort(0, tempDir, ServerType.GRPC);
+    shuffleServerConf.setString("rss.coordinator.quorum", getQuorum());
     ShuffleServer shuffleServer = new ShuffleServer(shuffleServerConf);
     shuffleServer.start();
+    // this make sure the server can be shutdown
+    grpcShuffleServers.add(shuffleServer);
     Uninterruptibles.sleepUninterruptibly(3, TimeUnit.SECONDS);
 
-    CoordinatorClient client =
-        CoordinatorClientFactory.getInstance()
-            .createCoordinatorClient(ClientType.GRPC, LOCALHOST, COORDINATOR_PORT_1 + 13);
-    request =
-        new RssAccessClusterRequest(
-            accessId, Sets.newHashSet(Constants.SHUFFLE_SERVER_VERSION), 2000, "user");
-    response = client.accessCluster(request);
-    assertEquals(StatusCode.INTERNAL_ERROR, response.getStatusCode());
-    assertTrue(response.getMessage().startsWith("UNAVAILABLE: io exception"));
-
     request =
         new RssAccessClusterRequest(
             accessId, Sets.newHashSet(Constants.SHUFFLE_SERVER_VERSION), 2000, "user");
     response = coordinatorClient.accessCluster(request);
     assertEquals(StatusCode.SUCCESS, response.getStatusCode());
     assertTrue(response.getMessage().startsWith("SUCCESS"));
-    shuffleServer.stopServer();
-    shutdownServers();
   }
 }

integration-test/common/src/test/java/org/apache/uniffle/test/AccessClusterTest.java

@@ -52,7 +52,7 @@ public static void setUp() throws Exception {
 
   @BeforeEach
   public void createClient() {
-    String hostUrl = String.format("http://%s:%d", LOCALHOST, jettyPorts.get(0));
+    String hostUrl = String.format("http://%s:%d", LOCALHOST, coordinators.get(0).getJettyPort());
     adminRestApi = new AdminRestApi(UniffleRestClient.builder(hostUrl).build());
   }
 

integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorAdminServiceTest.java

@@ -143,7 +143,6 @@ public void getShuffleAssignmentsTest(@TempDir File tmpDir) throws Exception {
     CoordinatorTestUtils.waitForRegister(coordinatorClient, 2);
 
     grpcShuffleServers.get(0).stopServer();
-    List<Integer> ports = reserveJettyPorts(1);
     ShuffleServerConf shuffleServerConf = shuffleServerConfWithoutPort(0, tmpDir, ServerType.GRPC);
     shuffleServerConf.set(ShuffleServerConf.STORAGE_MEDIA_PROVIDER_ENV_KEY, "RSS_ENV_KEY");
     String baseDir = shuffleServerConf.get(ShuffleServerConf.RSS_STORAGE_BASE_PATH).get(0);
@@ -154,7 +153,7 @@ public void getShuffleAssignmentsTest(@TempDir File tmpDir) throws Exception {
             () -> {
               shuffleServerConf.setString("rss.coordinator.quorum", getQuorum());
               shuffleServerConf.setInteger(RssBaseConf.RPC_SERVER_PORT, 0);
-              shuffleServerConf.setInteger(RssBaseConf.JETTY_HTTP_PORT, ports.get(0));
+              shuffleServerConf.setInteger(RssBaseConf.JETTY_HTTP_PORT, 0);
               ShuffleServer ss = new ShuffleServer(shuffleServerConf);
               ss.start();
               grpcShuffleServers.set(0, ss);
@@ -298,7 +297,6 @@ public void shuffleServerHeartbeatTest(@TempDir File tempDir) throws Exception {
     assertTrue(node.getTags().contains(Constants.SHUFFLE_SERVER_VERSION));
     assertTrue(scm.getTagToNodes().get(Constants.SHUFFLE_SERVER_VERSION).contains(node));
 
-    List<Integer> ports = reserveJettyPorts(1);
     ShuffleServerConf shuffleServerConf = shuffleServerConfWithoutPort(0, tempDir, ServerType.GRPC);
     shuffleServerConf.set(ShuffleServerConf.STORAGE_MEDIA_PROVIDER_ENV_KEY, "RSS_ENV_KEY");
     String baseDir = shuffleServerConf.get(ShuffleServerConf.RSS_STORAGE_BASE_PATH).get(0);
@@ -311,7 +309,7 @@ public void shuffleServerHeartbeatTest(@TempDir File tempDir) throws Exception {
               shuffleServerConf.set(ShuffleServerConf.TAGS, Lists.newArrayList("SSD"));
               shuffleServerConf.setString("rss.coordinator.quorum", getQuorum());
               shuffleServerConf.setInteger(RssBaseConf.RPC_SERVER_PORT, 0);
-              shuffleServerConf.setInteger(RssBaseConf.JETTY_HTTP_PORT, ports.get(0));
+              shuffleServerConf.setInteger(RssBaseConf.JETTY_HTTP_PORT, 0);
               ShuffleServer ss = new ShuffleServer(shuffleServerConf);
               ss.start();
               grpcShuffleServers.set(0, ss);