Cross language exception (#10023)

java/runtime/src/main/java/io/ray/runtime/AbstractRayRuntime.java

@@ -8,7 +8,6 @@ import io.ray.api.BaseActorHandle;
 import io.ray.api.ObjectRef;
 import io.ray.api.PyActorHandle;
 import io.ray.api.WaitResult;
-import io.ray.api.exception.RayException;
 import io.ray.api.function.PyActorClass;
 import io.ray.api.function.PyActorMethod;
 import io.ray.api.function.PyFunction;
@@ -81,7 +80,7 @@ public abstract class AbstractRayRuntime implements RayRuntimeInternal {
   }
 
   @Override
-  public <T> T get(ObjectRef<T> objectRef) throws RayException {
+  public <T> T get(ObjectRef<T> objectRef) throws RuntimeException {
     List<T> ret = get(ImmutableList.of(objectRef));
     return ret.get(0);
   }

java/runtime/src/main/java/io/ray/runtime/RayRuntimeProxy.java

@@ -1,8 +1,8 @@
 package io.ray.runtime;
 
-import io.ray.api.exception.RayException;
 import io.ray.api.runtime.RayRuntime;
 import io.ray.runtime.config.RunMode;
+import io.ray.runtime.exception.RayException;
 import java.lang.reflect.InvocationHandler;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;

java/runtime/src/main/java/io/ray/runtime/exception/CrossLanguageException.java

@@ -0,0 +1,18 @@
+package io.ray.runtime.exception;
+
+import io.ray.runtime.generated.Common.Language;
+
+public class CrossLanguageException extends RayException {
+
+  private Language language;
+
+  public CrossLanguageException(io.ray.runtime.generated.Common.RayException exception) {
+    super(String.format("An exception raised from %s:\n%s", exception.getLanguage().name(),
+        exception.getFormattedExceptionString()));
+    this.language = exception.getLanguage();
+  }
+
+  public Language getLanguage() {
+    return this.language;
+  }
+}

java/runtime/src/main/java/io/ray/runtime/exception/RayActorException.java

@@ -0,0 +1,15 @@
+package io.ray.runtime.exception;
+
+/**
+ * Indicates that the actor died unexpectedly before finishing a task.
+ *
+ * This exception could happen either because the actor process dies while executing a task, or
+ * because a task is submitted to a dead actor.
+ */
+public class RayActorException extends RayException {
+
+  public RayActorException() {
+    super("The actor died unexpectedly before finishing this task.");
+  }
+
+}

java/runtime/src/main/java/io/ray/runtime/exception/RayException.java

@@ -0,0 +1,40 @@
+package io.ray.runtime.exception;
+
+import com.google.protobuf.ByteString;
+import com.google.protobuf.InvalidProtocolBufferException;
+import io.ray.runtime.generated.Common.Language;
+import io.ray.runtime.serializer.Serializer;
+
+public class RayException extends RuntimeException {
+
+  public RayException(String message) {
+    super(message);
+  }
+
+  public RayException(String message, Throwable cause) {
+    super(message, cause);
+  }
+
+  public byte[] toBytes() {
+    String formattedException = org.apache.commons.lang3.exception.ExceptionUtils
+        .getStackTrace(this);
+    io.ray.runtime.generated.Common.RayException.Builder builder =
+        io.ray.runtime.generated.Common.RayException.newBuilder();
+    builder.setLanguage(Language.JAVA);
+    builder.setFormattedExceptionString(formattedException);
+    builder.setSerializedException(ByteString.copyFrom(Serializer.encode(this).getLeft()));
+    return builder.build().toByteArray();
+  }
+
+  public static RayException fromBytes(byte[] serialized)
+      throws InvalidProtocolBufferException {
+    io.ray.runtime.generated.Common.RayException exception =
+        io.ray.runtime.generated.Common.RayException.parseFrom(serialized);
+    if (exception.getLanguage() == Language.JAVA) {
+      return Serializer
+          .decode(exception.getSerializedException().toByteArray(), RayException.class);
+    } else {
+      return new CrossLanguageException(exception);
+    }
+  }
+}

java/runtime/src/main/java/io/ray/runtime/exception/RayTaskException.java

@@ -0,0 +1,13 @@
+package io.ray.runtime.exception;
+
+import io.ray.runtime.util.NetworkUtil;
+import io.ray.runtime.util.SystemUtil;
+
+public class RayTaskException extends RayException {
+
+  public RayTaskException(String message, Throwable cause) {
+    super(String.format("(pid=%d, ip=%s) %s",
+        SystemUtil.pid(), NetworkUtil.getIpAddress(null), message), cause);
+  }
+
+}

java/runtime/src/main/java/io/ray/runtime/exception/RayWorkerException.java

@@ -0,0 +1,12 @@
+package io.ray.runtime.exception;
+
+/**
+ * Indicates that the worker died unexpectedly while executing a task.
+ */
+public class RayWorkerException extends RayException {
+
+  public RayWorkerException() {
+    super("The worker died unexpectedly while executing this task.");
+  }
+
+}

java/runtime/src/main/java/io/ray/runtime/exception/UnreconstructableException.java

@@ -0,0 +1,23 @@
+package io.ray.runtime.exception;
+
+import io.ray.api.id.ObjectId;
+
+/**
+ * Indicates that an object is lost (either evicted or explicitly deleted) and cannot be
+ * restarted.
+ *
+ * Note, this exception only happens for actor objects. If actor's current state is after object's
+ * creating task, the actor cannot re-run the task to reconstruct the object.
+ */
+public class UnreconstructableException extends RayException {
+
+  public final ObjectId objectId;
+
+  public UnreconstructableException(ObjectId objectId) {
+    super(String.format(
+        "Object %s is lost (either evicted or explicitly deleted) and cannot be reconstructed.",
+        objectId));
+    this.objectId = objectId;
+  }
+
+}

java/runtime/src/main/java/io/ray/runtime/object/ObjectSerializer.java

@@ -1,11 +1,12 @@
 package io.ray.runtime.object;
 
-import io.ray.api.exception.RayActorException;
-import io.ray.api.exception.RayTaskException;
-import io.ray.api.exception.RayWorkerException;
-import io.ray.api.exception.UnreconstructableException;
+import com.google.protobuf.InvalidProtocolBufferException;
 import io.ray.api.id.ObjectId;
-import io.ray.runtime.generated.Gcs.ErrorType;
+import io.ray.runtime.exception.RayActorException;
+import io.ray.runtime.exception.RayTaskException;
+import io.ray.runtime.exception.RayWorkerException;
+import io.ray.runtime.exception.UnreconstructableException;
+import io.ray.runtime.generated.Common.ErrorType;
 import io.ray.runtime.serializer.Serializer;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
@@ -70,7 +71,21 @@ public class ObjectSerializer {
       } else if (Arrays.equals(meta, UNRECONSTRUCTABLE_EXCEPTION_META)) {
         return new UnreconstructableException(objectId);
       } else if (Arrays.equals(meta, TASK_EXECUTION_EXCEPTION_META)) {
-        return Serializer.decode(data, objectType);
+        // Serialization logic of task execution exception: an instance of
+        // `io.ray.runtime.exception.RayTaskException`
+        //    -> a `RayException` protobuf message
+        //    -> protobuf-serialized bytes
+        //    -> MessagePack-serialized bytes.
+        // So here the `data` variable is MessagePack-serialized bytes, and the `serialized`
+        // variable is protobuf-serialized bytes. They are not the same.
+        byte[] serialized = Serializer.decode(data, byte[].class);
+        try {
+          return RayTaskException.fromBytes(serialized);
+        } catch (InvalidProtocolBufferException e) {
+          throw new IllegalArgumentException(
+              "Can't deserialize RayTaskException object: " + objectId
+                  .toString());
+        }
       } else if (Arrays.equals(meta, OBJECT_METADATA_TYPE_PYTHON)) {
         throw new IllegalArgumentException("Can't deserialize Python object: " + objectId
             .toString());
@@ -107,7 +122,12 @@ public class ObjectSerializer {
       }
       return new NativeRayObject(bytes, OBJECT_METADATA_TYPE_RAW);
     } else if (object instanceof RayTaskException) {
-      byte[] serializedBytes = Serializer.encode(object).getLeft();
+      RayTaskException taskException = (RayTaskException) object;
+      byte[] serializedBytes = Serializer.encode(taskException.toBytes()).getLeft();
+      // serializedBytes is MessagePack serialized bytes
+      // taskException.toBytes() is protobuf serialized bytes
+      // Only OBJECT_METADATA_TYPE_RAW is raw bytes,
+      // any other type should be the MessagePack serialized bytes.
       return new NativeRayObject(serializedBytes, TASK_EXECUTION_EXCEPTION_META);
     } else {
       try {

java/runtime/src/main/java/io/ray/runtime/object/ObjectStore.java

@@ -3,10 +3,10 @@ package io.ray.runtime.object;
 import com.google.common.base.Preconditions;
 import io.ray.api.ObjectRef;
 import io.ray.api.WaitResult;
-import io.ray.api.exception.RayException;
 import io.ray.api.id.ObjectId;
 import io.ray.api.id.UniqueId;
 import io.ray.runtime.context.WorkerContext;
+import io.ray.runtime.exception.RayException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -160,8 +160,7 @@ public abstract class ObjectStore {
    * Delete a list of objects from the object store.
    *
    * @param objectIds IDs of the objects to delete.
-   * @param localOnly Whether only delete the objects in local node, or all nodes in the
-   *     cluster.
+   * @param localOnly Whether only delete the objects in local node, or all nodes in the cluster.
    * @param deleteCreatingTasks Whether also delete the tasks that created these objects.
    */
   public abstract void delete(List<ObjectId> objectIds, boolean localOnly,
@@ -169,6 +168,7 @@ public abstract class ObjectStore {
 
   /**
    * Increase the local reference count for this object ID.
+   *
    * @param workerId The ID of the worker to increase on.
    * @param objectId The object ID to increase the reference count for.
    */
@@ -176,6 +176,7 @@ public abstract class ObjectStore {
 
   /**
    * Decrease the reference count for this object ID.
+   *
    * @param workerId The ID of the worker to decrease on.
    * @param objectId The object ID to decrease the reference count for.
    */

java/runtime/src/main/java/io/ray/runtime/task/TaskExecutor.java

@@ -1,12 +1,12 @@
 package io.ray.runtime.task;
 
 import com.google.common.base.Preconditions;
-import io.ray.api.exception.RayTaskException;
 import io.ray.api.id.ActorId;
 import io.ray.api.id.JobId;
 import io.ray.api.id.TaskId;
 import io.ray.api.id.UniqueId;
 import io.ray.runtime.RayRuntimeInternal;
+import io.ray.runtime.exception.RayTaskException;
 import io.ray.runtime.functionmanager.JavaFunctionDescriptor;
 import io.ray.runtime.functionmanager.RayFunction;
 import io.ray.runtime.generated.Common.TaskType;