Fix compile and lint error

Signed-off-by: 黑驰 <[email protected]>

Author: wumuzi520

python/ray/dashboard/modules/virtual_cluster/tests/test_virtual_cluster.py

@@ -447,3 +447,7 @@ def _get_virtual_clusters():
             return False
 
     wait_for_condition(_get_virtual_clusters, timeout=10)
+
+
+if __name__ == "__main__":
+    pass

src/ray/gcs/gcs_server/gcs_virtual_cluster.cc

@@ -357,7 +357,7 @@ Status DivisibleCluster::CreateJobCluster(const std::string &job_cluster_id,
   ReplicaInstances replica_instances_to_add;
   // Lookup undivided alive node instances based on `replica_sets_to_add`.
   auto success = LookupUndividedNodeInstances(
-      replica_sets, replica_instances_to_add, [this](const auto &node_instance) {
+      replica_sets, replica_instances_to_add, [](const auto &node_instance) {
         return !node_instance.is_dead();
       });
   if (!success) {
@@ -472,9 +472,12 @@ bool DivisibleCluster::ReplenishNodeInstances(
     const NodeInstanceReplenishCallback &callback) {
   RAY_CHECK(callback != nullptr);
   bool any_node_instance_replenished = false;
-  for (const auto &[job_cluster_id, job_cluster] : job_clusters_) {
+  for (const auto &[_, job_cluster] : job_clusters_) {
+    // Explicitly capture the `job_cluster` by value to avoid the compile error.
+    // e.g. error: 'job_cluster_id' in capture list does not name a variable
+    const auto &job_cluster_id = job_cluster->GetID();
     bool replenished = job_cluster->ReplenishNodeInstances(
-        [this, &job_cluster_id, &callback, &any_node_instance_replenished](
+        [this, &job_cluster_id, &any_node_instance_replenished, &callback](
             std::shared_ptr<NodeInstance> node_instance) {
           const auto &template_id = node_instance->template_id();
           if (auto replenished_node_instance = callback(node_instance)) {
@@ -515,7 +518,11 @@ bool DivisibleCluster::ReplenishNodeInstances(
 
     if (replenished) {
       // Flush and publish the job cluster data.
-      async_data_flusher_(job_cluster->ToProto(), nullptr);
+      auto status = async_data_flusher_(job_cluster->ToProto(), nullptr);
+      if (!status.ok()) {
+        RAY_LOG(ERROR) << "Failed to flush and publish the job cluster " << job_cluster_id
+                       << " data: " << status.message();
+      }
     }
   }
 
@@ -845,7 +852,7 @@ Status PrimaryCluster::DetermineNodeInstanceAdditionsAndRemovals(
   // Lookup undivided alive node instances from main cluster based on
   // `replica_sets_to_add`.
   auto success = LookupUndividedNodeInstances(
-      replica_sets_to_add, replica_instances_to_add, [this](const auto &node_instance) {
+      replica_sets_to_add, replica_instances_to_add, [](const auto &node_instance) {
         return !node_instance.is_dead();
       });
   if (!success) {
@@ -1047,7 +1054,11 @@ void PrimaryCluster::ReplenishAllClusterNodeInstances() {
   for (auto &[_, logical_cluster] : logical_clusters_) {
     if (logical_cluster->ReplenishNodeInstances(node_instance_replenish_callback)) {
       // Flush the logical cluster data.
-      async_data_flusher_(logical_cluster->ToProto(), nullptr);
+      auto status = async_data_flusher_(logical_cluster->ToProto(), nullptr);
+      if (!status.ok()) {
+        RAY_LOG(ERROR) << "Failed to flush logical cluster " << logical_cluster->GetID()
+                       << " data, status: " << status.ToString();
+      }
     }
   }
 

src/ray/gcs/gcs_server/gcs_virtual_cluster_manager.cc

@@ -131,9 +131,9 @@ void GcsVirtualClusterManager::OnJobFinished(const rpc::JobTableData &job_data)
 
   auto status = divisible_cluster->RemoveJobCluster(
       virtual_cluster_id,
-      [this, job_cluster_id](const Status &status,
-                             std::shared_ptr<rpc::VirtualClusterTableData> data,
-                             const ReplicaSets *replica_sets_to_recommend) {
+      [job_cluster_id](const Status &status,
+                       std::shared_ptr<rpc::VirtualClusterTableData> data,
+                       const ReplicaSets *replica_sets_to_recommend) {
         if (!status.ok() || !data->is_removed()) {
           RAY_LOG(WARNING) << "Failed to remove job cluster " << job_cluster_id.Binary()
                            << " when handling job finished event. status: "
@@ -450,7 +450,12 @@ void GcsVirtualClusterManager::OnDetachedActorDestroy(
     JobCluster *job_cluster = dynamic_cast<JobCluster *>(virtual_cluster.get());
     job_cluster->OnDetachedActorDestroy(actor_id);
     if (!job_cluster->InUse() && job_cluster->IsFinished()) {
-      primary_cluster_->RemoveVirtualCluster(virtual_cluster_id, nullptr);
+      auto status = primary_cluster_->RemoveVirtualCluster(virtual_cluster_id, nullptr);
+      if (!status.ok()) {
+        RAY_LOG(WARNING) << "Failed to remove virtual cluster " << virtual_cluster_id
+                         << " after handling detached actor destroy event. status: "
+                         << status.message();
+      }
     }
   }
 }
@@ -483,7 +488,13 @@ void GcsVirtualClusterManager::OnDetachedPlacementGroupDestroy(
     JobCluster *job_cluster = dynamic_cast<JobCluster *>(virtual_cluster.get());
     job_cluster->OnDetachedPlacementGroupDestroy(placement_group_id);
     if (!job_cluster->InUse() && job_cluster->IsFinished()) {
-      primary_cluster_->RemoveVirtualCluster(virtual_cluster_id, nullptr);
+      auto status = primary_cluster_->RemoveVirtualCluster(virtual_cluster_id, nullptr);
+      if (!status.ok()) {
+        RAY_LOG(WARNING)
+            << "Failed to remove virtual cluster " << virtual_cluster_id
+            << " after handling detached placement group destroy event. status: "
+            << status.message();
+      }
     }
   }
 }

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

@@ -1005,7 +1005,7 @@ TEST_F(FailoverTest, FailoverWithDeadNodes) {
     };
     ASSERT_TRUE(
         virtual_cluster_1->ReplenishNodeInstances(node_instance_replenish_callback));
-    async_data_flusher_(virtual_cluster_1->ToProto(), nullptr);
+    ASSERT_TRUE(async_data_flusher_(virtual_cluster_1->ToProto(), nullptr).ok());
 
     // Mock a gcs_init_data.
     instrumented_io_context io_service;

src/ray/protobuf/gcs_service.proto

@@ -940,7 +940,8 @@ service VirtualClusterInfoGcsService {
   rpc CreateOrUpdateVirtualCluster(CreateOrUpdateVirtualClusterRequest)
       returns (CreateOrUpdateVirtualClusterReply);
   // Remove a virtual cluster.
-  rpc RemoveVirtualCluster(RemoveVirtualClusterRequest) returns (RemoveVirtualClusterReply);
+  rpc RemoveVirtualCluster(RemoveVirtualClusterRequest)
+      returns (RemoveVirtualClusterReply);
   // Get virtual clusters.
   rpc GetVirtualClusters(GetVirtualClustersRequest) returns (GetVirtualClustersReply);
   // Create job cluster.