Add more logs for reconnection

Author: BewareMyPower

lib/ClientConnection.cc

@@ -1329,7 +1329,7 @@ void ClientConnection::close(Result result, bool detach) {
     if (result != ResultDisconnected && result != ResultRetryable) {
         LOG_ERROR(cnxString_ << "Connection closed with " << result);
     } else {
-        LOG_INFO(cnxString_ << "Connection disconnected");
+        LOG_WARN(cnxString_ << "Connection disconnected: " << result);
     }
     // Remove the connection from the pool before completing any promise
     if (detach) {

lib/ConsumerImpl.cc

@@ -222,7 +222,7 @@ void ConsumerImpl::onNegativeAcksSend(const std::set<MessageId>& messageIds) {
 
 void ConsumerImpl::connectionOpened(const ClientConnectionPtr& cnx) {
     if (state_ == Closed) {
-        LOG_DEBUG(getName() << "connectionOpened : Consumer is already closed");
+        LOG_WARN(getName() << "connectionOpened : Consumer is already closed");
         return;
     }
 
@@ -246,6 +246,7 @@ void ConsumerImpl::connectionOpened(const ClientConnectionPtr& cnx) {
 
     ClientImplPtr client = client_.lock();
     uint64_t requestId = client->newRequestId();
+    LOG_WARN(getName() << "start subscribing the topic");
     SharedBuffer cmd = Commands::newSubscribe(
         *topic_, subscription_, consumerId_, requestId, getSubType(), consumerName_, subscriptionMode_,
         subscribeMessageId, readCompacted_, config_.getProperties(), config_.getSubscriptionProperties(),
@@ -274,6 +275,7 @@ void ConsumerImpl::sendFlowPermitsToBroker(const ClientConnectionPtr& cnx, int n
 }
 
 void ConsumerImpl::handleCreateConsumer(const ClientConnectionPtr& cnx, Result result) {
+    LOG_WARN(getName() << "created consumer: " << result << ", " << cnx->cnxString());
     static bool firstTime = true;
     if (result == ResultOk) {
         if (firstTime) {
@@ -294,7 +296,7 @@ void ConsumerImpl::handleCreateConsumer(const ClientConnectionPtr& cnx, Result r
             availablePermits_ = 0;
         }
 
-        LOG_DEBUG(getName() << "Send initial flow permits: " << config_.getReceiverQueueSize());
+        LOG_WARN(getName() << "Send initial flow permits: " << config_.getReceiverQueueSize());
         if (consumerTopicType_ == NonPartitioned || !firstTime) {
             if (config_.getReceiverQueueSize() != 0) {
                 sendFlowPermitsToBroker(cnx, config_.getReceiverQueueSize());
@@ -1204,7 +1206,7 @@ void ConsumerImpl::negativeAcknowledge(const MessageId& messageId) {
 }
 
 void ConsumerImpl::disconnectConsumer() {
-    LOG_INFO("Broker notification of Closed consumer: " << consumerId_);
+    LOG_WARN("Broker notification of Closed consumer: " << consumerId_);
     resetCnx();
     scheduleReconnection();
 }

lib/HandlerBase.cc

@@ -67,36 +67,37 @@ void HandlerBase::setCnx(const ClientConnectionPtr& cnx) {
 
 void HandlerBase::grabCnx() {
     if (getCnx().lock()) {
-        LOG_INFO(getName() << "Ignoring reconnection request since we're already connected");
+        LOG_WARN(getName() << "Ignoring reconnection request since we're already connected");
         return;
     }
 
     bool expectedState = false;
     if (!reconnectionPending_.compare_exchange_strong(expectedState, true)) {
-        LOG_DEBUG(getName() << "Ignoring reconnection attempt since there's already a pending reconnection");
+        LOG_WARN(getName() << "Ignoring reconnection attempt since there's already a pending reconnection");
         return;
     }
 
-    LOG_INFO(getName() << "Getting connection from pool");
+    LOG_WARN(getName() << "Getting connection from pool");
     ClientImplPtr client = client_.lock();
     if (!client) {
         LOG_WARN(getName() << "Client is invalid when calling grabCnx()");
         connectionFailed(ResultConnectError);
         return;
     }
     auto weakSelf = get_weak_from_this();
+    auto name = getName();
     client->getConnection(*topic_).addListener(
-        [this, weakSelf](Result result, const ClientConnectionPtr& cnx) {
+        [this, weakSelf, name](Result result, const ClientConnectionPtr& cnx) {
             auto self = weakSelf.lock();
             if (!self) {
-                LOG_DEBUG("HandlerBase Weak reference is not valid anymore");
+                LOG_WARN(name << " HandlerBase Weak reference is not valid anymore: " << result);
                 return;
             }
 
             reconnectionPending_ = false;
 
             if (result == ResultOk) {
-                LOG_DEBUG(getName() << "Connected to broker: " << cnx->cnxString());
+                LOG_WARN(getName() << "Connected to broker: " << cnx->cnxString());
                 connectionOpened(cnx);
             } else {
                 connectionFailed(result);
@@ -116,6 +117,7 @@ void HandlerBase::handleDisconnection(Result result, const ClientConnectionPtr&
     }
 
     resetCnx();
+    LOG_WARN(getName() << " disconnected (result: " << result << ", state: " << state << ")");
 
     if (result == ResultRetryable) {
         scheduleReconnection();
@@ -144,7 +146,7 @@ void HandlerBase::scheduleReconnection() {
     if (state == Pending || state == Ready) {
         TimeDuration delay = backoff_.next();
 
-        LOG_INFO(getName() << "Schedule reconnection in " << (delay.total_milliseconds() / 1000.0) << " s");
+        LOG_WARN(getName() << "Schedule reconnection in " << (delay.total_milliseconds() / 1000.0) << " s");
         timer_->expires_from_now(delay);
         // passing shared_ptr here since time_ will get destroyed, so tasks will be cancelled
         // so we will not run into the case where grabCnx is invoked on out of scope handler

lib/RetryableOperation.h

@@ -108,27 +108,30 @@ class RetryableOperation : public std::enable_shared_from_this<RetryableOperatio
             timer_->expires_from_now(delay);
 
             auto nextRemainingTime = remainingTime - delay;
-            LOG_INFO("Reschedule " << name_ << " for " << delay.total_milliseconds()
+            LOG_WARN("Reschedule " << name_ << " for " << delay.total_milliseconds()
                                    << " ms, remaining time: " << nextRemainingTime.total_milliseconds()
                                    << " ms");
-            timer_->async_wait([this, weakSelf, nextRemainingTime](const boost::system::error_code& ec) {
-                auto self = weakSelf.lock();
-                if (!self) {
-                    return;
-                }
-                if (ec) {
-                    if (ec == boost::asio::error::operation_aborted) {
-                        LOG_DEBUG("Timer for " << name_ << " is cancelled");
-                        promise_.setFailed(ResultTimeout);
+            auto name = name_;
+            timer_->async_wait(
+                [this, weakSelf, nextRemainingTime, name](const boost::system::error_code& ec) {
+                    auto self = weakSelf.lock();
+                    if (!self) {
+                        LOG_WARN(name << " is expired in the callback");
+                        return;
+                    }
+                    if (ec) {
+                        if (ec == boost::asio::error::operation_aborted) {
+                            LOG_WARN("Timer for " << name_ << " is cancelled");
+                            promise_.setFailed(ResultTimeout);
+                        } else {
+                            LOG_WARN("Timer for " << name_ << " failed: " << ec.message());
+                        }
                     } else {
-                        LOG_WARN("Timer for " << name_ << " failed: " << ec.message());
+                        LOG_DEBUG("Run operation " << name_ << ", remaining time: "
+                                                   << nextRemainingTime.total_milliseconds() << " ms");
+                        runImpl(nextRemainingTime);
                     }
-                } else {
-                    LOG_DEBUG("Run operation " << name_ << ", remaining time: "
-                                               << nextRemainingTime.total_milliseconds() << " ms");
-                    runImpl(nextRemainingTime);
-                }
-            });
+                });
         });
         return promise_.getFuture();
     }