[Java] Replace binary rewrite with Remote Lambda Cache (SerdeLambda)

.gitignore

@@ -147,3 +147,8 @@ build
 
 # Vscode
 .vscode/
+
+*.iml
+java/**/target
+java/run
+java/test/lib

java/api/src/main/java/org/ray/api/Ray.java

@@ -103,13 +103,6 @@ static RayApi internal() {
     return impl;
   }
 
-  /**
-   * whether to use remote lambda.
-   */
-  public static boolean isRemoteLambda() {
-    return impl.isRemoteLambda();
-  }
-
   /**
    * for ray's app's log.
    */

java/api/src/main/java/org/ray/api/RayActor.java

@@ -10,7 +10,7 @@
  * Ray actor abstraction.
  */
 public class RayActor<T> extends RayObject<T> implements Externalizable {
-
+  public static final RayActor<?> nil = new RayActor<>(UniqueID.nil, UniqueID.nil);
   private static final long serialVersionUID = 1877485807405645036L;
 
   private int taskCounter = 0;
@@ -84,4 +84,4 @@ public void readExternal(ObjectInput in) throws IOException, ClassNotFoundExcept
     this.actorHandleId = (UniqueID) in.readObject();
     this.taskCursor = (UniqueID) in.readObject();
   }
-}
+}

java/api/src/main/java/org/ray/api/RayApi.java

@@ -3,7 +3,7 @@
 import java.io.Serializable;
 import java.util.Collection;
 import java.util.List;
-import org.ray.api.internal.Callable;
+import org.ray.api.internal.RayFunc;
 import org.ray.util.exception.TaskExecutionException;
 
 /**
@@ -54,14 +54,13 @@ public interface RayApi {
    * submit a new task by invoking a remote function.
    *
    * @param taskId      nil
-   * @param funcRun     the target running function with @RayRemote
+   * @param funcCls     the target running function's class
+   * @param lambda      the target running function
    * @param returnCount the number of to-be-returned objects from funcRun
    * @param args        arguments to this funcRun, can be its original form or RayObject
    * @return a set of ray objects with their return ids
    */
-  RayObjects call(UniqueID taskId, Callable funcRun, int returnCount, Object... args);
-
-  RayObjects call(UniqueID taskId, Class<?> funcCls, Serializable lambda, int returnCount,
+  RayObjects call(UniqueID taskId, Class<?> funcCls, RayFunc lambda, int returnCount,
                   Object... args);
 
   /**
@@ -71,34 +70,27 @@ RayObjects call(UniqueID taskId, Class<?> funcCls, Serializable lambda, int retu
    * outputs with a set of labels (usually with Integer or String).
    *
    * @param taskId    nil
-   * @param funcRun   the target running function with @RayRemote
-   * @param returnIds a set of labels to be used by the returned objects
+   * @param funcCls   the target running function's class
+   * @param lambda    the target running function
+   * @param returnids a set of labels to be used by the returned objects
    * @param args      arguments to this funcRun, can be its original form or
    *                  RayObject<original-type>
    * @return a set of ray objects with their labels and return ids
    */
-  <R, RIDT> RayMap<RIDT, R> callWithReturnLabels(UniqueID taskId, Callable funcRun,
-                                                 Collection<RIDT> returnIds, Object... args);
-
   <R, RIDT> RayMap<RIDT, R> callWithReturnLabels(UniqueID taskId, Class<?> funcCls,
-                                                 Serializable lambda, Collection<RIDT> returnids,
-                                                 Object... args);
+      RayFunc lambda, Collection<RIDT> returnids, Object... args);
 
   /**
    * a special case for the above RID-based labeling as <0...returnCount - 1>.
    *
    * @param taskId      nil
-   * @param funcRun     the target running function with @RayRemote
+   * @param funcCls     the target running function's class
+   * @param lambda      the target running function
    * @param returnCount the number of to-be-returned objects from funcRun
    * @param args        arguments to this funcRun, can be its original form or
    *                    RayObject<original-type>
    * @return an array of returned objects with their Unique ids
    */
-  <R> RayList<R> callWithReturnIndices(UniqueID taskId, Callable funcRun, Integer returnCount,
-                                       Object... args);
-
-  <R> RayList<R> callWithReturnIndices(UniqueID taskId, Class<?> funcCls, Serializable lambda,
+  <R> RayList<R> callWithReturnIndices(UniqueID taskId, Class<?> funcCls, RayFunc lambda,
                                        Integer returnCount, Object... args);
-
-  boolean isRemoteLambda();
 }