Remove actor table info from storage when a driver exits (#8761)

* delete contents of table related to specified job when the job is dead

* check status

* implement GetByJobId in gcs table storage

* add test case

* add test case

* fix test cases

* expose MGET and make match_pattern only related with SCAN

* add test case for table storage

* delete checkpoint

* make MGetValues static

* add most test case

* add object test case

* avoid accessing to storage when get matched object ids per job id

* rename job info handler

* use listener to sense job finished

* clear actor state

* add comments, remove actions in task handler

* let raylet do object cleaning. only remove non-detached actors

* only remove informations of non-detached actor

* remove unused methods

src/ray/gcs/gcs_server/gcs_actor_manager.cc

@@ -764,5 +764,52 @@ void GcsActorManager::LoadInitialData(const EmptyCallback &done) {
   RAY_CHECK_OK(gcs_table_storage_->ActorTable().GetAll(callback));
 }
 
+void GcsActorManager::OnJobFinished(const JobID &job_id) {
+  auto on_done = [this,
+                  job_id](const std::unordered_map<ActorID, ActorTableData> &result) {
+    if (!result.empty()) {
+      std::vector<ActorID> non_detached_actors;
+      std::unordered_set<ActorID> non_detached_actors_set;
+      for (auto &item : result) {
+        if (!item.second.is_detached()) {
+          non_detached_actors.push_back(item.first);
+          non_detached_actors_set.insert(item.first);
+        }
+      }
+      RAY_CHECK_OK(
+          gcs_table_storage_->ActorTable().BatchDelete(non_detached_actors, nullptr));
+
+      // Get checkpoint id first from checkpoint id table and delete all checkpoints
+      // related to this job
+      RAY_CHECK_OK(gcs_table_storage_->ActorCheckpointIdTable().GetByJobId(
+          job_id, [this, non_detached_actors_set](
+                      const std::unordered_map<ActorID, ActorCheckpointIdData> &result) {
+            if (!result.empty()) {
+              std::vector<ActorID> checkpoints;
+              std::vector<ActorCheckpointID> checkpoint_ids;
+              for (auto &item : result) {
+                if (non_detached_actors_set.find(item.first) !=
+                    non_detached_actors_set.end()) {
+                  checkpoints.push_back(item.first);
+                  for (auto &id : item.second.checkpoint_ids()) {
+                    checkpoint_ids.push_back(ActorCheckpointID::FromBinary(id));
+                  }
+                }
+              }
+
+              RAY_CHECK_OK(gcs_table_storage_->ActorCheckpointIdTable().BatchDelete(
+                  checkpoints, nullptr));
+              RAY_CHECK_OK(gcs_table_storage_->ActorCheckpointTable().BatchDelete(
+                  checkpoint_ids, nullptr));
+            }
+          }));
+    }
+  };
+
+  // Only non-detached actors should be deleted. We get all actors of this job and to the
+  // filtering.
+  RAY_CHECK_OK(gcs_table_storage_->ActorTable().GetByJobId(job_id, on_done));
+}
+
 }  // namespace gcs
 }  // namespace ray

src/ray/gcs/gcs_server/gcs_actor_manager.h

@@ -222,6 +222,12 @@ class GcsActorManager : public rpc::ActorInfoHandler {
   /// \param done Callback that will be called when load is complete.
   void LoadInitialData(const EmptyCallback &done);
 
+  /// Delete non-detached actor information from durable storage once the associated job
+  /// finishes.
+  ///
+  /// \param job_id The id of finished job.
+  void OnJobFinished(const JobID &job_id);
+
  private:
   /// A data structure representing an actor's owner.
   struct Owner {

src/ray/gcs/gcs_server/gcs_server.cc

@@ -61,7 +61,17 @@ void GcsServer::Start() {
   InitGcsActorManager();
 
   // Register rpc service.
-  job_info_handler_ = InitJobInfoHandler();
+  gcs_object_manager_ = InitObjectManager();
+  object_info_service_.reset(
+      new rpc::ObjectInfoGrpcService(main_service_, *gcs_object_manager_));
+  rpc_server_.RegisterService(*object_info_service_);
+
+  task_info_handler_ = InitTaskInfoHandler();
+  task_info_service_.reset(
+      new rpc::TaskInfoGrpcService(main_service_, *task_info_handler_));
+  rpc_server_.RegisterService(*task_info_service_);
+
+  InitJobInfoHandler();
   job_info_service_.reset(new rpc::JobInfoGrpcService(main_service_, *job_info_handler_));
   rpc_server_.RegisterService(*job_info_service_);
 
@@ -73,16 +83,6 @@ void GcsServer::Start() {
       new rpc::NodeInfoGrpcService(main_service_, *gcs_node_manager_));
   rpc_server_.RegisterService(*node_info_service_);
 
-  gcs_object_manager_ = InitObjectManager();
-  object_info_service_.reset(
-      new rpc::ObjectInfoGrpcService(main_service_, *gcs_object_manager_));
-  rpc_server_.RegisterService(*object_info_service_);
-
-  task_info_handler_ = InitTaskInfoHandler();
-  task_info_service_.reset(
-      new rpc::TaskInfoGrpcService(main_service_, *task_info_handler_));
-  rpc_server_.RegisterService(*task_info_service_);
-
   stats_handler_ = InitStatsHandler();
   stats_service_.reset(new rpc::StatsGrpcService(main_service_, *stats_handler_));
   rpc_server_.RegisterService(*stats_service_);
@@ -198,9 +198,12 @@ void GcsServer::InitGcsActorManager() {
   RAY_CHECK_OK(gcs_pub_sub_->SubscribeAll(WORKER_FAILURE_CHANNEL, on_subscribe, nullptr));
 }
 
-std::unique_ptr<rpc::JobInfoHandler> GcsServer::InitJobInfoHandler() {
-  return std::unique_ptr<rpc::DefaultJobInfoHandler>(
-      new rpc::DefaultJobInfoHandler(gcs_table_storage_, gcs_pub_sub_));
+void GcsServer::InitJobInfoHandler() {
+  job_info_handler_ = std::unique_ptr<rpc::GcsJobInfoHandler>(
+      new rpc::GcsJobInfoHandler(gcs_table_storage_, gcs_pub_sub_));
+  job_info_handler_->AddJobFinishedListener([this](std::shared_ptr<JobID> job_id) {
+    gcs_actor_manager_->OnJobFinished(*job_id);
+  });
 }
 
 std::unique_ptr<GcsObjectManager> GcsServer::InitObjectManager() {

src/ray/gcs/gcs_server/gcs_server.h

@@ -81,7 +81,7 @@ class GcsServer {
   virtual void InitGcsActorManager();
 
   /// The job info handler
-  virtual std::unique_ptr<rpc::JobInfoHandler> InitJobInfoHandler();
+  virtual void InitJobInfoHandler();
 
   /// The object manager
   virtual std::unique_ptr<GcsObjectManager> InitObjectManager();

src/ray/gcs/gcs_server/job_info_handler_impl.cc

@@ -18,9 +18,9 @@
 namespace ray {
 namespace rpc {
 
-void DefaultJobInfoHandler::HandleAddJob(const rpc::AddJobRequest &request,
-                                         rpc::AddJobReply *reply,
-                                         rpc::SendReplyCallback send_reply_callback) {
+void GcsJobInfoHandler::HandleAddJob(const rpc::AddJobRequest &request,
+                                     rpc::AddJobReply *reply,
+                                     rpc::SendReplyCallback send_reply_callback) {
   JobID job_id = JobID::FromBinary(request.data().job_id());
   RAY_LOG(INFO) << "Adding job, job id = " << job_id
                 << ", driver pid = " << request.data().driver_pid();
@@ -41,7 +41,7 @@ void DefaultJobInfoHandler::HandleAddJob(const rpc::AddJobRequest &request,
   }
 }
 
-void DefaultJobInfoHandler::HandleMarkJobFinished(
+void GcsJobInfoHandler::HandleMarkJobFinished(
     const rpc::MarkJobFinishedRequest &request, rpc::MarkJobFinishedReply *reply,
     rpc::SendReplyCallback send_reply_callback) {
   JobID job_id = JobID::FromBinary(request.job_id());
@@ -55,6 +55,7 @@ void DefaultJobInfoHandler::HandleMarkJobFinished(
     } else {
       RAY_CHECK_OK(gcs_pub_sub_->Publish(JOB_CHANNEL, job_id.Binary(),
                                          job_table_data->SerializeAsString(), nullptr));
+      ClearJobInfos(job_id);
       RAY_LOG(INFO) << "Finished marking job state, job id = " << job_id;
     }
     GCS_RPC_SEND_REPLY(send_reply_callback, reply, status);
@@ -66,9 +67,25 @@ void DefaultJobInfoHandler::HandleMarkJobFinished(
   }
 }
 
-void DefaultJobInfoHandler::HandleGetAllJobInfo(
-    const rpc::GetAllJobInfoRequest &request, rpc::GetAllJobInfoReply *reply,
-    rpc::SendReplyCallback send_reply_callback) {
+void GcsJobInfoHandler::ClearJobInfos(const JobID &job_id) {
+  // Notify all listeners.
+  for (auto &listener : job_finished_listeners_) {
+    listener(std::make_shared<JobID>(job_id));
+  }
+}
+
+/// Add listener to monitor the add action of nodes.
+///
+/// \param listener The handler which process the add of nodes.
+void GcsJobInfoHandler::AddJobFinishedListener(
+    std::function<void(std::shared_ptr<JobID>)> listener) {
+  RAY_CHECK(listener);
+  job_finished_listeners_.emplace_back(std::move(listener));
+}
+
+void GcsJobInfoHandler::HandleGetAllJobInfo(const rpc::GetAllJobInfoRequest &request,
+                                            rpc::GetAllJobInfoReply *reply,
+                                            rpc::SendReplyCallback send_reply_callback) {
   RAY_LOG(INFO) << "Getting all job info.";
   auto on_done = [reply, send_reply_callback](
                      const std::unordered_map<JobID, JobTableData> &result) {

src/ray/gcs/gcs_server/job_info_handler_impl.h

@@ -14,6 +14,7 @@
 
 #pragma once
 
+#include "gcs_object_manager.h"
 #include "gcs_table_storage.h"
 #include "ray/gcs/pubsub/gcs_pub_sub.h"
 #include "ray/gcs/redis_gcs_client.h"
@@ -23,10 +24,10 @@ namespace ray {
 namespace rpc {
 
 /// This implementation class of `JobInfoHandler`.
-class DefaultJobInfoHandler : public rpc::JobInfoHandler {
+class GcsJobInfoHandler : public rpc::JobInfoHandler {
  public:
-  explicit DefaultJobInfoHandler(std::shared_ptr<gcs::GcsTableStorage> gcs_table_storage,
-                                 std::shared_ptr<gcs::GcsPubSub> gcs_pub_sub)
+  explicit GcsJobInfoHandler(std::shared_ptr<gcs::GcsTableStorage> gcs_table_storage,
+                             std::shared_ptr<gcs::GcsPubSub> gcs_pub_sub)
       : gcs_table_storage_(std::move(gcs_table_storage)),
         gcs_pub_sub_(std::move(gcs_pub_sub)) {}
 
@@ -40,9 +41,17 @@ class DefaultJobInfoHandler : public rpc::JobInfoHandler {
   void HandleGetAllJobInfo(const GetAllJobInfoRequest &request, GetAllJobInfoReply *reply,
                            SendReplyCallback send_reply_callback) override;
 
+  void AddJobFinishedListener(
+      std::function<void(std::shared_ptr<JobID>)> listener) override;
+
  private:
   std::shared_ptr<gcs::GcsTableStorage> gcs_table_storage_;
   std::shared_ptr<gcs::GcsPubSub> gcs_pub_sub_;
+
+  /// Listeners which monitors the finish of jobs.
+  std::vector<std::function<void(std::shared_ptr<JobID>)>> job_finished_listeners_;
+
+  void ClearJobInfos(const JobID &job_id);
 };
 
 }  // namespace rpc

src/ray/gcs/gcs_server/test/gcs_server_rpc_test.cc

@@ -102,20 +102,24 @@ class GcsServerTest : public ::testing::Test {
     return WaitReady(promise.get_future(), timeout_ms_);
   }
 
-  rpc::ActorTableData GetActorInfo(const std::string &actor_id) {
+  boost::optional<rpc::ActorTableData> GetActorInfo(const std::string &actor_id) {
     rpc::GetActorInfoRequest request;
     request.set_actor_id(actor_id);
-    rpc::ActorTableData actor_table_data;
+    boost::optional<rpc::ActorTableData> actor_table_data_opt;
     std::promise<bool> promise;
     client_->GetActorInfo(
-        request, [&actor_table_data, &promise](const Status &status,
-                                               const rpc::GetActorInfoReply &reply) {
+        request, [&actor_table_data_opt, &promise](const Status &status,
+                                                   const rpc::GetActorInfoReply &reply) {
           RAY_CHECK_OK(status);
-          actor_table_data.CopyFrom(reply.actor_table_data());
+          if (reply.has_actor_table_data()) {
+            actor_table_data_opt = reply.actor_table_data();
+          } else {
+            actor_table_data_opt = boost::none;
+          }
           promise.set_value(true);
         });
     EXPECT_TRUE(WaitReady(promise.get_future(), timeout_ms_));
-    return actor_table_data;
+    return actor_table_data_opt;
   }
 
   bool AddActorCheckpoint(const rpc::AddActorCheckpointRequest &request) {
@@ -129,38 +133,47 @@ class GcsServerTest : public ::testing::Test {
     return WaitReady(promise.get_future(), timeout_ms_);
   }
 
-  rpc::ActorCheckpointData GetActorCheckpoint(const std::string &actor_id,
-                                              const std::string &checkpoint_id) {
+  boost::optional<rpc::ActorCheckpointData> GetActorCheckpoint(
+      const std::string &actor_id, const std::string &checkpoint_id) {
     rpc::GetActorCheckpointRequest request;
     request.set_actor_id(actor_id);
     request.set_checkpoint_id(checkpoint_id);
-    rpc::ActorCheckpointData checkpoint_data;
+    boost::optional<rpc::ActorCheckpointData> checkpoint_data_opt;
     std::promise<bool> promise;
     client_->GetActorCheckpoint(
-        request, [&checkpoint_data, &promise](const Status &status,
-                                              const rpc::GetActorCheckpointReply &reply) {
+        request, [&checkpoint_data_opt, &promise](
+                     const Status &status, const rpc::GetActorCheckpointReply &reply) {
           RAY_CHECK_OK(status);
-          checkpoint_data.CopyFrom(reply.checkpoint_data());
+          if (reply.has_checkpoint_data()) {
+            checkpoint_data_opt = reply.checkpoint_data();
+          } else {
+            checkpoint_data_opt = boost::none;
+          }
           promise.set_value(true);
         });
     EXPECT_TRUE(WaitReady(promise.get_future(), timeout_ms_));
-    return checkpoint_data;
+    return checkpoint_data_opt;
   }
 
-  rpc::ActorCheckpointIdData GetActorCheckpointID(const std::string &actor_id) {
+  boost::optional<rpc::ActorCheckpointIdData> GetActorCheckpointID(
+      const std::string &actor_id) {
     rpc::GetActorCheckpointIDRequest request;
     request.set_actor_id(actor_id);
-    rpc::ActorCheckpointIdData checkpoint_id_data;
+    boost::optional<rpc::ActorCheckpointIdData> checkpoint_id_data_opt;
     std::promise<bool> promise;
     client_->GetActorCheckpointID(
-        request, [&checkpoint_id_data, &promise](
+        request, [&checkpoint_id_data_opt, &promise](
                      const Status &status, const rpc::GetActorCheckpointIDReply &reply) {
           RAY_CHECK_OK(status);
-          checkpoint_id_data.CopyFrom(reply.checkpoint_id_data());
+          if (reply.has_checkpoint_id_data()) {
+            checkpoint_id_data_opt = reply.checkpoint_id_data();
+          } else {
+            checkpoint_id_data_opt = boost::none;
+          }
           promise.set_value(true);
         });
     EXPECT_TRUE(WaitReady(promise.get_future(), timeout_ms_));
-    return checkpoint_id_data;
+    return checkpoint_id_data_opt;
   }
 
   bool RegisterNode(const rpc::RegisterNodeRequest &request) {
@@ -310,7 +323,9 @@ class GcsServerTest : public ::testing::Test {
     client_->GetTask(request, [&task_data, &promise](const Status &status,
                                                      const rpc::GetTaskReply &reply) {
       if (status.ok()) {
-        task_data.CopyFrom(reply.task_data());
+        if (reply.has_task_data()) {
+          task_data.CopyFrom(reply.task_data());
+        }
       }
       promise.set_value(true);
     });
@@ -409,8 +424,9 @@ TEST_F(GcsServerTest, TestActorInfo) {
   rpc::RegisterActorInfoRequest register_actor_info_request;
   register_actor_info_request.mutable_actor_table_data()->CopyFrom(*actor_table_data);
   ASSERT_TRUE(RegisterActorInfo(register_actor_info_request));
-  rpc::ActorTableData result = GetActorInfo(actor_table_data->actor_id());
-  ASSERT_TRUE(result.state() ==
+  boost::optional<rpc::ActorTableData> result =
+      GetActorInfo(actor_table_data->actor_id());
+  ASSERT_TRUE(result->state() ==
               rpc::ActorTableData_ActorState::ActorTableData_ActorState_ALIVE);
 
   // Update actor state
@@ -421,7 +437,7 @@ TEST_F(GcsServerTest, TestActorInfo) {
   update_actor_info_request.mutable_actor_table_data()->CopyFrom(*actor_table_data);
   ASSERT_TRUE(UpdateActorInfo(update_actor_info_request));
   result = GetActorInfo(actor_table_data->actor_id());
-  ASSERT_TRUE(result.state() ==
+  ASSERT_TRUE(result->state() ==
               rpc::ActorTableData_ActorState::ActorTableData_ActorState_DEAD);
 
   // Add actor checkpoint
@@ -434,14 +450,14 @@ TEST_F(GcsServerTest, TestActorInfo) {
   rpc::AddActorCheckpointRequest add_actor_checkpoint_request;
   add_actor_checkpoint_request.mutable_checkpoint_data()->CopyFrom(checkpoint);
   ASSERT_TRUE(AddActorCheckpoint(add_actor_checkpoint_request));
-  rpc::ActorCheckpointData checkpoint_result =
+  boost::optional<rpc::ActorCheckpointData> checkpoint_result =
       GetActorCheckpoint(actor_table_data->actor_id(), checkpoint_id.Binary());
-  ASSERT_TRUE(checkpoint_result.actor_id() == actor_table_data->actor_id());
-  ASSERT_TRUE(checkpoint_result.checkpoint_id() == checkpoint_id.Binary());
-  rpc::ActorCheckpointIdData checkpoint_id_result =
+  ASSERT_TRUE(checkpoint_result->actor_id() == actor_table_data->actor_id());
+  ASSERT_TRUE(checkpoint_result->checkpoint_id() == checkpoint_id.Binary());
+  boost::optional<rpc::ActorCheckpointIdData> checkpoint_id_result =
       GetActorCheckpointID(actor_table_data->actor_id());
-  ASSERT_TRUE(checkpoint_id_result.actor_id() == actor_table_data->actor_id());
-  ASSERT_TRUE(checkpoint_id_result.checkpoint_ids_size() == 1);
+  ASSERT_TRUE(checkpoint_id_result->actor_id() == actor_table_data->actor_id());
+  ASSERT_TRUE(checkpoint_id_result->checkpoint_ids_size() == 1);
 }
 
 TEST_F(GcsServerTest, TestJobInfo) {
@@ -460,6 +476,120 @@ TEST_F(GcsServerTest, TestJobInfo) {
   ASSERT_TRUE(MarkJobFinished(mark_job_finished_request));
 }
 
+TEST_F(GcsServerTest, TestJobGarbageCollection) {
+  // Create job_table_data
+  JobID job_id = JobID::FromInt(1);
+  auto job_table_data = Mocker::GenJobTableData(job_id);
+
+  // Add job
+  rpc::AddJobRequest add_job_request;
+  add_job_request.mutable_data()->CopyFrom(*job_table_data);
+  ASSERT_TRUE(AddJob(add_job_request));
+
+  // Register actor for job
+  auto actor_table_data = Mocker::GenActorTableData(job_id);
+  rpc::RegisterActorInfoRequest register_actor_info_request;
+  register_actor_info_request.mutable_actor_table_data()->CopyFrom(*actor_table_data);
+  ASSERT_TRUE(RegisterActorInfo(register_actor_info_request));
+  boost::optional<rpc::ActorTableData> actor_result =
+      GetActorInfo(actor_table_data->actor_id());
+  ASSERT_TRUE(actor_result->state() ==
+              rpc::ActorTableData_ActorState::ActorTableData_ActorState_ALIVE);
+
+  // Add actor checkpoint
+  ActorCheckpointID checkpoint_id = ActorCheckpointID::FromRandom();
+  rpc::ActorCheckpointData checkpoint;
+  checkpoint.set_actor_id(actor_table_data->actor_id());
+  checkpoint.set_checkpoint_id(checkpoint_id.Binary());
+  checkpoint.set_execution_dependency(checkpoint_id.Binary());
+
+  rpc::AddActorCheckpointRequest add_actor_checkpoint_request;
+  add_actor_checkpoint_request.mutable_checkpoint_data()->CopyFrom(checkpoint);
+  ASSERT_TRUE(AddActorCheckpoint(add_actor_checkpoint_request));
+  boost::optional<rpc::ActorCheckpointData> checkpoint_result =
+      GetActorCheckpoint(actor_table_data->actor_id(), checkpoint_id.Binary());
+  ASSERT_TRUE(checkpoint_result->actor_id() == actor_table_data->actor_id());
+  ASSERT_TRUE(checkpoint_result->checkpoint_id() == checkpoint_id.Binary());
+  boost::optional<rpc::ActorCheckpointIdData> checkpoint_id_result =
+      GetActorCheckpointID(actor_table_data->actor_id());
+  ASSERT_TRUE(checkpoint_id_result->actor_id() == actor_table_data->actor_id());
+  ASSERT_TRUE(checkpoint_id_result->checkpoint_ids_size() == 1);
+
+  // Register detached actor for job
+  auto detached_actor_table_data = Mocker::GenActorTableData(job_id);
+  detached_actor_table_data->set_is_detached(true);
+  rpc::RegisterActorInfoRequest register_detached_actor_info_request;
+  register_detached_actor_info_request.mutable_actor_table_data()->CopyFrom(
+      *detached_actor_table_data);
+  ASSERT_TRUE(RegisterActorInfo(register_detached_actor_info_request));
+  boost::optional<rpc::ActorTableData> detached_actor_result =
+      GetActorInfo(detached_actor_table_data->actor_id());
+  ASSERT_TRUE(detached_actor_result->state() ==
+              rpc::ActorTableData_ActorState::ActorTableData_ActorState_ALIVE);
+
+  // Add checkpoint for detached actor
+  ActorCheckpointID detached_checkpoint_id = ActorCheckpointID::FromRandom();
+  rpc::ActorCheckpointData detached_checkpoint;
+  detached_checkpoint.set_actor_id(detached_actor_table_data->actor_id());
+  detached_checkpoint.set_checkpoint_id(detached_checkpoint_id.Binary());
+  detached_checkpoint.set_execution_dependency(detached_checkpoint_id.Binary());
+
+  rpc::AddActorCheckpointRequest add_detached_actor_checkpoint_request;
+  add_detached_actor_checkpoint_request.mutable_checkpoint_data()->CopyFrom(
+      detached_checkpoint);
+  ASSERT_TRUE(AddActorCheckpoint(add_detached_actor_checkpoint_request));
+  boost::optional<rpc::ActorCheckpointData> detached_checkpoint_result =
+      GetActorCheckpoint(detached_actor_table_data->actor_id(),
+                         detached_checkpoint_id.Binary());
+  ASSERT_TRUE(detached_checkpoint_result->actor_id() ==
+              detached_actor_table_data->actor_id());
+  ASSERT_TRUE(detached_checkpoint_result->checkpoint_id() ==
+              detached_checkpoint_id.Binary());
+  boost::optional<rpc::ActorCheckpointIdData> detached_checkpoint_id_result =
+      GetActorCheckpointID(detached_actor_table_data->actor_id());
+  ASSERT_TRUE(detached_checkpoint_id_result->actor_id() ==
+              detached_actor_table_data->actor_id());
+  ASSERT_TRUE(detached_checkpoint_id_result->checkpoint_ids_size() == 1);
+
+  // Mark job finished
+  rpc::MarkJobFinishedRequest mark_job_finished_request;
+  mark_job_finished_request.set_job_id(job_table_data->job_id());
+  ASSERT_TRUE(MarkJobFinished(mark_job_finished_request));
+
+  std::function<bool()> condition_func = [this, &actor_table_data]() -> bool {
+    return !GetActorInfo(actor_table_data->actor_id()).has_value();
+  };
+  ASSERT_TRUE(WaitForCondition(condition_func, 10 * 1000));
+
+  condition_func = [this, &actor_table_data, &checkpoint_id]() -> bool {
+    return !GetActorCheckpoint(actor_table_data->actor_id(), checkpoint_id.Binary())
+                .has_value();
+  };
+  ASSERT_TRUE(WaitForCondition(condition_func, 10 * 1000));
+
+  condition_func = [this, &actor_table_data]() -> bool {
+    return !GetActorCheckpointID(actor_table_data->actor_id()).has_value();
+  };
+  ASSERT_TRUE(WaitForCondition(condition_func, 10 * 1000));
+
+  detached_actor_result = GetActorInfo(detached_actor_table_data->actor_id());
+  ASSERT_TRUE(detached_actor_result->state() ==
+              rpc::ActorTableData_ActorState::ActorTableData_ActorState_ALIVE);
+
+  detached_checkpoint_result = GetActorCheckpoint(detached_actor_table_data->actor_id(),
+                                                  detached_checkpoint_id.Binary());
+  ASSERT_TRUE(detached_checkpoint_result->actor_id() ==
+              detached_actor_table_data->actor_id());
+  ASSERT_TRUE(detached_checkpoint_result->checkpoint_id() ==
+              detached_checkpoint_id.Binary());
+
+  detached_checkpoint_id_result =
+      GetActorCheckpointID(detached_actor_table_data->actor_id());
+  ASSERT_TRUE(detached_checkpoint_id_result->actor_id() ==
+              detached_actor_table_data->actor_id());
+  ASSERT_TRUE(detached_checkpoint_id_result->checkpoint_ids_size() == 1);
+}
+
 TEST_F(GcsServerTest, TestNodeInfo) {
   // Create gcs node info
   auto gcs_node_info = Mocker::GenNodeInfo();

src/ray/gcs/store_client/in_memory_store_client.cc

@@ -121,7 +121,11 @@ Status InMemoryStoreClient::AsyncDeleteByIndex(const std::string &table_name,
     }
     table->index_keys_.erase(iter);
   }
-  main_io_service_.post([callback]() { callback(Status::OK()); });
+  main_io_service_.post([callback]() {
+    if (callback) {
+      callback(Status::OK());
+    }
+  });
   return Status::OK();
 }
 

src/ray/gcs/store_client/redis_store_client.cc

@@ -169,7 +169,9 @@ Status RedisStoreClient::AsyncDeleteByIndex(const std::string &table_name,
       };
       RAY_CHECK_OK(AsyncBatchDelete(table_name, keys, batch_delete_callback));
     } else {
-      callback(status);
+      if (callback) {
+        callback(status);
+      }
     }
   };
 
@@ -203,7 +205,9 @@ Status RedisStoreClient::DeleteByKeys(const std::vector<std::string> &keys,
                             callback](const std::shared_ptr<CallbackReply> &reply) {
       ++(*finished_count);
       if (*finished_count == size) {
-        callback(Status::OK());
+        if (callback) {
+          callback(Status::OK());
+        }
       }
     };
     RAY_CHECK_OK(item.first->RunArgvAsync(item.second, delete_callback));

src/ray/rpc/gcs_server/gcs_rpc_server.h

@@ -63,6 +63,9 @@ class JobInfoGcsServiceHandler {
   virtual void HandleGetAllJobInfo(const GetAllJobInfoRequest &request,
                                    GetAllJobInfoReply *reply,
                                    SendReplyCallback send_reply_callback) = 0;
+
+  virtual void AddJobFinishedListener(
+      std::function<void(std::shared_ptr<JobID>)> listener) = 0;
 };
 
 /// The `GrpcService` for `JobInfoGcsService`.