Support transfer ObjectRef from JAVA application to PYTHON deployment (ray-project#45729)

## Why are these changes needed?

In the current implementation, it's not possible to transfer an ObjectRef from a Java application to a Python deployment. For more details, please refer to this issue: ray-project#45676

**Root Cause**
In the original approach, all arguments were bundled together and passed as a list parameter, like so: {1, 2, "parameter3"}. If one of these parameters was an ObjectRef, it would lead to serialization issues, preventing cross-language transfer: {1, 2, "parameter3", objectRef4}.

**Proposed Solution**
In the new approach, each argument is passed independently. The ObjectRef is implemented as PassByReference, which avoids the previous serialization issues and enables the transfer of ObjectRef: {1, 2, "parameter3", objectRef4}. 

This change enhances the interoperability between Java applications and Python deployments in Ray.

Author: yucai

java/serve/src/main/java/io/ray/serve/router/ReplicaSet.java

@@ -6,20 +6,17 @@
 import io.ray.api.ObjectRef;
 import io.ray.api.PyActorHandle;
 import io.ray.api.Ray;
+import io.ray.api.call.PyActorTaskCaller;
 import io.ray.api.function.PyActorMethod;
 import io.ray.serve.common.Constants;
 import io.ray.serve.exception.RayServeException;
 import io.ray.serve.generated.ActorNameList;
 import io.ray.serve.replica.RayServeWrappedReplica;
 import io.ray.serve.util.CollectionUtil;
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
+import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
 import org.apache.commons.lang3.RandomUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -108,12 +105,15 @@ private ObjectRef<Object> tryAssignReplica(Query query) {
         handles.get(randomIndex); // TODO controll concurrency using maxConcurrentQueries
     LOGGER.debug("Assigned query {} to replica {}.", query.getMetadata().getRequestId(), replica);
     if (replica instanceof PyActorHandle) {
-      return ((PyActorHandle) replica)
-          .task(
-              PyActorMethod.of("handle_request_from_java"),
-              query.getMetadata().toByteArray(),
-              query.getArgs())
-          .remote();
+      Object[] args =
+          Stream.concat(
+                  Stream.of(query.getMetadata().toByteArray()),
+                  Arrays.stream((Object[]) query.getArgs()))
+              .toArray();
+      PyActorTaskCaller<Object> pyCaller =
+          new PyActorTaskCaller<>(
+              (PyActorHandle) replica, PyActorMethod.of("handle_request_from_java"), args);
+      return pyCaller.remote();
     } else {
       return ((ActorHandle<RayServeWrappedReplica>) replica)
           .task(

java/serve/src/test/java/io/ray/serve/deployment/CrossLanguageDeploymentTest.java

@@ -1,5 +1,7 @@
 package io.ray.serve.deployment;
 
+import io.ray.api.ObjectRef;
+import io.ray.api.Ray;
 import io.ray.serve.BaseServeTest;
 import io.ray.serve.api.Serve;
 import io.ray.serve.generated.DeploymentLanguage;
@@ -58,6 +60,21 @@ public void createPyClassTest() {
     Assert.assertEquals(handle.method("increase").remote("6").result(), "34");
   }
 
+  @Test
+  public void createPyClassWithObjectRefTest() {
+    Application deployment =
+        Serve.deployment()
+            .setLanguage(DeploymentLanguage.PYTHON)
+            .setName("createPyClassWithObjectRefTest")
+            .setDeploymentDef(PYTHON_MODULE + ".Counter")
+            .setNumReplicas(1)
+            .bind("28");
+
+    DeploymentHandle handle = Serve.run(deployment).get();
+    ObjectRef<Integer> numRef = Ray.put(10);
+    Assert.assertEquals(handle.method("increase").remote(numRef).result(), "38");
+  }
+
   @Test
   public void createPyMethodTest() {
     Application deployment =
@@ -71,6 +88,20 @@ public void createPyMethodTest() {
     Assert.assertEquals(handle.method("__call__").remote("6").result(), "6");
   }
 
+  @Test
+  public void createPyMethodWithObjectRefTest() {
+    Application deployment =
+        Serve.deployment()
+            .setLanguage(DeploymentLanguage.PYTHON)
+            .setName("createPyMethodWithObjectRefTest")
+            .setDeploymentDef(PYTHON_MODULE + ".echo_server")
+            .setNumReplicas(1)
+            .bind();
+    DeploymentHandle handle = Serve.run(deployment).get();
+    ObjectRef<String> numRef = Ray.put("10");
+    Assert.assertEquals(handle.method("__call__").remote(numRef).result(), "10");
+  }
+
   @Test
   public void userConfigTest() throws InterruptedException {
     Application deployment =

java/test/src/main/java/io/ray/test/CrossLanguageInvocationTest.java

@@ -139,6 +139,13 @@ public void testCallingPythonFunction() {
       Assert.assertEquals(((Number) r7array[0]).intValue(), input7array[0]);
       Assert.assertEquals(((Number) r7array[1]).intValue(), input7array[1]);
     }
+    // objectRef
+    {
+      ObjectRef<Integer> input = Ray.put(1);
+      ObjectRef<Integer> res =
+          Ray.task(PyFunction.of(PYTHON_MODULE, "py_return_input", Integer.class), input).remote();
+      Assert.assertEquals(res.get(), input.get());
+    }
     // Unsupported types, all Java specific types, e.g. List / Map...
     {
       Assert.expectThrows(
@@ -173,6 +180,11 @@ public void testCallingPythonActor() {
     ObjectRef<byte[]> res =
         actor.task(PyActorMethod.of("increase", byte[].class), "1".getBytes()).remote();
     Assert.assertEquals(res.get(), "2".getBytes());
+
+    ObjectRef<String> numRef = Ray.put("2");
+    ObjectRef<byte[]> res2 =
+        actor.task(PyActorMethod.of("increase", byte[].class), numRef).remote();
+    Assert.assertEquals(res2.get(), "4".getBytes());
   }
 
   @Test

python/ray/serve/_private/replica.py

@@ -560,7 +560,7 @@ async def handle_request_from_java(
         )
         with self._wrap_user_method_call(request_metadata):
             return await self._user_callable_wrapper.call_user_method(
-                request_metadata, request_args[0], request_kwargs
+                request_metadata, request_args, request_kwargs
             )
 
     async def is_allocated(self) -> str: