diff --git a/python/ray/_private/node.py b/python/ray/_private/node.py
index 5fb963e05cd9..e9a3960ffe22 100644
--- a/python/ray/_private/node.py
+++ b/python/ray/_private/node.py
@@ -270,7 +270,11 @@ def __init__(
 
         # Pick a GCS server port.
         if head:
-            gcs_server_port = os.getenv(ray_constants.GCS_PORT_ENVIRONMENT_VARIABLE)
+            from ray.autoscaler._private.constants import env_integer
+
+            gcs_server_port = env_integer(
+                ray_constants.GCS_PORT_ENVIRONMENT_VARIABLE, 0
+            )
             if gcs_server_port:
                 ray_params.update_if_absent(gcs_server_port=gcs_server_port)
             if ray_params.gcs_server_port is None or ray_params.gcs_server_port == 0:
diff --git a/python/ray/autoscaler/_private/constants.py b/python/ray/autoscaler/_private/constants.py
index 4ace80e91ec7..548d9de0fecc 100644
--- a/python/ray/autoscaler/_private/constants.py
+++ b/python/ray/autoscaler/_private/constants.py
@@ -11,12 +11,19 @@
 
 
 def env_integer(key, default):
+    # If env is set incorrectly by user, it will be passed to user.
     if key in os.environ:
-        val = os.environ[key]
-        if val == "inf":
-            return sys.maxsize
-        else:
-            return int(val)
+        try:
+            val = os.environ[key]
+            if val == "inf":
+                return sys.maxsize
+            else:
+                return int(val)
+        except Exception:
+            raise ValueError(
+                f"Env which should be int is set incorrectly. Error env is {key}: {val}",
+            )
+
     return default
 
 
diff --git a/python/ray/tests/BUILD b/python/ray/tests/BUILD
index 26623a66b6c0..cc0425336484 100644
--- a/python/ray/tests/BUILD
+++ b/python/ray/tests/BUILD
@@ -304,6 +304,7 @@ py_test_module_list(
     "test_usage_stats.py",
     "test_placement_group_3.py",
     "test_placement_group_5.py",
+    "test_set_gcs_server_port.py",
   ],
   size = "large",
   tags = ["exclusive", "large_size_python_tests_shard_0", "team:core"],
diff --git a/python/ray/tests/test_set_gcs_server_port.py b/python/ray/tests/test_set_gcs_server_port.py
new file mode 100644
index 000000000000..cee141251c1f
--- /dev/null
+++ b/python/ray/tests/test_set_gcs_server_port.py
@@ -0,0 +1,69 @@
+import os
+import random
+import sys
+import socket
+
+import pytest
+
+import ray
+
+# LocalHost
+host = "127.0.0.1"
+
+
+# Function to check if the port is occupied.
+def detect_listened_port(port: int):
+    try:
+        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        s.settimeout(1)
+        s.connect((host, port))
+        return True
+    except Exception:
+        return False
+    finally:
+        s.close()
+
+
+# If RAY_GCS_SERVER_PORT isn't set, gcs_server_port will be picked by
+# random.randint(port, 65535) when ray.init() is called.
+def test_as_gcs_server_port_is_not_set():
+    try:
+        ray.init()
+    except Exception:
+        assert False
+    finally:
+        ray.shutdown()
+
+
+# If RAY_GCS_SERVER_PORT is set, gcs_server_port will be specified.
+# Connection can be established.
+def test_as_gcs_server_port_is_set():
+    port_int = random.randint(6000, 10000)
+    port_str = str(port_int)
+    os.environ["RAY_GCS_SERVER_PORT"] = port_str
+    ray.init()
+    assert detect_listened_port(port_int)
+    ray.shutdown()
+
+
+# If RAY_GCS_SERVER_PORT is set as string, it will be ignored so that
+# ray.init() will be executed normally.
+def test_as_gcs_server_port_is_set_incorrectly():
+    os.environ["RAY_GCS_SERVER_PORT"] = "non-numeric"
+    try:
+        ray.init()
+    except Exception as e:
+        assert type(e) is ValueError
+    finally:
+        ray.shutdown()
+
+
+if __name__ == "__main__":
+
+    # Make subprocess happy in bazel.
+    os.environ["LC_ALL"] = "en_US.UTF-8"
+    os.environ["LANG"] = "en_US.UTF-8"
+    if os.environ.get("PARALLEL_CI"):
+        sys.exit(pytest.main(["-n", "auto", "--boxed", "-vs", __file__]))
+    else:
+        sys.exit(pytest.main(["-sv", __file__]))