Fix invalid memory access for UnAckedMessageTrackerEnabled and ConsumerStatsImpl

Author: BewareMyPower

lib/ConsumerImpl.cc

@@ -129,7 +129,8 @@ ConsumerImpl::ConsumerImpl(const ClientImplPtr& client, const std::string& topic
     unsigned int statsIntervalInSeconds = client->getClientConfig().getStatsIntervalInSeconds();
     if (statsIntervalInSeconds) {
         consumerStatsBasePtr_ = std::make_shared<ConsumerStatsImpl>(
-            consumerStr_, client->getIOExecutorProvider()->get(), statsIntervalInSeconds);
+            consumerStr_, client->getIOExecutorProvider()->get()->createDeadlineTimer(),
+            statsIntervalInSeconds);
     } else {
         consumerStatsBasePtr_ = std::make_shared<ConsumerStatsDisabled>();
     }

lib/UnAckedMessageTrackerEnabled.cc

@@ -32,7 +32,11 @@ namespace pulsar {
 
 void UnAckedMessageTrackerEnabled::timeoutHandler() {
     timeoutHandlerHelper();
-    ExecutorServicePtr executorService = client_->getIOExecutorProvider()->get();
+    auto client = client_.lock();
+    if (client == nullptr) {
+        return;
+    }
+    ExecutorServicePtr executorService = client->getIOExecutorProvider()->get();
     timer_ = executorService->createDeadlineTimer();
     timer_->expires_from_now(std::chrono::milliseconds(tickDurationInMs_));
     std::weak_ptr<UnAckedMessageTrackerEnabled> weakSelf{shared_from_this()};

lib/UnAckedMessageTrackerEnabled.h

@@ -33,6 +33,7 @@ namespace pulsar {
 class ClientImpl;
 class ConsumerImplBase;
 using ClientImplPtr = std::shared_ptr<ClientImpl>;
+using ClientImplWeakPtr = std::weak_ptr<ClientImpl>;
 
 class UnAckedMessageTrackerEnabled : public std::enable_shared_from_this<UnAckedMessageTrackerEnabled>,
                                      public UnAckedMessageTrackerInterface {
@@ -58,7 +59,7 @@ class UnAckedMessageTrackerEnabled : public std::enable_shared_from_this<UnAcked
     std::deque<std::set<MessageId>> timePartitions;
     std::recursive_mutex lock_;
     ConsumerImplBase& consumerReference_;
-    const ClientImplPtr& client_;
+    ClientImplWeakPtr client_;
     DeadlineTimerPtr timer_;  // DO NOT place this before client_!
     long timeoutMs_;
     long tickDurationInMs_;

lib/stats/ConsumerStatsImpl.cc

@@ -19,9 +19,6 @@
 
 #include "ConsumerStatsImpl.h"
 
-#include <functional>
-
-#include "lib/ExecutorService.h"
 #include "lib/LogUtils.h"
 #include "lib/Utils.h"
 
@@ -30,11 +27,9 @@ DECLARE_LOG_OBJECT();
 
 using Lock = std::unique_lock<std::mutex>;
 
-ConsumerStatsImpl::ConsumerStatsImpl(const std::string& consumerStr, const ExecutorServicePtr& executor,
+ConsumerStatsImpl::ConsumerStatsImpl(const std::string& consumerStr, DeadlineTimerPtr timer,
                                      unsigned int statsIntervalInSeconds)
-    : consumerStr_(consumerStr),
-      timer_(executor->createDeadlineTimer()),
-      statsIntervalInSeconds_(statsIntervalInSeconds) {}
+    : consumerStr_(consumerStr), timer_(std::move(timer)), statsIntervalInSeconds_(statsIntervalInSeconds) {}
 
 ConsumerStatsImpl::ConsumerStatsImpl(const ConsumerStatsImpl& stats)
     : consumerStr_(stats.consumerStr_),

lib/stats/ConsumerStatsImpl.h

@@ -27,12 +27,8 @@
 
 #include "ConsumerStatsBase.h"
 #include "lib/AsioTimer.h"
-#include "lib/ExecutorService.h"
 namespace pulsar {
 
-class ExecutorService;
-using ExecutorServicePtr = std::shared_ptr<ExecutorService>;
-
 class ConsumerStatsImpl : public std::enable_shared_from_this<ConsumerStatsImpl>, public ConsumerStatsBase {
    private:
     std::string consumerStr_;
@@ -55,7 +51,7 @@ class ConsumerStatsImpl : public std::enable_shared_from_this<ConsumerStatsImpl>
     friend class PulsarFriend;
 
    public:
-    ConsumerStatsImpl(const std::string&, const ExecutorServicePtr&, unsigned int);
+    ConsumerStatsImpl(const std::string&, DeadlineTimerPtr, unsigned int);
     ConsumerStatsImpl(const ConsumerStatsImpl& stats);
     void flushAndReset(const ASIO_ERROR&);
     void start() override;