[GCS] Fix detached actor with empty name (#9283)

src/ray/gcs/gcs_server/gcs_actor_manager.cc

@@ -69,11 +69,7 @@ ActorID GcsActor::GetActorID() const {
 
 bool GcsActor::IsDetached() const { return actor_table_data_.is_detached(); }
 
-std::string GcsActor::GetName() const {
-  RAY_CHECK(actor_table_data_.is_detached())
-      << "Actor names are only valid for detached actors.";
-  return actor_table_data_.name();
-}
+std::string GcsActor::GetName() const { return actor_table_data_.name(); }
 
 TaskSpecification GcsActor::GetCreationTaskSpecification() const {
   const auto &task_spec = actor_table_data_.task_spec();
@@ -390,7 +386,7 @@ Status GcsActorManager::RegisterActor(
   }
 
   auto actor = std::make_shared<GcsActor>(request);
-  if (actor->IsDetached()) {
+  if (!actor->GetName().empty()) {
     auto it = named_actors_.find(actor->GetName());
     if (it == named_actors_.end()) {
       named_actors_.emplace(actor->GetName(), actor->GetActorID());
@@ -661,8 +657,8 @@ void GcsActorManager::ReconstructActor(const ActorID &actor_id, bool need_resche
         }));
     gcs_actor_scheduler_->Schedule(actor);
   } else {
-    // For detached actors, make sure to remove its name.
-    if (actor->IsDetached()) {
+    // Remove actor from `named_actors_` if its name is not empty.
+    if (!actor->GetName().empty()) {
       auto it = named_actors_.find(actor->GetName());
       if (it != named_actors_.end()) {
         RAY_CHECK(it->second == actor->GetActorID());
@@ -757,7 +753,7 @@ void GcsActorManager::LoadInitialData(const EmptyCallback &done) {
         auto actor = std::make_shared<GcsActor>(item.second);
         registered_actors_.emplace(item.first, actor);
 
-        if (actor->IsDetached()) {
+        if (!actor->GetName().empty()) {
           named_actors_.emplace(actor->GetName(), actor->GetActorID());
         }
 

src/ray/gcs/gcs_server/gcs_actor_manager.h

@@ -275,7 +275,7 @@ class GcsActorManager : public rpc::ActorInfoHandler {
   /// All registered actors (pending actors are also included).
   /// TODO(swang): Use unique_ptr instead of shared_ptr.
   absl::flat_hash_map<ActorID, std::shared_ptr<GcsActor>> registered_actors_;
-  /// Maps detached actor names to their actor ID for lookups by name.
+  /// Maps actor names to their actor ID for lookups by name.
   absl::flat_hash_map<std::string, ActorID> named_actors_;
   /// The pending actors which will not be scheduled until there's a resource change.
   std::vector<std::shared_ptr<GcsActor>> pending_actors_;

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

@@ -393,6 +393,30 @@ TEST_F(GcsActorManagerTest, TestDetachedActorRestartWhenCreatorDead) {
   ASSERT_EQ(actor->GetState(), rpc::ActorTableData::ALIVE);
 }
 
+TEST_F(GcsActorManagerTest, TestActorWithEmptyName) {
+  auto job_id = JobID::FromInt(1);
+
+  // Gen `CreateActorRequest` with an empty name.
+  // (name,actor_id) => ("", actor_id_1)
+  auto request1 =
+      Mocker::GenCreateActorRequest(job_id, 0, /*is_detached=*/true, /*name=*/"");
+  Status status = gcs_actor_manager_->RegisterActor(
+      request1, [](std::shared_ptr<gcs::GcsActor> actor) {});
+  // Ensure successful registration.
+  ASSERT_TRUE(status.ok());
+  // Make sure actor who empty name is not treated as a named actor.
+  ASSERT_TRUE(gcs_actor_manager_->GetActorIDByName("").IsNil());
+
+  // Gen another `CreateActorRequest` with an empty name.
+  // (name,actor_id) => ("", actor_id_2)
+  auto request2 =
+      Mocker::GenCreateActorRequest(job_id, 0, /*is_detached=*/true, /*name=*/"");
+  status = gcs_actor_manager_->RegisterActor(request2,
+                                             [](std::shared_ptr<gcs::GcsActor> actor) {});
+  // Ensure successful registration.
+  ASSERT_TRUE(status.ok());
+}
+
 TEST_F(GcsActorManagerTest, TestNamedActors) {
   auto job_id_1 = JobID::FromInt(1);
   auto job_id_2 = JobID::FromInt(2);