diff --git a/src/ray/common/ray_config_def.h b/src/ray/common/ray_config_def.h
index a1af82469..c702102ad 100644
--- a/src/ray/common/ray_config_def.h
+++ b/src/ray/common/ray_config_def.h
@@ -266,6 +266,9 @@ RAY_CONFIG(int64_t, ping_gcs_rpc_server_interval_milliseconds, 1000)
 /// Maximum number of times to retry ping gcs rpc server when gcs server restarts.
 RAY_CONFIG(int32_t, ping_gcs_rpc_server_max_retries, 1)
 
+/// Minimum interval between reconnecting gcs rpc server when gcs server restarts.
+RAY_CONFIG(int32_t, minimum_gcs_reconnect_interval_milliseconds, 5000)
+
 /// Whether start the Plasma Store as a Raylet thread.
 RAY_CONFIG(bool, plasma_store_as_thread, false)
 
diff --git a/src/ray/gcs/gcs_client/service_based_accessor.cc b/src/ray/gcs/gcs_client/service_based_accessor.cc
index cc2907c63..2cf1d2caf 100644
--- a/src/ray/gcs/gcs_client/service_based_accessor.cc
+++ b/src/ray/gcs/gcs_client/service_based_accessor.cc
@@ -90,17 +90,23 @@ Status ServiceBasedJobInfoAccessor::AsyncSubscribeAll(
 
 void ServiceBasedJobInfoAccessor::AsyncResubscribe(bool is_pubsub_server_restarted) {
   RAY_LOG(DEBUG) << "Reestablishing subscription for job info.";
+  auto fetch_all_done = [](const Status &status) {
+    RAY_LOG(INFO) << "Finished fetching all job information from gcs server after gcs "
+                     "server or pub-sub server is restarted.";
+  };
+
   // If only the GCS sever has restarted, we only need to fetch data from the GCS server.
   // If the pub-sub server has also restarted, we need to resubscribe to the pub-sub
   // server first, then fetch data from the GCS server.
   if (is_pubsub_server_restarted) {
     if (subscribe_operation_ != nullptr) {
-      RAY_CHECK_OK(subscribe_operation_(
-          [this](const Status &status) { fetch_all_data_operation_(nullptr); }));
+      RAY_CHECK_OK(subscribe_operation_([this, fetch_all_done](const Status &status) {
+        fetch_all_data_operation_(fetch_all_done);
+      }));
     }
   } else {
     if (fetch_all_data_operation_ != nullptr) {
-      fetch_all_data_operation_(nullptr);
+      fetch_all_data_operation_(fetch_all_done);
     }
   }
 }
@@ -301,14 +307,20 @@ Status ServiceBasedActorInfoAccessor::AsyncUnsubscribe(const ActorID &actor_id)
 
 void ServiceBasedActorInfoAccessor::AsyncResubscribe(bool is_pubsub_server_restarted) {
   RAY_LOG(DEBUG) << "Reestablishing subscription for actor info.";
+  auto fetch_all_done = [](const Status &status) {
+    RAY_LOG(INFO) << "Finished fetching all actor information from gcs server after gcs "
+                     "server or pub-sub server is restarted.";
+  };
+
   // If only the GCS sever has restarted, we only need to fetch data from the GCS server.
   // If the pub-sub server has also restarted, we need to resubscribe to the pub-sub
   // server first, then fetch data from the GCS server.
   absl::MutexLock lock(&mutex_);
   if (is_pubsub_server_restarted) {
     if (subscribe_all_operation_ != nullptr) {
-      RAY_CHECK_OK(subscribe_all_operation_(
-          [this](const Status &status) { fetch_all_data_operation_(nullptr); }));
+      RAY_CHECK_OK(subscribe_all_operation_([this, fetch_all_done](const Status &status) {
+        fetch_all_data_operation_(fetch_all_done);
+      }));
     }
     for (auto &item : subscribe_operations_) {
       auto &actor_id = item.first;
@@ -325,7 +337,7 @@ void ServiceBasedActorInfoAccessor::AsyncResubscribe(bool is_pubsub_server_resta
     }
   } else {
     if (fetch_all_data_operation_ != nullptr) {
-      fetch_all_data_operation_(nullptr);
+      fetch_all_data_operation_(fetch_all_done);
     }
     for (auto &item : fetch_data_operations_) {
       item.second(nullptr);
@@ -651,20 +663,27 @@ void ServiceBasedNodeInfoAccessor::HandleNotification(const GcsNodeInfo &node_in
 
 void ServiceBasedNodeInfoAccessor::AsyncResubscribe(bool is_pubsub_server_restarted) {
   RAY_LOG(DEBUG) << "Reestablishing subscription for node info.";
+  auto fetch_all_done = [](const Status &status) {
+    RAY_LOG(INFO) << "Finished fetching all node information from gcs server after gcs "
+                     "server or pub-sub server is restarted.";
+  };
+
   // If only the GCS sever has restarted, we only need to fetch data from the GCS server.
   // If the pub-sub server has also restarted, we need to resubscribe to the pub-sub
   // server first, then fetch data from the GCS server.
   if (is_pubsub_server_restarted) {
     if (subscribe_node_operation_ != nullptr) {
-      RAY_CHECK_OK(subscribe_node_operation_(
-          [this](const Status &status) { fetch_node_data_operation_(nullptr); }));
+      RAY_CHECK_OK(
+          subscribe_node_operation_([this, fetch_all_done](const Status &status) {
+            fetch_node_data_operation_(fetch_all_done);
+          }));
     }
     if (subscribe_batch_resource_usage_operation_ != nullptr) {
       RAY_CHECK_OK(subscribe_batch_resource_usage_operation_(nullptr));
     }
   } else {
     if (fetch_node_data_operation_ != nullptr) {
-      fetch_node_data_operation_(nullptr);
+      fetch_node_data_operation_(fetch_all_done);
     }
   }
 }
diff --git a/src/ray/gcs/gcs_client/service_based_gcs_client.cc b/src/ray/gcs/gcs_client/service_based_gcs_client.cc
index 359f6cd81..884612106 100644
--- a/src/ray/gcs/gcs_client/service_based_gcs_client.cc
+++ b/src/ray/gcs/gcs_client/service_based_gcs_client.cc
@@ -25,7 +25,9 @@ namespace ray {
 namespace gcs {
 
 ServiceBasedGcsClient::ServiceBasedGcsClient(const GcsClientOptions &options)
-    : GcsClient(options) {}
+    : GcsClient(options),
+      last_reconnect_timestamp_ms_(0),
+      last_reconnect_address_(std::make_pair("", -1)) {}
 
 Status ServiceBasedGcsClient::Connect(boost::asio::io_service &io_service) {
   RAY_CHECK(!is_connected_);
@@ -175,7 +177,7 @@ void ServiceBasedGcsClient::GcsServiceFailureDetected(rpc::GcsServiceFailureType
     ReconnectGcsServer();
     // NOTE(ffbin): Currently we don't support the case where the pub-sub server restarts,
     // because we use the same Redis server for both GCS storage and pub-sub. So the
-    // following flag is alway false.
+    // following flag is always false.
     resubscribe_func_(false);
     // Resend resource usage after reconnected, needed by resource view in GCS.
     node_accessor_->AsyncReReportResourceUsage();
@@ -191,11 +193,27 @@ void ServiceBasedGcsClient::ReconnectGcsServer() {
   int index = 0;
   for (; index < RayConfig::instance().ping_gcs_rpc_server_max_retries(); ++index) {
     if (get_server_address_func_(&address)) {
-      RAY_LOG(DEBUG) << "Attemptting to reconnect to GCS server: " << address.first << ":"
-                     << address.second;
+      // After GCS is restarted, the gcs client will reestablish the connection. At
+      // present, every failed RPC request will trigger `ReconnectGcsServer`. In order to
+      // avoid repeated connections in a short period of time, we add a protection
+      // mechanism: if the address does not change (meaning gcs server doesn't restart),
+      // the connection can be made at most once in
+      // `minimum_gcs_reconnect_interval_milliseconds` milliseconds.
+      if (last_reconnect_address_ == address &&
+          (current_sys_time_ms() - last_reconnect_timestamp_ms_) <
+              RayConfig::instance().minimum_gcs_reconnect_interval_milliseconds()) {
+        RAY_LOG(INFO)
+            << "Repeated reconnection in "
+            << RayConfig::instance().minimum_gcs_reconnect_interval_milliseconds()
+            << "milliseconds, return directly.";
+        return;
+      }
+
+      RAY_LOG(INFO) << "Attemptting to reconnect to GCS server: " << address.first << ":"
+                    << address.second;
       if (Ping(address.first, address.second, 100)) {
-        RAY_LOG(DEBUG) << "Reconnected to GCS server: " << address.first << ":"
-                       << address.second;
+        RAY_LOG(INFO) << "Reconnected to GCS server: " << address.first << ":"
+                      << address.second;
         break;
       }
     }
@@ -205,6 +223,8 @@ void ServiceBasedGcsClient::ReconnectGcsServer() {
 
   if (index < RayConfig::instance().ping_gcs_rpc_server_max_retries()) {
     gcs_rpc_client_->Reset(address.first, address.second, *client_call_manager_);
+    last_reconnect_address_ = address;
+    last_reconnect_timestamp_ms_ = current_sys_time_ms();
   } else {
     RAY_LOG(FATAL) << "Couldn't reconnect to GCS server. The last attempted GCS "
                       "server address was "
diff --git a/src/ray/gcs/gcs_client/service_based_gcs_client.h b/src/ray/gcs/gcs_client/service_based_gcs_client.h
index 05fb4c467..906165099 100644
--- a/src/ray/gcs/gcs_client/service_based_gcs_client.h
+++ b/src/ray/gcs/gcs_client/service_based_gcs_client.h
@@ -72,6 +72,8 @@ class RAY_EXPORT ServiceBasedGcsClient : public GcsClient {
   std::function<bool(std::pair<std::string, int> *)> get_server_address_func_;
   std::function<void(bool)> resubscribe_func_;
   std::pair<std::string, int> current_gcs_server_address_;
+  int64_t last_reconnect_timestamp_ms_;
+  std::pair<std::string, int> last_reconnect_address_;
 };
 
 }  // namespace gcs