introduce a more elegant way to use jetty port

Author: summaryzb

common/src/main/java/org/apache/uniffle/common/web/JettyServer.java

@@ -169,13 +169,15 @@ public Server getServer() {
     return this.server;
   }
 
-  public void start() throws Exception {
+  public int start() throws Exception {
     try {
       server.start();
+      httpPort = ((ServerConnector) server.getConnectors()[0]).getLocalPort();
     } catch (BindException e) {
       ExitUtils.terminate(1, "Fail to start jetty http server, port is " + httpPort, e, LOG);
     }
     LOG.info("Jetty http server started, listening on port {}", httpPort);
+    return httpPort;
   }
 
   public void stop() throws Exception {

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

@@ -17,43 +17,26 @@
 
 package org.apache.uniffle.common.web;
 
-import java.io.FileNotFoundException;
-
 import org.eclipse.jetty.server.Handler;
 import org.eclipse.jetty.server.Server;
 import org.eclipse.jetty.server.ServerConnector;
 import org.eclipse.jetty.servlet.ServletContextHandler;
 import org.eclipse.jetty.util.thread.ExecutorThreadPool;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 import org.apache.uniffle.common.config.RssBaseConf;
-import org.apache.uniffle.common.port.PortRegistry;
 import org.apache.uniffle.common.util.ExitUtils;
 import org.apache.uniffle.common.util.ExitUtils.ExitException;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 public class JettyServerTest {
-
-  int port;
-
-  @BeforeEach
-  public void beforeEach() {
-    port = PortRegistry.reservePort();
-  }
-
-  @AfterEach
-  public void afterEach() {
-    PortRegistry.release(port);
-  }
-
   @Test
-  public void jettyServerTest() throws FileNotFoundException {
+  public void jettyServerTest() throws Exception {
     RssBaseConf conf = new RssBaseConf();
-    conf.setInteger("rss.jetty.http.port", port);
+    conf.setInteger("rss.jetty.http.port", 0);
     JettyServer jettyServer = new JettyServer(conf);
     Server server = jettyServer.getServer();
 
@@ -65,21 +48,26 @@ public void jettyServerTest() throws FileNotFoundException {
     assertEquals(server, server.getHandler().getServer());
     assertTrue(server.getConnectors()[0] instanceof ServerConnector);
     ServerConnector connector = (ServerConnector) server.getConnectors()[0];
-    assertEquals(port, connector.getPort());
+    assertEquals(0, connector.getPort());
+    jettyServer.start();
+    assertEquals(jettyServer.getHttpPort(), connector.getLocalPort());
+    assertNotEquals(jettyServer.getHttpPort(), 0);
 
     assertEquals(1, server.getHandlers().length);
     Handler handler = server.getHandler();
     assertTrue(handler instanceof ServletContextHandler);
+    jettyServer.stop();
   }
 
   @Test
   public void jettyServerStartTest() throws Exception {
     RssBaseConf conf = new RssBaseConf();
-    conf.setInteger("rss.jetty.http.port", port);
+    conf.setInteger("rss.jetty.http.port", 0);
     JettyServer jettyServer1 = new JettyServer(conf);
-    JettyServer jettyServer2 = new JettyServer(conf);
-    jettyServer1.start();
 
+    int portExist = jettyServer1.start();
+    conf.setInteger("rss.jetty.http.port", portExist);
+    JettyServer jettyServer2 = new JettyServer(conf);
     ExitUtils.disableSystemExit();
     final String expectMessage = "Fail to start jetty http server";
     final int expectStatus = 1;

common/src/test/java/org/apache/uniffle/common/web/JettyServerTest.java

@@ -63,6 +63,7 @@ public class CoordinatorServer {
   private final CoordinatorConf coordinatorConf;
   private final long startTimeMs;
   private JettyServer jettyServer;
+  private int jettyPort;
   private ServerInterface server;
   private ClusterManager clusterManager;
   private AssignmentStrategy assignmentStrategy;
@@ -106,7 +107,7 @@ public static void main(String[] args) throws Exception {
   public void start() throws Exception {
     LOG.info(
         "{} version: {}", this.getClass().getSimpleName(), Constants.VERSION_AND_REVISION_SHORT);
-    jettyServer.start();
+    jettyPort = jettyServer.start();
     rpcListenPort = server.start();
     if (metricReporter != null) {
       metricReporter.start();
@@ -285,4 +286,8 @@ public long getStartTimeMs() {
   public int getRpcListenPort() {
     return rpcListenPort;
   }
+
+  public int getJettyPort() {
+    return jettyPort;
+  }
 }

coordinator/src/main/java/org/apache/uniffle/coordinator/CoordinatorServer.java

@@ -35,7 +35,6 @@
 
 import org.apache.uniffle.client.request.RssAccessClusterRequest;
 import org.apache.uniffle.client.response.RssAccessClusterResponse;
-import org.apache.uniffle.common.config.RssBaseConf;
 import org.apache.uniffle.common.rpc.ServerType;
 import org.apache.uniffle.common.rpc.StatusCode;
 import org.apache.uniffle.common.util.Constants;
@@ -171,12 +170,8 @@ 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"));
-
-    List<Integer> ports = reserveJettyPorts(1);
     ShuffleServerConf shuffleServerConf = shuffleServerConfWithoutPort(0, tempDir, ServerType.GRPC);
     shuffleServerConf.setString("rss.coordinator.quorum", getQuorum());
-    shuffleServerConf.setInteger(RssBaseConf.RPC_SERVER_PORT, 0);
-    shuffleServerConf.setInteger(RssBaseConf.JETTY_HTTP_PORT, ports.get(0));
     ShuffleServer shuffleServer = new ShuffleServer(shuffleServerConf);
     shuffleServer.start();
     // this make sure the server can be shutdown

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);