[ClientBuilder] Code takes precedence over environment (#16112)

* no override address

* correct ordering

python/ray/client_builder.py

@@ -166,11 +166,11 @@ def client(address: Optional[str] = None) -> ClientBuilder:
         * ``"module://inner_address"``: load module.ClientBuilder & pass
             inner_address
     """
-    override_address = os.environ.get(RAY_ADDRESS_ENVIRONMENT_VARIABLE)
-    if override_address:
+    env_address = os.environ.get(RAY_ADDRESS_ENVIRONMENT_VARIABLE)
+    if env_address and address is None:
         logger.debug(
-            f"Using address ({override_address}) instead of "
-            f"({address}) because {RAY_ADDRESS_ENVIRONMENT_VARIABLE} is set")
-        address = override_address
+            f"Using address ({env_address}) instead of auto-detection "
+            f"because {RAY_ADDRESS_ENVIRONMENT_VARIABLE} is set.")
+        address = env_address
 
     return _get_builder_from_address(address)

python/ray/tests/test_client_builder.py

@@ -1,3 +1,4 @@
+import os
 import pytest
 import subprocess
 import sys
@@ -210,6 +211,27 @@ def test_disconnect(call_ray_stop_only):
     ctx.disconnect()
     # Check idempotency
     ctx.disconnect()
-
     with pytest.raises(ray.exceptions.RaySystemError):
         ray.put(300)
+
+
+def test_address_resolution(ray_start_regular_shared):
+    server = ray_client_server.serve("localhost:50055")
+    with ray.client("localhost:50055").connect():
+        assert ray.util.client.ray.is_connected()
+
+    try:
+        os.environ["RAY_ADDRESS"] = "local"
+        with ray.client("localhost:50055").connect():
+            # client(...) takes precedence of RAY_ADDRESS=local
+            assert ray.util.client.ray.is_connected()
+
+        ray.client(None).connect()
+        assert ray.worker.global_worker.node.is_head()
+
+    finally:
+        if os.environ["RAY_ADDRESS"]:
+            del os.environ["RAY_ADDRESS"]
+
+        server.stop(0)
+        subprocess.check_output("ray stop --force", shell=True)