diff --git a/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc b/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc
index f53248114..d59ec24d8 100644
--- a/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc
+++ b/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc
@@ -971,14 +971,14 @@ TEST_F(ServiceBasedGcsClientTest, TestWorkerInfo) {
   auto worker_data = Mocker::GenWorkerTableData();
   worker_data->mutable_worker_address()->set_worker_id(WorkerID::FromRandom().Binary());
   ASSERT_TRUE(ReportWorkerFailure(worker_data));
-  WaitForExpectedCount(worker_failure_count, 0);
+  WaitForExpectedCount(worker_failure_count, 1);
 
   // Add a worker to GCS.
   ASSERT_TRUE(AddWorker(worker_data));
 
   // Report a worker failure to GCS when this worker is actually exist.
   ASSERT_TRUE(ReportWorkerFailure(worker_data));
-  WaitForExpectedCount(worker_failure_count, 1);
+  WaitForExpectedCount(worker_failure_count, 2);
 }
 
 TEST_F(ServiceBasedGcsClientTest, TestErrorInfo) {
diff --git a/src/ray/gcs/gcs_server/gcs_worker_manager.cc b/src/ray/gcs/gcs_server/gcs_worker_manager.cc
index bc380fd90..c9e5e8b94 100644
--- a/src/ray/gcs/gcs_server/gcs_worker_manager.cc
+++ b/src/ray/gcs/gcs_server/gcs_worker_manager.cc
@@ -40,45 +40,29 @@ void GcsWorkerManager::HandleReportWorkerFailure(
   worker_failure_data->CopyFrom(request.worker_failure());
   worker_failure_data->set_is_alive(false);
 
-  // Before handle ReportWorkerFailureRequest, you should check if the worker is exists.
-  auto on_get_done = [this, worker_address, worker_id, node_id, worker_failure_data,
-                      reply, send_reply_callback](
-                         const Status &status,
-                         const boost::optional<WorkerTableData> &result) {
-    if (result) {
-      auto on_put_done = [this, worker_address, worker_id, node_id, worker_failure_data,
-                          reply, send_reply_callback](const Status &status) {
-        if (!status.ok()) {
-          RAY_LOG(ERROR) << "Failed to report worker failure, worker id = " << worker_id
-                         << ", node id = " << node_id
-                         << ", address = " << worker_address.ip_address();
-        } else {
-          stats::UnintentionalWorkerFailures.Record(1);
-          RAY_CHECK_OK(gcs_pub_sub_->Publish(WORKER_CHANNEL, worker_id.Binary(),
-                                             worker_failure_data->SerializeAsString(),
-                                             nullptr));
-        }
-        GCS_RPC_SEND_REPLY(send_reply_callback, reply, status);
-      };
-
-      // The worker exists in worker table, you can update the info of this worker.
-      Status report_status = gcs_table_storage_->WorkerTable().Put(
-          worker_id, *worker_failure_data, on_put_done);
-      if (!report_status.ok()) {
-        on_put_done(report_status);
-      }
+  auto on_done = [this, worker_address, worker_id, node_id, worker_failure_data, reply,
+                  send_reply_callback](const Status &status) {
+    if (!status.ok()) {
+      RAY_LOG(ERROR) << "Failed to report worker failure, worker id = " << worker_id
+                     << ", node id = " << node_id
+                     << ", address = " << worker_address.ip_address();
     } else {
-      // The worker doesn't exists in worker table.
-      RAY_LOG(WARNING) << "Failed to report worker failure, the worker doesn't "
-                          "exist, worker id = "
-                       << worker_id << ", node id = " << node_id
-                       << ", address = " << worker_address.ip_address();
-      GCS_RPC_SEND_REPLY(send_reply_callback, reply, status);
+      stats::UnintentionalWorkerFailures.Record(1);
+      RAY_CHECK_OK(gcs_pub_sub_->Publish(WORKER_CHANNEL, worker_id.Binary(),
+                                         worker_failure_data->SerializeAsString(),
+                                         nullptr));
     }
+    GCS_RPC_SEND_REPLY(send_reply_callback, reply, status);
   };
-  Status status = gcs_table_storage_->WorkerTable().Get(worker_id, on_get_done);
+
+  // As soon as the worker starts, it will register with GCS. It ensures that GCS receives
+  // the worker registration information first and then the worker failure message, so we
+  // delete the get operation. Related issues:
+  // https://github.com/ray-project/ray/pull/11599
+  Status status =
+      gcs_table_storage_->WorkerTable().Put(worker_id, *worker_failure_data, on_done);
   if (!status.ok()) {
-    on_get_done(status, boost::none);
+    on_done(status);
   }
 }
 
diff --git a/src/ray/gcs/gcs_server/test/gcs_server_rpc_test.cc b/src/ray/gcs/gcs_server/test/gcs_server_rpc_test.cc
index d6d897e3f..814f429b8 100644
--- a/src/ray/gcs/gcs_server/test/gcs_server_rpc_test.cc
+++ b/src/ray/gcs/gcs_server/test/gcs_server_rpc_test.cc
@@ -708,7 +708,7 @@ TEST_F(GcsServerTest, TestWorkerInfo) {
   report_worker_failure_request.mutable_worker_failure()->CopyFrom(*worker_failure_data);
   ASSERT_TRUE(ReportWorkerFailure(report_worker_failure_request));
   std::vector<rpc::WorkerTableData> worker_table_data = GetAllWorkerInfo();
-  ASSERT_TRUE(worker_table_data.size() == 0);
+  ASSERT_TRUE(worker_table_data.size() == 1);
 
   // Add worker info
   auto worker_data = Mocker::GenWorkerTableData();
@@ -716,7 +716,7 @@ TEST_F(GcsServerTest, TestWorkerInfo) {
   rpc::AddWorkerInfoRequest add_worker_request;
   add_worker_request.mutable_worker_data()->CopyFrom(*worker_data);
   ASSERT_TRUE(AddWorkerInfo(add_worker_request));
-  ASSERT_TRUE(GetAllWorkerInfo().size() == 1);
+  ASSERT_TRUE(GetAllWorkerInfo().size() == 2);
 
   // Get worker info
   boost::optional<rpc::WorkerTableData> result =