[fix] Avoid resource leakage of AckGroupingTracker (#185)

Author: erobot

lib/AckGroupingTrackerEnabled.cc

@@ -51,6 +51,7 @@ AckGroupingTrackerEnabled::AckGroupingTrackerEnabled(ClientImplPtr clientPtr,
                                                      const HandlerBasePtr& handlerPtr, uint64_t consumerId,
                                                      long ackGroupingTimeMs, long ackGroupingMaxSize)
     : AckGroupingTracker(),
+      isClosed_(false),
       handlerWeakPtr_(handlerPtr),
       consumerId_(consumerId),
       nextCumulativeAckMsgId_(MessageId::earliest()),
@@ -110,6 +111,7 @@ void AckGroupingTrackerEnabled::addAcknowledgeCumulative(const MessageId& msgId)
 }
 
 void AckGroupingTrackerEnabled::close() {
+    isClosed_ = true;
     this->flush();
     std::lock_guard<std::mutex> lock(this->mutexTimer_);
     if (this->timer_) {
@@ -164,6 +166,10 @@ void AckGroupingTrackerEnabled::flushAndClean() {
 }
 
 void AckGroupingTrackerEnabled::scheduleTimer() {
+    if (isClosed_) {
+        return;
+    }
+
     std::lock_guard<std::mutex> lock(this->mutexTimer_);
     this->timer_ = this->executor_->createDeadlineTimer();
     this->timer_->expires_from_now(boost::posix_time::milliseconds(std::max(1L, this->ackGroupingTimeMs_)));

lib/AckGroupingTrackerEnabled.h

@@ -21,6 +21,7 @@
 
 #include <pulsar/MessageId.h>
 
+#include <atomic>
 #include <boost/asio/deadline_timer.hpp>
 #include <cstdint>
 #include <mutex>
@@ -71,6 +72,9 @@ class AckGroupingTrackerEnabled : public AckGroupingTracker {
     //! Method for scheduling grouping timer.
     void scheduleTimer();
 
+    //! State
+    std::atomic_bool isClosed_;
+
     //! The connection handler.
     HandlerBaseWeakPtr handlerWeakPtr_;
 

lib/ConsumerImpl.cc

@@ -1212,6 +1212,9 @@ void ConsumerImpl::closeAsync(ResultCallback originalCallback) {
 const std::string& ConsumerImpl::getName() const { return consumerStr_; }
 
 void ConsumerImpl::shutdown() {
+    if (ackGroupingTrackerPtr_) {
+        ackGroupingTrackerPtr_->close();
+    }
     incomingMessages_.clear();
     possibleSendToDeadLetterTopicMessages_.clear();
     resetCnx();