[improve][client] Enhance connection and timeout logging (#539)

Author: zhanglistar

lib/ClientConnection.cc

@@ -56,6 +56,16 @@ using namespace ASIO::ip;
 
 namespace pulsar {
 
+namespace {
+static std::ostream& operator<<(std::ostream& os, const tcp::resolver::results_type& results) {
+    for (const auto& entry : results) {
+        const auto& ep = entry.endpoint();
+        os << ep.address().to_string() << ":" << ep.port() << " ";
+    }
+    return os;
+}
+}  // anonymous namespace
+
 using proto::BaseCommand;
 
 static const uint32_t DefaultBufferSize = 64 * 1024;
@@ -486,7 +496,7 @@ void ClientConnection::handleTcpConnected(const ASIO_ERROR& err, const tcp::endp
             handleHandshake(ASIO_SUCCESS);
         }
     } else {
-        LOG_ERROR(cnxString_ << "Failed to establish connection: " << err.message());
+        LOG_ERROR(cnxString_ << "Failed to establish connection to " << endpoint << ": " << err.message());
         if (err == ASIO::error::operation_aborted) {
             close();
         } else {
@@ -603,16 +613,25 @@ void ClientConnection::handleResolve(ASIO_ERROR err, const tcp::resolver::result
         return;
     }
 
+    if (!results.empty()) {
+        LOG_DEBUG(cnxString_ << "Resolved " << results.size() << " endpoints");
+        for (const auto& entry : results) {
+            const auto& ep = entry.endpoint();
+            LOG_DEBUG(cnxString_ << "  " << ep.address().to_string() << ":" << ep.port());
+        }
+    }
+
     auto weakSelf = weak_from_this();
-    connectTimeoutTask_->setCallback([weakSelf](const PeriodicTask::ErrorCode& ec) {
+    connectTimeoutTask_->setCallback([weakSelf, results = tcp::resolver::results_type(results)](
+                                         const PeriodicTask::ErrorCode& ec) {
         ClientConnectionPtr ptr = weakSelf.lock();
         if (!ptr) {
-            // Connection was already destroyed
+            LOG_DEBUG("Connect timeout callback skipped: connection was already destroyed");
             return;
         }
 
         if (ptr->state_ != Ready) {
-            LOG_ERROR(ptr->cnxString_ << "Connection was not established in "
+            LOG_ERROR(ptr->cnxString_ << "Connection to " << results << " was not established in "
                                       << ptr->connectTimeoutTask_->getPeriodMs() << " ms, close the socket");
             PeriodicTask::ErrorCode err;
             ptr->socket_->close(err);
@@ -1212,20 +1231,23 @@ Future<Result, ResponseData> ClientConnection::sendRequestWithId(const SharedBuf
 void ClientConnection::handleRequestTimeout(const ASIO_ERROR& ec,
                                             const PendingRequestData& pendingRequestData) {
     if (!ec && !pendingRequestData.hasGotResponse->load()) {
+        LOG_WARN(cnxString_ << "Network request timeout to broker, remote: " << physicalAddress_);
         pendingRequestData.promise.setFailed(ResultTimeout);
     }
 }
 
 void ClientConnection::handleLookupTimeout(const ASIO_ERROR& ec,
                                            const LookupRequestData& pendingRequestData) {
     if (!ec) {
+        LOG_WARN(cnxString_ << "Lookup request timeout to broker, remote: " << physicalAddress_);
         pendingRequestData.promise->setFailed(ResultTimeout);
     }
 }
 
 void ClientConnection::handleGetLastMessageIdTimeout(const ASIO_ERROR& ec,
                                                      const ClientConnection::LastMessageIdRequestData& data) {
     if (!ec) {
+        LOG_WARN(cnxString_ << "GetLastMessageId request timeout to broker, remote: " << physicalAddress_);
         data.promise->setFailed(ResultTimeout);
     }
 }

lib/ConsumerImpl.cc

@@ -1104,6 +1104,14 @@ Result ConsumerImpl::receiveHelper(Message& msg, int timeout) {
         if (state_ != Ready) {
             return ResultAlreadyClosed;
         }
+        auto cnx = getCnx().lock();
+        if (cnx) {
+            LOG_WARN(getName() << " Receive timeout after " << timeout << " ms, connection: "
+                               << cnx->cnxString() << ", queue size: " << incomingMessages_.size());
+        } else {
+            LOG_WARN(getName() << " Receive timeout after " << timeout
+                               << " ms, no connection, queue size: " << incomingMessages_.size());
+        }
         return ResultTimeout;
     }
 }

lib/MultiTopicsConsumerImpl.cc

@@ -21,6 +21,7 @@
 #include <chrono>
 #include <stdexcept>
 
+#include "ClientConnection.h"
 #include "ClientImpl.h"
 #include "ConsumerImpl.h"
 #include "ExecutorService.h"
@@ -600,6 +601,14 @@ Result MultiTopicsConsumerImpl::receive(Message& msg, int timeout) {
         if (state_ != Ready) {
             return ResultAlreadyClosed;
         }
+        auto cnx = getCnx().lock();
+        if (cnx) {
+            LOG_WARN(getName() << " Receive timeout after " << timeout << " ms, connection: "
+                               << cnx->cnxString() << ", queue size: " << incomingMessages_.size());
+        } else {
+            LOG_WARN(getName() << " Receive timeout after " << timeout
+                               << " ms, no connection, queue size: " << incomingMessages_.size());
+        }
         return ResultTimeout;
     }
 }

lib/ProducerImpl.cc

@@ -868,6 +868,15 @@ void ProducerImpl::handleSendTimeout(const ASIO_ERROR& err) {
     }
 
     lock.unlock();
+    auto cnx = getCnx().lock();
+    if (cnx) {
+        LOG_WARN(getName() << "Send timeout due to queueing delay, connection: " << cnx->cnxString()
+                           << ", pending messages: " << pendingMessages.size()
+                           << ", queue size: " << pendingMessagesQueue_.size());
+    } else {
+        LOG_WARN(getName() << "Send timeout due to queueing delay, no connection, pending messages: "
+                           << pendingMessages.size() << ", queue size: " << pendingMessagesQueue_.size());
+    }
     for (const auto& op : pendingMessages) {
         op->complete(ResultTimeout, {});
     }

lib/UnAckedMessageTrackerEnabled.cc

@@ -20,6 +20,7 @@
 
 #include <functional>
 
+#include "ClientConnection.h"
 #include "ClientImpl.h"
 #include "ConsumerImplBase.h"
 #include "ExecutorService.h"
@@ -57,9 +58,16 @@ void UnAckedMessageTrackerEnabled::timeoutHandlerHelper() {
 
     std::set<MessageId> msgIdsToRedeliver;
     if (!headPartition.empty()) {
-        LOG_INFO(consumerReference_.getName().c_str()
-                 << ": " << headPartition.size() << " Messages were not acked within "
-                 << timePartitions.size() * tickDurationInMs_ << " time");
+        auto cnx = consumerReference_.getCnx().lock();
+        if (cnx) {
+            LOG_WARN(consumerReference_.getName()
+                     << " Unacked messages timeout: " << headPartition.size() << " messages not acked within "
+                     << timeoutMs_ << " ms, connection: " << cnx->cnxString());
+        } else {
+            LOG_WARN(consumerReference_.getName()
+                     << " Unacked messages timeout: " << headPartition.size() << " messages not acked within "
+                     << timeoutMs_ << " ms, no connection");
+        }
         for (auto it = headPartition.begin(); it != headPartition.end(); it++) {
             msgIdsToRedeliver.insert(*it);
             messageIdPartitionMap.erase(*it);