[Java] Generate head redis port randomly (#6879)

* Random head port

* address comments.

java/runtime/src/main/java/org/ray/runtime/config/RayConfig.java

@@ -11,6 +11,8 @@ import java.io.File;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.Random;
+
 import org.ray.api.id.JobId;
 import org.ray.runtime.generated.Common.WorkerType;
 import org.ray.runtime.util.NetworkUtil;
@@ -144,7 +146,12 @@ public class RayConfig {
     } else {
       this.redisAddress = null;
     }
-    headRedisPort = config.getInt("ray.redis.head-port");
+
+    if (config.hasPath("ray.redis.head-port")) {
+      headRedisPort = config.getInt("ray.redis.head-port");
+    } else {
+      headRedisPort = NetworkUtil.getUnusedPort();
+    }
     numberRedisShards = config.getInt("ray.redis.shard-number");
     headRedisPassword = config.getString("ray.redis.head-password");
     redisPassword = config.getString("ray.redis.password");

java/runtime/src/main/resources/ray.default.conf

@@ -58,7 +58,8 @@ ray {
     // Redis server, Raylet and object store.
     address: ""
     // If `redis.server` isn't provided, which port we should use to start redis server.
-    head-port: 6379
+    // If `head-port` is not provided, it will be generated randomly.
+    // head-port: 6379
     // Below passwords should be consistent with the one defined in python/ray/ray_constants.py.
     // The password used to start the redis server on the head node.
     head-password: "5241590000000000"

java/test/src/main/java/org/ray/api/test/RayConfigTest.java

@@ -7,6 +7,8 @@ import org.testng.annotations.Test;
 
 public class RayConfigTest {
 
+  public final static int NUM_RETRIES = 5;
+
   @Test
   public void testCreateRayConfig() {
     try {
@@ -18,6 +20,27 @@ public class RayConfigTest {
       // Unset system properties.
       System.clearProperty("ray.job.resource-path");
     }
+  }
 
+  @Test
+  public void testGenerateHeadPortRandomly() {
+    boolean isSame = true;
+    final int port1 = RayConfig.create().headRedisPort;
+    // If we the 2 ports are the same, let's retry.
+    // This is used to avoid any flaky chance.
+    for (int i = 0; i < NUM_RETRIES; ++i) {
+      final int port2 = RayConfig.create().headRedisPort;
+      if (port1 != port2) {
+        isSame = false;
+        break;
+      }
+    }
+    Assert.assertFalse(isSame);
+  }
+
+  @Test
+  public void testSpecifyHeadPort() {
+    System.setProperty("ray.redis.head-port", "11111");
+    Assert.assertEquals(RayConfig.create().headRedisPort, 11111);
   }
 }