Fix AckGroupingTracker not flushed during close

Author: BewareMyPower

lib/AckGroupingTrackerEnabled.cc

@@ -20,6 +20,7 @@
 #include "AckGroupingTrackerEnabled.h"
 
 #include <climits>
+#include <memory>
 #include <mutex>
 
 #include "ClientConnection.h"
@@ -172,9 +173,10 @@ void AckGroupingTrackerEnabled::scheduleTimer() {
     std::lock_guard<std::mutex> lock(this->mutexTimer_);
     this->timer_ = this->executor_->createDeadlineTimer();
     this->timer_->expires_from_now(std::chrono::milliseconds(std::max(1L, this->ackGroupingTimeMs_)));
-    auto self = shared_from_this();
-    this->timer_->async_wait([this, self](const ASIO_ERROR& ec) -> void {
-        if (!ec) {
+    std::weak_ptr<AckGroupingTracker> weakSelf = shared_from_this();
+    this->timer_->async_wait([this, weakSelf](const ASIO_ERROR& ec) -> void {
+        auto self = weakSelf.lock();
+        if (self && !ec) {
             this->flush();
             this->scheduleTimer();
         }

lib/ConsumerImpl.cc

@@ -1331,6 +1331,11 @@ void ConsumerImpl::closeAsync(const ResultCallback& originalCallback) {
     incomingMessages_.close();
 
     // Flush pending grouped ACK requests.
+    if (ackGroupingTrackerPtr_.use_count() != 1) {
+        LOG_ERROR("AckGroupingTracker is shared by other "
+                  << (ackGroupingTrackerPtr_.use_count() - 1)
+                  << " threads, which will prevent flushing the ACKs");
+    }
     ackGroupingTrackerPtr_.reset();
     negativeAcksTracker_->close();