add log on timeout in consumer side

Author: zhanglistar

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/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);