[Placement Group] Fix stress tests to pass when actors are scheduled. (#11151)

* Fix stress tests to pass when actors are created.

* Addressed code review.

src/ray/gcs/gcs_server/gcs_placement_group_manager.cc

@@ -294,7 +294,8 @@ void GcsPlacementGroupManager::RemovePlacementGroup(
         // that the creation of placement group has failed.
         auto it = placement_group_to_register_callback_.find(placement_group_id);
         if (it != placement_group_to_register_callback_.end()) {
-          it->second(Status::NotFound("Placement group is removed before it is created"));
+          it->second(
+              Status::NotFound("Placement group is removed before it is created."));
           placement_group_to_register_callback_.erase(it);
         }
         on_placement_group_removed(status);

src/ray/gcs/gcs_server/gcs_placement_group_scheduler.cc

@@ -252,8 +252,6 @@ void GcsPlacementGroupScheduler::DestroyPlacementGroupBundleResourcesIfExists(
   // Check if we can find committed bundle locations.
   const auto &maybe_bundle_locations =
       committed_bundle_location_index_.GetBundleLocations(placement_group_id);
-  // If bundle location has been already removed, it means bundles
-  // are already destroyed. Do nothing.
   if (maybe_bundle_locations.has_value()) {
     committed_bundle_locations = maybe_bundle_locations.value();
   }
@@ -267,6 +265,8 @@ void GcsPlacementGroupScheduler::DestroyPlacementGroupBundleResourcesIfExists(
   }
 
   // Cancel all resource reservation.
+  RAY_LOG(INFO) << "Cancelling all bundles of a placement group, id is "
+                << placement_group_id;
   for (const auto &iter : *(committed_bundle_locations)) {
     auto &bundle_spec = iter.second.second;
     auto &node_id = iter.second.first;
@@ -296,6 +296,7 @@ void GcsPlacementGroupScheduler::PrepareResources(
     callback(Status::NotFound("Node is already dead."));
     return;
   }
+
   const auto lease_client = GetLeaseClientFromNode(node.value());
   const auto node_id = NodeID::FromBinary(node.value()->node_id());
   RAY_LOG(DEBUG) << "Preparing resource from node " << node_id
@@ -323,6 +324,7 @@ void GcsPlacementGroupScheduler::CommitResources(
   RAY_CHECK(node.has_value());
   const auto lease_client = GetLeaseClientFromNode(node.value());
   const auto node_id = NodeID::FromBinary(node.value()->node_id());
+
   RAY_LOG(DEBUG) << "Committing resource to a node " << node_id
                  << " for a bundle: " << bundle->DebugString();
   lease_client->CommitBundleResources(
@@ -353,6 +355,7 @@ void GcsPlacementGroupScheduler::CancelResourceReserve(
   RAY_LOG(DEBUG) << "Cancelling the resource reserved for bundle: "
                  << bundle_spec->DebugString() << " at node " << node_id;
   const auto return_client = GetLeaseClientFromNode(node.value());
+
   return_client->CancelResourceReserve(
       *bundle_spec, [bundle_spec, node_id](const Status &status,
                                            const rpc::CancelResourceReserveReply &reply) {

src/ray/raylet/node_manager.cc

@@ -1842,11 +1842,10 @@ void NodeManager::HandleCancelResourceReserve(
   }
 
   // Return bundle resources.
-  if (ReturnBundleResources(bundle_spec)) {
-    // Call task dispatch to assign work to the released resources.
-    TryLocalInfeasibleTaskScheduling();
-    DispatchTasks(local_queues_.GetReadyTasksByClass());
-  }
+  ReturnBundleResources(bundle_spec);
+  TryLocalInfeasibleTaskScheduling();
+  DispatchTasks(local_queues_.GetReadyTasksByClass());
+
   send_reply_callback(Status::OK(), nullptr, nullptr);
 }
 
@@ -2158,10 +2157,10 @@ void NodeManager::ScheduleTasks(
           << task.GetTaskSpecification().GetRequiredResources().ToString()
           << " for execution and "
           << task.GetTaskSpecification().GetRequiredPlacementResources().ToString()
-          << " for placement, however there are no nodes in the cluster that can "
-          << "provide the requested resources. To resolve this issue, consider "
-          << "reducing the resource requests of this task or add nodes that "
-          << "can fit the task.";
+          << " for placement, however the cluster currently cannot provide the requested "
+             "resources. The required resources may be added as autoscaling takes place "
+             "or placement groups are scheduled. Otherwise, consider reducing the "
+             "resource requirements of the task.";
       auto error_data_ptr =
           gcs::CreateErrorTableData(type, error_message.str(), current_time_ms(),
                                     task.GetTaskSpecification().JobId());
@@ -3258,8 +3257,11 @@ void NodeManager::HandleGetNodeStats(const rpc::GetNodeStatsRequest &node_stats_
           if (status.ok()) {
             worker_stats->mutable_core_worker_stats()->MergeFrom(r.core_worker_stats());
           } else {
-            RAY_LOG(ERROR) << "Failed to send get core worker stats request: "
-                           << status.ToString();
+            RAY_LOG(WARNING) << "Failed to send get core worker stats request, "
+                             << "worker id is " << worker->WorkerId() << ", status is "
+                             << status.ToString()
+                             << ". This is likely since the worker has died before the "
+                                "request was sent.";
             worker_stats->set_fetch_error(status.ToString());
           }
           if (reply->num_workers() == all_workers.size()) {