[ISSUE #1001] C++ client should block on build when sync settings failed (#1002)

Author: lizhimins

.github/workflows/cpp_build.yml

@@ -11,7 +11,7 @@ jobs:
         # Disable VS 2022 before https://github.com/bazelbuild/bazel/issues/18592 issue is solved
         # Remove macos-11 since there is no such runner available
         # os: [ubuntu-20.04, ubuntu-22.04, macos-11, macos-12, windows-2019, windows-2022]
-        os: [ubuntu-20.04, ubuntu-22.04, windows-2019]
+        os: [ubuntu-22.04, windows-2019]
     steps:
       - uses: actions/checkout@v2
       - name: Compile On Linux

cpp/source/client/ClientManagerImpl.cpp

@@ -94,22 +94,21 @@ void ClientManagerImpl::start() {
     SPDLOG_WARN("Unexpected client instance state: {}", state_.load(std::memory_order_relaxed));
     return;
   }
+
   state_.store(State::STARTING, std::memory_order_relaxed);
 
   callback_thread_pool_->start();
-
   scheduler_->start();
 
   std::weak_ptr<ClientManagerImpl> client_instance_weak_ptr = shared_from_this();
-
   auto heartbeat_functor = [client_instance_weak_ptr]() {
     auto client_instance = client_instance_weak_ptr.lock();
     if (client_instance) {
       client_instance->doHeartbeat();
     }
   };
-  heartbeat_task_id_ =
-      scheduler_->schedule(heartbeat_functor, HEARTBEAT_TASK_NAME, std::chrono::seconds(1), std::chrono::seconds(10));
+  heartbeat_task_id_ = scheduler_->schedule(
+      heartbeat_functor, HEARTBEAT_TASK_NAME, std::chrono::seconds(1), std::chrono::seconds(10));
   SPDLOG_DEBUG("Heartbeat task-id={}", heartbeat_task_id_);
 
   state_.store(State::STARTED, std::memory_order_relaxed);

cpp/source/client/TelemetryBidiReactor.cpp

@@ -38,7 +38,6 @@ TelemetryBidiReactor::TelemetryBidiReactor(std::weak_ptr<Client> client,
   auto ptr = client_.lock();
   auto deadline = std::chrono::system_clock::now() + std::chrono::hours(1);
   context_.set_deadline(deadline);
-  sync_settings_future_ = sync_settings_promise_.get_future();
   Metadata metadata;
   Signature::sign(ptr->config(), metadata);
   for (const auto& entry : metadata) {
@@ -56,8 +55,19 @@ TelemetryBidiReactor::~TelemetryBidiReactor() {
 }
 
 bool TelemetryBidiReactor::awaitApplyingSettings() {
-  sync_settings_future_.get();
-  return true;
+  auto settings_future = sync_settings_promise_.get_future();
+  std::future_status status = settings_future.wait_for(std::chrono::seconds(3));
+  if (status == std::future_status::ready) {
+    if (settings_future.get()) {
+      return true;
+    }
+  }
+  {
+    absl::MutexLock lk(&state_mtx_);
+    state_ = StreamState::Closed;
+    state_cv_.SignalAll();
+  }
+  return false;
 }
 
 void TelemetryBidiReactor::OnWriteDone(bool ok) {
@@ -283,21 +293,29 @@ void TelemetryBidiReactor::tryWriteNext() {
                 static_cast<std::uint8_t>(state_));
     return;
   }
+
   if (writes_.empty()) {
-    SPDLOG_DEBUG("No TelemetryCommand to write. Peer={}", peer_address_);
+    SPDLOG_DEBUG("No pending TelemetryCommand to write. Peer={}", peer_address_);
     return;
   }
 
   if (!writes_.empty()) {
-    SPDLOG_DEBUG("Writing telemetry command to {}: {}", peer_address_, writes_.front().ShortDebugString());
-    AddHold();
-    StartWrite(&(writes_.front()));
+    SPDLOG_DEBUG("Writing TelemetryCommand to {}: {}", peer_address_, writes_.front().ShortDebugString());
+    if (StreamState::Ready == state_) {
+      AddHold();
+      StartWrite(&(writes_.front()));
+    } else {
+      SPDLOG_WARN("Writing TelemetryCommand error due to unexpected state. State={}, Peer={}",
+                  static_cast<uint8_t>(state_), peer_address_);
+    }
   }
 }
 
 void TelemetryBidiReactor::signalClose() {
   absl::MutexLock lk(&state_mtx_);
-  state_ = StreamState::Closing;
+  if (state_ == StreamState::Ready) {
+    state_ = StreamState::Closing;
+  }
 }
 
 void TelemetryBidiReactor::close() {
@@ -361,7 +379,7 @@ void TelemetryBidiReactor::OnReadInitialMetadataDone(bool ok) {
   if (!ok) {
     // for read stream
     // Remove the hold corresponding to AddHold in TelemetryBidiReactor::TelemetryBidiReactor.
-    RemoveHold();
+    // RemoveHold();
 
     SPDLOG_DEBUG("Change state {} --> {}", static_cast<std::uint8_t>(state_),
                  static_cast<std::uint8_t>(StreamState::Closing));

cpp/source/client/include/TelemetryBidiReactor.h

@@ -40,19 +40,25 @@ enum class StreamState : std::uint8_t
   Closed = 2,
 };
 
-/// stream-state: ready --> closing --> closed
+/// TelemetryBidiReactor: Manages a bidirectional gRPC stream for telemetry data
 ///
-/// requirement:
-///    1, close --> blocking wait till bidireactor is closed;
-///    2, when session is closed and client is still active, recreate a new session to accept incoming commands from
-///    server
+/// Stream State Transitions:
+///    Ready --> Closing --> Closed
 ///
+/// Key Features:
+///    1. Close Operation: Performs a blocking wait until the bidirectional reactor is fully closed.
+///    2. Session Management: If the session closes while the client is still active,
+///       it automatically initiates the creation of a new session to maintain
+///       communication with the server.
+///
+/// The reactor handles reading from and writing to the stream, manages stream state,
+/// and applies settings received from the server.
 class TelemetryBidiReactor : public grpc::ClientBidiReactor<TelemetryCommand, TelemetryCommand>,
                              public std::enable_shared_from_this<TelemetryBidiReactor> {
 public:
   TelemetryBidiReactor(std::weak_ptr<Client> client, rmq::MessagingService::Stub* stub, std::string peer_address);
 
-  ~TelemetryBidiReactor();
+  ~TelemetryBidiReactor() override;
 
   /// Notifies the application that all operations associated with this RPC
   /// have completed and all Holds have been removed. OnDone provides the RPC
@@ -125,7 +131,6 @@ class TelemetryBidiReactor : public grpc::ClientBidiReactor<TelemetryCommand, Te
   absl::CondVar state_cv_;
 
   std::promise<bool> sync_settings_promise_;
-  std::future<bool> sync_settings_future_;
 
   void applySettings(const rmq::Settings& settings);
 
@@ -137,6 +142,7 @@ class TelemetryBidiReactor : public grpc::ClientBidiReactor<TelemetryCommand, Te
 
   /// Attempt to write pending telemetry command to server.
   void tryWriteNext() LOCKS_EXCLUDED(state_mtx_, writes_mtx_);
+
   void signalClose();
 };
 

cpp/source/rocketmq/ClientImpl.cpp

@@ -106,26 +106,30 @@ void ClientImpl::start() {
   name_server_resolver_->start();
 
   client_config_.client_id = clientId();
-
   if (!client_manager_) {
-    client_manager_ = std::make_shared<ClientManagerImpl>(client_config_.resource_namespace, client_config_.withSsl);
+    client_manager_ = std::make_shared<ClientManagerImpl>(
+        client_config_.resource_namespace, client_config_.withSsl);
+    client_manager_->start();
   }
-  client_manager_->start();
 
   const auto& endpoint = name_server_resolver_->resolve();
   if (endpoint.empty()) {
     SPDLOG_ERROR("Failed to resolve name server address");
-    abort();
+    return;
   }
 
-  createSession(endpoint, false);
-  {
-    absl::MutexLock lk(&session_map_mtx_);
-    session_map_[endpoint]->await();
+  while (true) {
+    createSession(endpoint, false);
+    {
+      absl::MutexLock lk(&session_map_mtx_);
+      if (session_map_.contains(endpoint) && session_map_[endpoint]->await()) {
+        break;
+      }
+      session_map_.erase(endpoint);
+    }
   }
 
   std::weak_ptr<ClientImpl> ptr(self());
-
   {
     // Query routes for topics of interest in synchronous
     std::vector<std::string> topics;
@@ -164,8 +168,9 @@ void ClientImpl::start() {
     }
   };
 
-  route_update_handle_ = client_manager_->getScheduler()->schedule(route_update_functor, UPDATE_ROUTE_TASK_NAME,
-                                                                   std::chrono::seconds(10), std::chrono::seconds(30));
+  route_update_handle_ = client_manager_->getScheduler()->schedule(
+      route_update_functor, UPDATE_ROUTE_TASK_NAME,
+      std::chrono::seconds(10), std::chrono::seconds(30));
 
   auto telemetry_functor = [ptr]() {
     std::shared_ptr<ClientImpl> base = ptr.lock();
@@ -597,8 +602,11 @@ void ClientImpl::notifyClientTermination(const NotifyClientTerminationRequest& r
   Signature::sign(client_config_, metadata);
 
   for (const auto& endpoint : endpoints) {
-    client_manager_->notifyClientTermination(endpoint, metadata, request,
-                                             absl::ToChronoMilliseconds(client_config_.request_timeout));
+    std::error_code ec = client_manager_->notifyClientTermination(
+        endpoint, metadata, request,absl::ToChronoMilliseconds(client_config_.request_timeout));
+    if (ec) {
+      SPDLOG_WARN("Notify client termination error, ErrorCode={}, Endpoint={}", ec.message(), endpoint);
+    }
   }
 }
 

cpp/source/rocketmq/ConsumeMessageServiceImpl.cpp

@@ -72,7 +72,7 @@ void ConsumeMessageServiceImpl::dispatch(std::shared_ptr<ProcessQueue> process_q
     return;
   }
 
-  for (auto message : messages) {
+  for (const auto& message : messages) {
     auto consume_task = std::make_shared<ConsumeTask>(shared_from_this(), process_queue, message);
     pool_->submit([consume_task]() { consume_task->process(); });
   }

cpp/source/rocketmq/ProducerImpl.cpp

@@ -56,7 +56,7 @@ void ProducerImpl::start() {
 
   State expecting = State::STARTING;
   if (!state_.compare_exchange_strong(expecting, State::STARTED)) {
-    SPDLOG_ERROR("Start with unexpected state. Expecting: {}, Actual: {}", State::STARTING,
+    SPDLOG_ERROR("Producer started with an unexpected state. Expecting: {}, Actual: {}", State::STARTING,
                  state_.load(std::memory_order_relaxed));
     return;
   }

cpp/source/rocketmq/PushConsumerImpl.cpp

@@ -65,7 +65,6 @@ void PushConsumerImpl::start() {
   if (!message_listener_) {
     SPDLOG_ERROR("Required message listener is missing");
     abort();
-    return;
   }
 
   client_config_.subscriber.group.set_resource_namespace(resourceNamespace());

cpp/source/rocketmq/include/ConsumeMessageServiceImpl.h

@@ -43,8 +43,8 @@ class ConsumeMessageServiceImpl
    * Make it noncopyable.
    */
   ConsumeMessageServiceImpl(const ConsumeMessageServiceImpl &other) = delete;
-  ConsumeMessageServiceImpl &
-  operator=(const ConsumeMessageServiceImpl &other) = delete;
+
+  ConsumeMessageServiceImpl &operator=(const ConsumeMessageServiceImpl &other) = delete;
 
   void start() override;