ignore object exists error for memory store provider (#5607)

src/ray/common/status.h

@@ -74,6 +74,7 @@ enum class StatusCode : char {
   TypeError = 3,
   Invalid = 4,
   IOError = 5,
+  ObjectExists = 6,
   UnknownError = 9,
   NotImplemented = 10,
   RedisError = 11
@@ -132,6 +133,10 @@ class RAY_EXPORT Status {
     return Status(StatusCode::RedisError, msg);
   }
 
+  static Status ObjectExists(const std::string &msg) {
+    return Status(StatusCode::ObjectExists, msg);
+  }
+
   // Returns true iff the status indicates success.
   bool ok() const { return (state_ == NULL); }
 
@@ -143,6 +148,7 @@ class RAY_EXPORT Status {
   bool IsUnknownError() const { return code() == StatusCode::UnknownError; }
   bool IsNotImplemented() const { return code() == StatusCode::NotImplemented; }
   bool IsRedisError() const { return code() == StatusCode::RedisError; }
+  bool IsObjectExists() const { return code() == StatusCode::ObjectExists; }
 
   // Return a string representation of this status suitable for printing.
   // Returns the string "OK" for success.

src/ray/core_worker/store_provider/memory_store/memory_store.cc

@@ -111,7 +111,7 @@ Status CoreWorkerMemoryStore::Put(const ObjectID &object_id, const RayObject &ob
   std::unique_lock<std::mutex> lock(lock_);
   auto iter = objects_.find(object_id);
   if (iter != objects_.end()) {
-    return Status::KeyError("object already exists");
+    return Status::ObjectExists("object already exists in the memory store");
   }
 
   auto object_entry =

src/ray/core_worker/store_provider/memory_store_provider.cc

@@ -18,7 +18,13 @@ CoreWorkerMemoryStoreProvider::CoreWorkerMemoryStoreProvider(
 
 Status CoreWorkerMemoryStoreProvider::Put(const RayObject &object,
                                           const ObjectID &object_id) {
-  return store_->Put(object_id, object);
+  auto status = store_->Put(object_id, object);
+  if (status.IsObjectExists()) {
+    // Object already exists in store, treat it as ok.
+    return Status::OK();
+  } else {
+    return status;
+  }
 }
 
 Status CoreWorkerMemoryStoreProvider::Get(