bugfix about test_dynres.py (#7615)

Co-authored-by: senlin.zsl <senlin.zsl@antfin.com>

src/ray/gcs/gcs_server/gcs_node_manager.cc

@@ -17,7 +17,7 @@ GcsNodeManager::GcsNodeManager(boost::asio::io_service &io_service,
 }
 
 void GcsNodeManager::HandleHeartbeat(const ClientID &node_id,
-                                     rpc::HeartbeatTableData &&heartbeat_data) {
+                                     const rpc::HeartbeatTableData &heartbeat_data) {
   heartbeats_[node_id] = num_heartbeats_timeout_;
   heartbeat_buffer_[node_id] = heartbeat_data;
 }

src/ray/gcs/gcs_server/gcs_node_manager.h

@@ -23,7 +23,8 @@ class GcsNodeManager {
   ///
   /// \param node_id The client ID of the Raylet that sent the heartbeat.
   /// \param heartbeat_data The heartbeat sent by the client.
-  void HandleHeartbeat(const ClientID &node_id, rpc::HeartbeatTableData &&heartbeat_data);
+  void HandleHeartbeat(const ClientID &node_id,
+                       const rpc::HeartbeatTableData &heartbeat_data);
 
  protected:
   /// Listen for heartbeats from Raylets and mark Raylets

src/ray/gcs/gcs_server/node_info_handler_impl.cc

@@ -88,10 +88,22 @@ void DefaultNodeInfoHandler::HandleReportHeartbeat(
     SendReplyCallback send_reply_callback) {
   ClientID node_id = ClientID::FromBinary(request.heartbeat().client_id());
   RAY_LOG(DEBUG) << "Reporting heartbeat, node id = " << node_id;
-  rpc::HeartbeatTableData heartbeat_data;
-  heartbeat_data.CopyFrom(request.heartbeat());
-  gcs_node_manager_.HandleHeartbeat(node_id, std::move(heartbeat_data));
-  send_reply_callback(Status::OK(), nullptr, nullptr);
+  auto on_done = [node_id, reply, send_reply_callback](Status status) {
+    if (!status.ok()) {
+      RAY_LOG(ERROR) << "Failed to report heartbeat: " << status.ToString()
+                     << ", node id = " << node_id;
+    }
+    GCS_RPC_SEND_REPLY(send_reply_callback, reply, status);
+  };
+
+  auto heartbeat_data = std::make_shared<rpc::HeartbeatTableData>();
+  heartbeat_data->CopyFrom(request.heartbeat());
+  gcs_node_manager_.HandleHeartbeat(node_id, *heartbeat_data);
+
+  Status status = gcs_client_.Nodes().AsyncReportHeartbeat(heartbeat_data, on_done);
+  if (!status.ok()) {
+    on_done(status);
+  }
   RAY_LOG(DEBUG) << "Finished reporting heartbeat, node id = " << node_id;
 }