[C++] Fixing use-after-free and constructor bugs in UnAckedMessageTra…

…ckerEnabled (#11630)

* [C++] Fix an off-by-one error in my original change from commit 83f0345.  This was caught by ASAN in my company's build system.

* [C++] Fixing use-after-free and constructor bugs in UnAckedMessageTrackerEnabled

This is very similar to a previous fix I submitted in commit 87ebe80.  It's the same basic problem, but this class isn't part of the HandlerBase hierarchy, so it needs an independent fix.  Essentially, when we create Boost ASIO timer objects from a connection pointer, they maintain a bare reference to the corresponding io_service object inside the connection object.  When the destructor runs, we need to destroy the timer *before* the connection object.  Keeping the correct order of these member variables is crucial to ensure we don't hit a use-after-free scenario.  This was crashing some of our service code in rare circumstances, and is easily caught with Valgrind or ASAN.

I also noticed a rather serious bug in one of the UnAckedMessageTrackerEnabled constructors: I believe the intent here was to use the c++11 "delegating constructors" feature, but I think it's written using the Java style, which doesn't work in C++ (see https://stackoverflow.com/questions/13961037/delegate-constructor-c). The semantics of the existing code would just create a new, separate UnAckedMessageTrackerEnabled on the stack in the constructor scope, then immediately destroy it!  I corrected the syntax to ensure this works correctly, and fixed up the other constructor to properly use C++ initializer list syntax.  Finally, I removed some dangerous c-style casts (which should *never* be used) in favor of C++ static_cast.

* [C++] Applied clang-format to previous change

* [C++] Apparently clang-format didn't like this one either....

pulsar-client-cpp/lib/ClientImpl.cc

@@ -52,7 +52,7 @@ namespace pulsar {
 
 static const char hexDigits[] = {'0', '1', '2', '3', '4', '5', '6', '7',
                                  '8', '9', 'a', 'b', 'c', 'd', 'e', 'f'};
-static std::uniform_int_distribution<> hexDigitsDist(0, sizeof(hexDigits));
+static std::uniform_int_distribution<> hexDigitsDist(0, sizeof(hexDigits) - 1);
 static std::mt19937 randomEngine =
     std::mt19937(std::chrono::high_resolution_clock::now().time_since_epoch().count());
 

pulsar-client-cpp/lib/UnAckedMessageTrackerEnabled.cc

@@ -69,20 +69,19 @@ void UnAckedMessageTrackerEnabled::timeoutHandlerHelper() {
 
 UnAckedMessageTrackerEnabled::UnAckedMessageTrackerEnabled(long timeoutMs, const ClientImplPtr client,
                                                            ConsumerImplBase& consumer)
-    : consumerReference_(consumer) {
-    UnAckedMessageTrackerEnabled(timeoutMs, timeoutMs, client, consumer);
-}
+    : UnAckedMessageTrackerEnabled(timeoutMs, timeoutMs, client, consumer) {}
 
 UnAckedMessageTrackerEnabled::UnAckedMessageTrackerEnabled(long timeoutMs, long tickDurationInMs,
                                                            const ClientImplPtr client,
                                                            ConsumerImplBase& consumer)
-    : consumerReference_(consumer) {
-    timeoutMs_ = timeoutMs;
-    tickDurationInMs_ = (timeoutMs >= tickDurationInMs) ? tickDurationInMs : timeoutMs;
-    client_ = client;
-
-    int blankPartitions = (int)std::ceil((double)timeoutMs_ / tickDurationInMs_);
-    for (int i = 0; i < blankPartitions + 1; i++) {
+    : consumerReference_(consumer),
+      client_(client),
+      timeoutMs_(timeoutMs),
+      tickDurationInMs_(timeoutMs >= tickDurationInMs ? tickDurationInMs : timeoutMs) {
+    const int blankPartitions =
+        static_cast<int>(std::ceil(static_cast<double>(timeoutMs_) / tickDurationInMs_)) + 1;
+
+    for (int i = 0; i < blankPartitions; i++) {
         std::set<MessageId> msgIds;
         timePartitions.push_back(msgIds);
     }

pulsar-client-cpp/lib/UnAckedMessageTrackerEnabled.h

@@ -44,9 +44,9 @@ class UnAckedMessageTrackerEnabled : public UnAckedMessageTrackerInterface {
     std::map<MessageId, std::set<MessageId>&> messageIdPartitionMap;
     std::deque<std::set<MessageId>> timePartitions;
     std::mutex lock_;
-    DeadlineTimerPtr timer_;
     ConsumerImplBase& consumerReference_;
     ClientImplPtr client_;
+    DeadlineTimerPtr timer_;  // DO NOT place this before client_!
     long timeoutMs_;
     long tickDurationInMs_;