[Serve] Upgrade deprecated calls

ci/pipeline/determine_tests_to_run.py

@@ -183,6 +183,7 @@ def get_commit_range():
                 RAY_CI_SERVE_AFFECTED = 1
                 RAY_CI_LINUX_WHEELS_AFFECTED = 1
                 RAY_CI_MACOS_WHEELS_AFFECTED = 1
+                RAY_CI_JAVA_AFFECTED = 1
             elif changed_file.startswith("python/ray/dashboard"):
                 RAY_CI_DASHBOARD_AFFECTED = 1
                 # https://github.com/ray-project/ray/pull/15981

python/ray/serve/_private/client.py

@@ -484,7 +484,7 @@ def get_deploy_args(
             "deployment_config_proto_bytes": deployment_config.to_proto_bytes(),
             "replica_config_proto_bytes": replica_config.to_proto_bytes(),
             "route_prefix": route_prefix,
-            "deployer_job_id": ray.get_runtime_context().job_id,
+            "deployer_job_id": ray.get_runtime_context().get_job_id(),
             "is_driver_deployment": is_driver_deployment,
         }
 

python/ray/serve/_private/common.py

@@ -162,7 +162,7 @@ def __init__(
         deployment_config: DeploymentConfig,
         replica_config: ReplicaConfig,
         start_time_ms: int,
-        deployer_job_id: "ray._raylet.JobID",
+        deployer_job_id: str,
         actor_name: Optional[str] = None,
         version: Optional[str] = None,
         end_time_ms: Optional[int] = None,
@@ -225,7 +225,7 @@ def from_proto(cls, proto: DeploymentInfoProto):
             "actor_name": proto.actor_name if proto.actor_name != "" else None,
             "version": proto.version if proto.version != "" else None,
             "end_time_ms": proto.end_time_ms if proto.end_time_ms != 0 else None,
-            "deployer_job_id": ray.get_runtime_context().job_id,
+            "deployer_job_id": ray.get_runtime_context().get_job_id(),
         }
 
         return cls(**data)

python/ray/serve/_private/deployment_state.py

@@ -473,7 +473,7 @@ def check_ready(self) -> Tuple[ReplicaStartupStatus, Optional[DeploymentVersion]
                 )
                 self._health_check_period_s = deployment_config.health_check_period_s
                 self._health_check_timeout_s = deployment_config.health_check_timeout_s
-                self._node_id = ray.get(self._allocated_obj_ref).hex()
+                self._node_id = ray.get(self._allocated_obj_ref)
             except Exception:
                 logger.exception(f"Exception in deployment '{self._deployment_name}'")
                 return ReplicaStartupStatus.FAILED, None

python/ray/serve/_private/replica.py

@@ -212,7 +212,7 @@ async def is_allocated(self) -> str:
 
             Return the NodeID of this replica
             """
-            return ray.get_runtime_context().node_id
+            return ray.get_runtime_context().get_node_id()
 
         async def is_initialized(
             self, user_config: Optional[Any] = None, _after: Optional[Any] = None

python/ray/serve/controller.py

@@ -271,7 +271,7 @@ def _put_serve_snapshot(self) -> None:
             entry = dict()
             entry["name"] = deployment_name
             entry["namespace"] = ray.get_runtime_context().namespace
-            entry["ray_job_id"] = deployment_info.deployer_job_id.hex()
+            entry["ray_job_id"] = deployment_info.deployer_job_id
             entry["class_name"] = deployment_info.replica_config.deployment_def_name
             entry["version"] = deployment_info.version
             entry["http_route"] = route_prefix
@@ -351,7 +351,7 @@ def deploy(
         deployment_config_proto_bytes: bytes,
         replica_config_proto_bytes: bytes,
         route_prefix: Optional[str],
-        deployer_job_id: Union["ray._raylet.JobID", bytes],
+        deployer_job_id: Union[str, bytes],
         is_driver_deployment: Optional[bool] = False,
     ) -> bool:
         if route_prefix is not None:
@@ -381,10 +381,13 @@ def deploy(
             autoscaling_policy = BasicAutoscalingPolicy(autoscaling_config)
         else:
             autoscaling_policy = None
+
+        # Java API passes in JobID as bytes
         if isinstance(deployer_job_id, bytes):
             deployer_job_id = ray.JobID.from_int(
                 int.from_bytes(deployer_job_id, "little")
-            )
+            ).hex()
+
         deployment_info = DeploymentInfo(
             actor_name=name,
             version=version,

python/ray/serve/tests/test_cross_language.py

@@ -37,7 +37,7 @@ def test_controller_starts_java_replica(shutdown_only):  # noqa: F811
             deployment_config_proto_bytes=config.to_proto_bytes(),
             replica_config_proto_bytes=replica_config.to_proto_bytes(),
             route_prefix=None,
-            deployer_job_id=ray.get_runtime_context().job_id,
+            deployer_job_id=ray.get_runtime_context().get_job_id(),
         )
     )
     assert updating

python/ray/serve/tests/test_deployment_state.py

@@ -180,7 +180,7 @@ def deployment_info(
             num_replicas=num_replicas, user_config=user_config, **config_opts
         ),
         replica_config=ReplicaConfig.create(lambda x: x),
-        deployer_job_id=ray.JobID.nil(),
+        deployer_job_id="",
         is_driver_deployment=is_driver_deployment,
     )
 

python/ray/serve/tests/test_standalone.py

@@ -143,7 +143,7 @@ def test_detached_deployment(ray_cluster):
 
     # Create first job, check we can run a simple serve endpoint
     ray.init(head_node.address, namespace=SERVE_NAMESPACE)
-    first_job_id = ray.get_runtime_context().job_id
+    first_job_id = ray.get_runtime_context().get_job_id()
     serve.start(detached=True)
 
     @serve.deployment(route_prefix="/say_hi_f")
@@ -159,7 +159,7 @@ def f(*args):
 
     # Create the second job, make sure we can still create new deployments.
     ray.init(head_node.address, namespace="serve")
-    assert ray.get_runtime_context().job_id != first_job_id
+    assert ray.get_runtime_context().get_job_id() != first_job_id
 
     @serve.deployment(route_prefix="/say_hi_g")
     def g(*args):