Move the error check to core_worker.cc

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>

python/ray/_raylet.pyx

@@ -3831,11 +3831,7 @@ cdef class CoreWorker:
             c_vector[CObjectID] free_ids = ObjectRefsToVector(object_refs)
 
         with nogil:
-            status = CCoreWorkerProcess.GetCoreWorker().Delete(free_ids, local_only)
-            if status.IsIOError():
-                check_status(CRayStatus.UnexpectedSystemExit(status.ToString()))
-            else:
-                check_status(status)
+            check_status(CCoreWorkerProcess.GetCoreWorker().Delete(free_ids, local_only))
 
     def get_local_ongoing_lineage_reconstruction_tasks(self):
         cdef:

src/ray/core_worker/core_worker.cc

@@ -1868,7 +1868,12 @@ Status CoreWorker::Delete(const std::vector<ObjectID> &object_ids, bool local_on
     }
   }
   // Also try to delete all objects locally.
-  return DeleteImpl(object_ids, local_only);
+  Status status = DeleteImpl(object_ids, local_only);
+  if (status.IsIOError()) {
+    return Status::UnexpectedSystemExit(status.ToString());
+  } else {
+    return status;
+  }
 }
 
 Status CoreWorker::GetLocationFromOwner(

src/ray/core_worker/core_worker.h

@@ -795,6 +795,11 @@ class CoreWorker : public rpc::CoreWorkerServiceHandler {
   /// This calls DeleteImpl() locally for objects we own, and DeleteImpl() remotely
   /// for objects we do not own.
   ///
+  /// If IOError is returned from DeleteImpl() when deleting objects locally, we will
+  /// return an UnexpectedSystemExit status instead. This is to make sure the tasks
+  /// that calls this function in application code can properly retry when hitting the 
+  /// IOError.
+  ///
   /// \param[in] object_ids IDs of the objects to delete.
   /// \param[in] local_only Whether only delete the objects in local node, or all nodes in
   /// the cluster.