remove usage of ray.worker.global_worker.connected (#445)

Author: pang-wu

python/raydp/tests/test_data_owner_transfer.py

@@ -7,6 +7,7 @@
 import ray
 from ray._private.client_mode_hook import client_mode_wrap
 from ray.exceptions import RayTaskError, OwnerDiedError
+import ray.util.client as ray_client
 import raydp
 from raydp.spark import PartitionObjectsOwner
 from pyspark.sql import SparkSession
@@ -42,7 +43,7 @@ def test_fail_without_data_ownership_transfer(ray_cluster, jdk17_extra_spark_con
 
   # skipping this to be compatible with ray 2.4.0
   # see issue #343
-  if not ray.worker.global_worker.connected:
+  if ray_client.ray.is_connected():
         pytest.skip("Skip this test if using ray client")
 
   from raydp.spark.dataset import spark_dataframe_to_ray_dataset
@@ -90,7 +91,7 @@ def test_data_ownership_transfer(ray_cluster, jdk17_extra_spark_configs):
   This test should be able to execute till the end without crash as expected.
   """
 
-  if not ray.worker.global_worker.connected:
+  if ray_client.ray.is_connected():
         pytest.skip("Skip this test if using ray client")
 
   from raydp.spark.dataset import spark_dataframe_to_ray_dataset
@@ -153,7 +154,7 @@ def wake(self):
       def set_objects(self, objects):
           self.objects = objects
 
-  if not ray.worker.global_worker.connected:
+  if ray_client.ray.is_connected():
       pytest.skip("Skip this test if using ray client")
 
   from raydp.spark.dataset import spark_dataframe_to_ray_dataset

python/raydp/tests/test_mpi.py

@@ -20,6 +20,7 @@
 import pytest
 import ray
 from ray.util import placement_group, remove_placement_group
+import ray.util.client as ray_client
 
 from raydp.mpi import create_mpi_job, MPIJobContext, WorkerContext
 
@@ -28,7 +29,7 @@
 def test_mpi_start(ray_cluster):
     if platform.system() == "Darwin":
         pytest.skip("Skip MPI test on MacOS")
-    if not ray.worker.global_worker.connected:
+    if ray_client.ray.is_connected():
         pytest.skip("Skip MPI test if using ray client")
     job = create_mpi_job(job_name="test",
                          world_size=2,
@@ -61,7 +62,7 @@ def func(context: WorkerContext):
 def test_mpi_get_rank_address(ray_cluster):
     if platform.system() == "Darwin":
         pytest.skip("Skip MPI test on MacOS")
-    if not ray.worker.global_worker.connected:
+    if ray_client.ray.is_connected():
         pytest.skip("Skip MPI test if using ray client")
     with create_mpi_job(job_name="test",
                         world_size=2,
@@ -79,7 +80,7 @@ def test_mpi_get_rank_address(ray_cluster):
 def test_mpi_with_script_prepare_fn(ray_cluster):
     if platform.system() == "Darwin":
         pytest.skip("Skip MPI test on MacOS")
-    if not ray.worker.global_worker.connected:
+    if ray_client.ray.is_connected():
         pytest.skip("Skip MPI test if using ray client")
     def script_prepare_fn(context: MPIJobContext):
         context.add_env("is_test", "True")
@@ -106,7 +107,7 @@ def f(context: WorkerContext):
 def test_mpi_with_pg(ray_cluster):
     if platform.system() == "Darwin":
         pytest.skip("Skip MPI test on MacOS")
-    if not ray.worker.global_worker.connected:
+    if ray_client.ray.is_connected():
         pytest.skip("Skip MPI test if using ray client")
     pg = placement_group(bundles=[{"CPU": 2}], strategy="STRICT_SPREAD")
     with create_mpi_job(job_name="test",

python/raydp/tests/test_spark_cluster.py

@@ -31,6 +31,7 @@
 import raydp.utils as utils
 from raydp.spark.ray_cluster_master import RayDPSparkMaster, RAYDP_SPARK_MASTER_SUFFIX
 from ray.cluster_utils import Cluster
+import ray.util.client as ray_client
 
 def test_spark(spark_on_ray_small):
     spark = spark_on_ray_small
@@ -162,7 +163,7 @@ def test_ray_dataset_roundtrip(jdk17_extra_spark_configs):
 
     # skipping this to be compatible with ray 2.4.0
     # see issue #343
-    if not ray.worker.global_worker.connected:
+    if ray_client.ray.is_connected():
         pytest.skip("Skip this test if using ray client")
     spark_df = spark.createDataFrame([(1, "a"), (2, "b"), (3, "c")], ["one", "two"])
     rows = [(r.one, r.two) for r in spark_df.take(3)]
@@ -188,7 +189,7 @@ def test_ray_dataset_roundtrip(jdk17_extra_spark_configs):
 def test_ray_dataset_to_spark(spark_on_ray_2_executors):
     # skipping this to be compatible with ray 2.4.0
     # see issue #343
-    if not ray.worker.global_worker.connected:
+    if ray_client.ray.is_connected():
         pytest.skip("Skip this test if using ray client")
     spark = spark_on_ray_2_executors
     n = 5