[Placement Group]Fix CommitResources crash bug (#10951)

src/ray/gcs/gcs_server/gcs_placement_group_scheduler.cc

@@ -386,17 +386,23 @@ void GcsPlacementGroupScheduler::CommitAllBundles(
     const auto &node = gcs_node_manager_.GetNode(node_id);
     const auto &bundle = bundle_to_commit.second.second;
 
-    // TODO(sang) Handle the case nodes are dead.
-    CommitResources(
-        bundle, node,
-        [this, lease_status_tracker, bundle, node_id, schedule_failure_handler,
-         schedule_success_handler](const Status &status) {
-          lease_status_tracker->MarkCommitRequestReturned(node_id, bundle, status);
-          if (lease_status_tracker->AllCommitRequestReturned()) {
-            OnAllBundleCommitRequestReturned(
-                lease_status_tracker, schedule_failure_handler, schedule_success_handler);
-          }
-        });
+    auto commit_resources_callback = [this, lease_status_tracker, bundle, node_id,
+                                      schedule_failure_handler,
+                                      schedule_success_handler](const Status &status) {
+      lease_status_tracker->MarkCommitRequestReturned(node_id, bundle, status);
+      if (lease_status_tracker->AllCommitRequestReturned()) {
+        OnAllBundleCommitRequestReturned(lease_status_tracker, schedule_failure_handler,
+                                         schedule_success_handler);
+      }
+    };
+
+    if (node != nullptr) {
+      CommitResources(bundle, node, commit_resources_callback);
+    } else {
+      RAY_LOG(INFO) << "Failed to commit resources because the node is dead, node id = "
+                    << node_id;
+      commit_resources_callback(Status::Interrupted("Node is dead"));
+    }
   }
 }
 
@@ -682,7 +688,8 @@ bool LeaseStatusTracker::AllCommitRequestsSuccessful() const {
   // We don't check cancel state here because we shouldn't destroy bundles when
   // commit requests failed. Cancel state should be treated separately.
   return AllCommitRequestReturned() &&
-         preparing_bundle_locations_->size() == bundles_to_schedule_.size();
+         preparing_bundle_locations_->size() == bundles_to_schedule_.size() &&
+         uncommitted_bundle_locations_->empty();
 }
 
 const std::shared_ptr<GcsPlacementGroup> &LeaseStatusTracker::GetPlacementGroup() const {

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

@@ -616,6 +616,36 @@ TEST_F(GcsPlacementGroupSchedulerTest, TestBundleLocationIndex) {
   ASSERT_TRUE(bundle_location_index.GetBundleLocationsOnNode(node2).value()->size() == 1);
 }
 
+TEST_F(GcsPlacementGroupSchedulerTest, TestNodeDeadDuringCommitResources) {
+  auto node0 = Mocker::GenNodeInfo(0);
+  auto node1 = Mocker::GenNodeInfo(1);
+  AddNode(node0);
+  AddNode(node1);
+  ASSERT_EQ(2, gcs_node_manager_->GetAllAliveNodes().size());
+
+  auto create_placement_group_request = Mocker::GenCreatePlacementGroupRequest();
+  auto placement_group =
+      std::make_shared<gcs::GcsPlacementGroup>(create_placement_group_request);
+
+  // Schedule the placement group.
+  // One node is dead, so one bundle failed to schedule.
+  auto failure_handler = [this](std::shared_ptr<gcs::GcsPlacementGroup> placement_group) {
+    absl::MutexLock lock(&vector_mutex_);
+    ASSERT_TRUE(placement_group->GetUnplacedBundles().size() == 1);
+    failure_placement_groups_.emplace_back(std::move(placement_group));
+  };
+  auto success_handler = [this](std::shared_ptr<gcs::GcsPlacementGroup> placement_group) {
+    absl::MutexLock lock(&vector_mutex_);
+    success_placement_groups_.emplace_back(std::move(placement_group));
+  };
+
+  scheduler_->ScheduleUnplacedBundles(placement_group, failure_handler, success_handler);
+  ASSERT_TRUE(raylet_clients_[0]->GrantPrepareBundleResources());
+  gcs_node_manager_->RemoveNode(ClientID::FromBinary(node1->node_id()));
+  ASSERT_TRUE(raylet_clients_[1]->GrantPrepareBundleResources());
+  WaitPendingDone(failure_placement_groups_, 1);
+}
+
 }  // namespace ray
 
 int main(int argc, char **argv) {