[Core] Add Warning Log for the Infeasible Task Early Termination Feature (#51364)

This PR adds logs for the notification of the infeasible task early
termination feature.

Whenever there are more infeasible requests than what exists in
autoscaling state, we will print the warning describe the infeasible
task behavior and suggest to edit the cluster config or enable
infeasible task early termination feature.

Th log is tested manually and the following shows up in `gcs.out`:
```
[2025-03-13 23:06:14,518 W 42352 3849655] (gcs_server) gcs_autoscaler_state_manager.cc:72: There are tasks with infeasible resource requests and won't be scheduled. See https://docs.ray.io/en/latest/ray-core/scheduling/index.html#ray-scheduling-resources for more details. Please consider taking the following actions to solve the issue: 1. Updating the ray cluster to include nodes with all required resources in order to let the tasks be scheduled. 2. To prevent the tasks from hanging, you can consider enabling the infeasible task early exit feature by setting the 'enable_infeasible_task_early_exit' system config to 'true'.In a future release of Ray, we are planning to enable infeasible task early exit by default.
```

---------

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>
Co-authored-by: Edward Oakes <ed.nmi.oakes@gmail.com>

Author: MengjinYan

python/ray/_private/worker.py

@@ -2239,7 +2239,7 @@ def listen_error_messages(worker, threads_stopped):
             _, error_data = worker.gcs_error_subscriber.poll()
             if error_data is None:
                 continue
-            if error_data["job_id"] not in [
+            if error_data["job_id"] is not None and error_data["job_id"] not in [
                 worker.current_job_id.binary(),
                 JobID.nil().binary(),
             ]:

src/ray/gcs/gcs_server/gcs_autoscaler_state_manager.cc

@@ -34,14 +34,16 @@ GcsAutoscalerStateManager::GcsAutoscalerStateManager(
     const GcsPlacementGroupManager &gcs_placement_group_manager,
     rpc::NodeManagerClientPool &raylet_client_pool,
     InternalKVInterface &kv,
-    instrumented_io_context &io_context)
+    instrumented_io_context &io_context,
+    GcsPublisher *gcs_publisher)
     : session_name_(std::move(session_name)),
       gcs_node_manager_(gcs_node_manager),
       gcs_actor_manager_(gcs_actor_manager),
       gcs_placement_group_manager_(gcs_placement_group_manager),
       raylet_client_pool_(raylet_client_pool),
       kv_(kv),
-      io_context_(io_context) {}
+      io_context_(io_context),
+      gcs_publisher_(gcs_publisher) {}
 
 void GcsAutoscalerStateManager::HandleGetClusterResourceState(
     rpc::autoscaler::GetClusterResourceStateRequest request,
@@ -65,23 +67,50 @@ void GcsAutoscalerStateManager::HandleReportAutoscalingState(
     rpc::SendReplyCallback send_reply_callback) {
   RAY_CHECK(thread_checker_.IsOnSameThread());
 
-  // Never seen any autoscaling state before - so just takes this.
-  if (!autoscaling_state_.has_value()) {
-    autoscaling_state_ = *std::move(request.mutable_autoscaling_state());
-    send_reply_callback(ray::Status::OK(), nullptr, nullptr);
-    return;
-  }
-
-  // Cancel the infeasible requests if the feature is enabled
-  std::function<void()> callback = [this]() {
+  // Create the callback to cancel the infeasible requests if the feature is enabled
+  bool has_new_infeasible_requests = false;
+  std::function<void()> callback = [this, &has_new_infeasible_requests]() {
     bool enable_infeasible_task_early_exit =
         RayConfig::instance().enable_infeasible_task_early_exit();
 
     if (enable_infeasible_task_early_exit) {
       this->CancelInfeasibleRequests();
+    } else if (has_new_infeasible_requests) {
+      // publish error message
+      std::string error_message =
+          "There are tasks with infeasible resource requests that cannot "
+          "be scheduled. See "
+          "https://docs.ray.io/en/latest/ray-core/scheduling/"
+          "index.html#ray-scheduling-resources "
+          "for more details. Possible solutions: "
+          "1. Updating the ray cluster to include nodes with all required resources "
+          "2. To cause the tasks with infeasible requests to raise an error instead "
+          "of hanging, set the 'RAY_enable_infeasible_task_early_exit=true'. "
+          "This feature will be turned on by default in a future release of Ray.";
+      RAY_LOG(WARNING) << error_message;
+
+      if (gcs_publisher_ != nullptr) {
+        std::string error_type = "infeasible_resource_requests";
+        auto error_data_ptr = gcs::CreateErrorTableData(
+            error_type, error_message, absl::FromUnixMillis(current_time_ms()));
+        RAY_CHECK_OK(
+            gcs_publisher_->PublishError(session_name_, *error_data_ptr, nullptr));
+      }
     }
   };
 
+  // Never seen any autoscaling state before - so just takes this.
+  if (!autoscaling_state_.has_value()) {
+    autoscaling_state_ = *std::move(request.mutable_autoscaling_state());
+
+    if (autoscaling_state_->infeasible_resource_requests_size() > 0) {
+      has_new_infeasible_requests = true;
+    }
+
+    send_reply_callback(ray::Status::OK(), callback, nullptr);
+    return;
+  }
+
   // We have a state cached. We discard the incoming state if it's older than the
   // cached state.
   if (request.autoscaling_state().autoscaler_state_version() <
@@ -96,6 +125,10 @@ void GcsAutoscalerStateManager::HandleReportAutoscalingState(
   }
 
   // We should overwrite the cache version.
+  if (autoscaling_state_->infeasible_resource_requests_size() <
+      request.mutable_autoscaling_state()->infeasible_resource_requests_size()) {
+    has_new_infeasible_requests = true;
+  }
   autoscaling_state_ = std::move(*request.mutable_autoscaling_state());
   send_reply_callback(ray::Status::OK(), callback, nullptr);
 }

src/ray/gcs/gcs_server/gcs_autoscaler_state_manager.h

@@ -21,6 +21,7 @@
 
 #include "ray/gcs/gcs_server/gcs_init_data.h"
 #include "ray/gcs/gcs_server/gcs_kv_manager.h"
+#include "ray/gcs/pubsub/gcs_pub_sub.h"
 #include "ray/rpc/gcs_server/gcs_rpc_server.h"
 #include "ray/rpc/node_manager/node_manager_client_pool.h"
 #include "ray/util/thread_checker.h"
@@ -42,7 +43,8 @@ class GcsAutoscalerStateManager : public rpc::autoscaler::AutoscalerStateHandler
                             const GcsPlacementGroupManager &gcs_placement_group_manager,
                             rpc::NodeManagerClientPool &raylet_client_pool,
                             InternalKVInterface &kv,
-                            instrumented_io_context &io_context);
+                            instrumented_io_context &io_context,
+                            GcsPublisher *gcs_publisher);
 
   void HandleGetClusterResourceState(
       rpc::autoscaler::GetClusterResourceStateRequest request,
@@ -188,6 +190,9 @@ class GcsAutoscalerStateManager : public rpc::autoscaler::AutoscalerStateHandler
   InternalKVInterface &kv_;
   instrumented_io_context &io_context_;
 
+  // A publisher for publishing gcs messages.
+  GcsPublisher *gcs_publisher_;
+
   // The default value of the last seen version for the request is 0, which indicates
   // no version has been reported. So the first reported version should be 1.
   // We currently provide two guarantees for this version:

src/ray/gcs/gcs_server/gcs_server.cc

@@ -708,7 +708,8 @@ void GcsServer::InitGcsAutoscalerStateManager(const GcsInitData &gcs_init_data)
       *gcs_placement_group_manager_,
       *raylet_client_pool_,
       kv_manager_->GetInstance(),
-      io_context_provider_.GetDefaultIOContext());
+      io_context_provider_.GetDefaultIOContext(),
+      gcs_publisher_.get());
   gcs_autoscaler_state_manager_->Initialize(gcs_init_data);
 
   autoscaler_state_service_.reset(new rpc::autoscaler::AutoscalerStateGrpcService(

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

@@ -96,7 +96,8 @@ class GcsAutoscalerStateManagerTest : public ::testing::Test {
                                       *gcs_placement_group_manager_,
                                       *client_pool_,
                                       kv_manager_->GetInstance(),
-                                      io_service_));
+                                      io_service_,
+                                      /*gcs_publisher=*/nullptr));
   }
 
  public: