[Serve] Define BackendConfig protobuf and adapt it in Java

java/BUILD.bazel

@@ -144,6 +144,7 @@ define_java_module(
         "@maven//:com_google_guava_guava",
         "@maven//:org_slf4j_slf4j_api",
         "@maven//:org_testng_testng",
+        "@maven//:com_caucho_hessian",
     ],
     visibility = ["//visibility:public"],
     deps = [
@@ -152,6 +153,7 @@ define_java_module(
         "@maven//:com_google_guava_guava",
         "@maven//:org_apache_commons_commons_lang3",
         "@maven//:org_slf4j_slf4j_api",
+        "@maven//:com_caucho_hessian",
     ],
 )
 
@@ -178,6 +180,11 @@ java_proto_compile(
     deps = ["@//src/ray/protobuf:gcs_proto"],
 )
 
+java_proto_compile(
+    name = "serve_java_proto",
+    deps = ["@//src/ray/protobuf:serve_proto"],
+)
+
 filegroup(
     name = "all_java_proto",
     srcs = [
@@ -201,6 +208,7 @@ genrule(
     srcs = [
         ":all_java_proto",
         ":copy_pom_file",
+        ":serve_java_proto",
     ],
     outs = ["cp_java_generated.out"],
     cmd = """
@@ -210,6 +218,10 @@ genrule(
         for f in $(locations //java:all_java_proto); do
             unzip "$$f" -x META-INF/MANIFEST.MF -d "$$WORK_DIR/java/runtime/src/main/java"
         done
+        rm -rf "$$WORK_DIR/java/serve/src/main/java/io/ray/serve/generated"
+        for f in $(locations //java:serve_java_proto); do
+            unzip "$$f" -x META-INF/MANIFEST.MF -d "$$WORK_DIR/java/serve/src/main/java"
+        done
         date > $@
     """,
     local = 1,

java/dependencies.bzl

@@ -24,6 +24,7 @@ def gen_java_deps():
             "com.lmax:disruptor:3.3.4",
             "org.yaml:snakeyaml:1.26",
             "net.java.dev.jna:jna:5.5.0",
+            "com.caucho:hessian:4.0.62",
             maven.artifact(
                 group = "org.testng",
                 artifact = "testng",

java/serve/pom.xml

@@ -27,6 +27,11 @@
       <artifactId>ray-runtime</artifactId>
       <version>${project.version}</version>
     </dependency>
+<dependency>
+  <groupId>com.caucho</groupId>
+  <artifactId>hessian</artifactId>
+  <version>4.0.62</version>
+</dependency>
 <dependency>
   <groupId>com.google.guava</groupId>
   <artifactId>guava</artifactId>

java/serve/src/main/java/io/ray/serve/RayServeReplica.java

@@ -9,6 +9,7 @@
 import io.ray.runtime.metric.MetricConfig;
 import io.ray.runtime.metric.Metrics;
 import io.ray.serve.api.Serve;
+import io.ray.serve.generated.BackendConfig;
 import io.ray.serve.poll.KeyListener;
 import io.ray.serve.poll.KeyType;
 import io.ray.serve.poll.LongPollClient;
@@ -190,7 +191,7 @@ private Method getRunnerMethod(Query query) {
   public void drainPendingQueries() {
     while (true) {
       try {
-        Thread.sleep(config.getExperimentalGracefulShutdownWaitLoopS() * 1000);
+        Thread.sleep((long) (config.getExperimentalGracefulShutdownWaitLoopS() * 1000));
       } catch (InterruptedException e) {
         LOGGER.error(
             "Replica {} was interrupted in sheep when draining pending queries", replicaTag);
@@ -212,8 +213,8 @@ public void drainPendingQueries() {
    *
    * @param userConfig new user's configuration
    */
-  private void reconfigure(Object userConfig) {
-    if (userConfig == null) {
+  private void reconfigure(com.google.protobuf.ByteString userConfig) {
+    if (userConfig == null || userConfig.size() == 0) {
       return;
     }
     try {

java/serve/src/main/java/io/ray/serve/RayServeWrappedReplica.java

@@ -4,7 +4,11 @@
 import io.ray.api.BaseActorHandle;
 import io.ray.api.Ray;
 import io.ray.serve.api.Serve;
+import io.ray.serve.generated.BackendConfig;
+import io.ray.serve.serializer.Hessian2Seserializer;
+import io.ray.serve.util.BackendConfigUtil;
 import io.ray.serve.util.ReflectUtil;
+import java.io.IOException;
 import java.lang.reflect.InvocationTargetException;
 import java.util.Optional;
 import org.apache.commons.lang3.StringUtils;
@@ -19,11 +23,17 @@ public RayServeWrappedReplica(
       String backendTag,
       String replicaTag,
       String backendDef,
-      Object[] initArgs,
-      BackendConfig backendConfig,
+      byte[] initArgsbytes,
+      byte[] backendConfigBytes,
       String controllerName)
       throws ClassNotFoundException, NoSuchMethodException, InstantiationException,
-          IllegalAccessException, IllegalArgumentException, InvocationTargetException {
+          IllegalAccessException, IllegalArgumentException, InvocationTargetException, IOException {
+
+    // Parse BackendConfig.
+    BackendConfig backendConfig = BackendConfigUtil.parseFrom(backendConfigBytes);
+
+    // Parse init args.
+    Object[] initArgs = parseInitArgs(initArgsbytes, backendConfig);
 
     // Instantiate the object defined by backendDef.
     Class backendClass = Class.forName(backendDef);
@@ -43,6 +53,24 @@ public RayServeWrappedReplica(
     backend = new RayServeReplica(callable, backendConfig, optional.get());
   }
 
+  private Object[] parseInitArgs(byte[] initArgsbytes, BackendConfig backendConfig)
+      throws IOException {
+
+    if (initArgsbytes == null || initArgsbytes.length == 0) {
+      return new Object[0];
+    }
+
+    if (!backendConfig.getIsCrossLanguage()) {
+      // If the construction request is from Java API, deserialize initArgsbytes to Object[]
+      // directly.
+      Object result = Hessian2Seserializer.decode(initArgsbytes);
+      return (Object[]) result;
+    } else {
+      // TODO
+      return new Object[0];
+    }
+  }
+
   /**
    * The entry method to process the request.
    *
@@ -51,7 +79,7 @@ public RayServeWrappedReplica(
    *     backendDef.
    * @return the result of request being processed
    */
-  public Object handle_request(RequestMetadata requestMetadata, Object[] requestArgs) {
+  public Object handleRequest(RequestMetadata requestMetadata, Object[] requestArgs) {
     return backend.handleRequest(new Query(requestArgs, requestMetadata));
   }
 
@@ -61,7 +89,7 @@ public void ready() {
   }
 
   /** Wait until there is no request in processing. It is used for stopping replica gracefully. */
-  public void drain_pending_queries() {
+  public void drainPendingQueries() {
     backend.drainPendingQueries();
   }
 }

java/serve/src/main/java/io/ray/serve/RequestMetadata.java

@@ -12,7 +12,7 @@ public class RequestMetadata implements Serializable {
 
   private String endpoint;
 
-  private String callMethod = "__call__";
+  private String callMethod = "call";
 
   private String httpMethod;
 

java/serve/src/main/java/io/ray/serve/serializer/Hessian2Seserializer.java

@@ -0,0 +1,37 @@
+package io.ray.serve.serializer;
+
+import com.caucho.hessian.io.Hessian2Input;
+import com.caucho.hessian.io.Hessian2Output;
+import com.caucho.hessian.io.SerializerFactory;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+
+public class Hessian2Seserializer {
+
+  private static SerializerFactory DEFAULT_FACTORY = new SerializerFactory();
+
+  public static byte[] encode(Object data) throws IOException {
+    return encode(data, DEFAULT_FACTORY);
+  }
+
+  public static byte[] encode(Object data, SerializerFactory serializerFactory) throws IOException {
+    ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+    Hessian2Output hessian2Output = new Hessian2Output(byteArrayOutputStream);
+    hessian2Output.setSerializerFactory(serializerFactory);
+    hessian2Output.writeObject(data);
+    hessian2Output.flush();
+    return byteArrayOutputStream.toByteArray();
+  }
+
+  public static Object decode(byte[] data) throws IOException {
+    return decode(data, DEFAULT_FACTORY);
+  }
+
+  public static Object decode(byte[] data, SerializerFactory serializerFactory) throws IOException {
+    ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(data);
+    Hessian2Input hessian2Input = new Hessian2Input(byteArrayInputStream);
+    hessian2Input.setSerializerFactory(serializerFactory);
+    return hessian2Input.readObject();
+  }
+}

java/serve/src/main/java/io/ray/serve/util/BackendConfigUtil.java

@@ -0,0 +1,69 @@
+package io.ray.serve.util;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import com.google.protobuf.InvalidProtocolBufferException;
+import io.ray.serve.RayServeException;
+import io.ray.serve.generated.BackendConfig;
+import io.ray.serve.generated.BackendLanguage;
+
+public class BackendConfigUtil {
+
+  public static BackendConfig parseFrom(byte[] backendConfigBytes)
+      throws InvalidProtocolBufferException {
+
+    // Parse BackendConfig from byte[].
+    BackendConfig inputBackendConfig = BackendConfig.parseFrom(backendConfigBytes);
+    if (inputBackendConfig == null) {
+      return null;
+    }
+
+    // Set default values.
+    BackendConfig.Builder builder = BackendConfig.newBuilder();
+
+    if (inputBackendConfig.getNumReplicas() == 0) {
+      builder.setNumReplicas(1);
+    } else {
+      builder.setNumReplicas(inputBackendConfig.getNumReplicas());
+    }
+
+    Preconditions.checkArgument(
+        inputBackendConfig.getMaxConcurrentQueries() >= 0, "max_concurrent_queries must be >= 0");
+    if (inputBackendConfig.getMaxConcurrentQueries() == 0) {
+      builder.setMaxConcurrentQueries(100);
+    } else {
+      builder.setMaxConcurrentQueries(inputBackendConfig.getMaxConcurrentQueries());
+    }
+
+    builder.setUserConfig(inputBackendConfig.getUserConfig());
+
+    if (inputBackendConfig.getExperimentalGracefulShutdownWaitLoopS() == 0) {
+      builder.setExperimentalGracefulShutdownWaitLoopS(2);
+    } else {
+      builder.setExperimentalGracefulShutdownWaitLoopS(
+          inputBackendConfig.getExperimentalGracefulShutdownWaitLoopS());
+    }
+
+    if (inputBackendConfig.getExperimentalGracefulShutdownTimeoutS() == 0) {
+      builder.setExperimentalGracefulShutdownTimeoutS(20);
+    } else {
+      builder.setExperimentalGracefulShutdownTimeoutS(
+          inputBackendConfig.getExperimentalGracefulShutdownTimeoutS());
+    }
+
+    builder.setIsCrossLanguage(inputBackendConfig.getIsCrossLanguage());
+
+    if (inputBackendConfig.getBackendLanguage() == BackendLanguage.UNRECOGNIZED) {
+      throw new RayServeException(
+          LogUtil.format(
+              "Unrecognized backend language {}. Backend language must be in {}.",
+              inputBackendConfig.getBackendLanguageValue(),
+              Lists.newArrayList(BackendLanguage.values())));
+    } else {
+      builder.setBackendLanguage(inputBackendConfig.getBackendLanguage());
+    }
+
+    return builder.build();
+  }
+
+}