Fix the flush callback might be called repeatedly (#353)

Fixes #352

### Motivation

#303 adds the flush
callback to the last `OpSendMsg` instead of adding to the batch message
container. However, `batchMessageAndSend` will create an `OpSendMsg` and
add it to the `pendingMessagesQueue_`.

https://github.com/apache/pulsar-client-cpp/blob/7bb94f45b917ed30b5302ac93ffa1f1942fc6313/lib/ProducerImpl.cc#L384-L389

In the code above, `pendingMessagesQueue_` could never be empty and the
callback will be added again by `opSendMsg->addTrackerCallback`. The 1st
time it's added in `createOpSendMsg` or `createOpSendMsgs` called by
`batchMessageAndSend`.

### Motivation

Add the callback to the last `OpSendMsg only when the batch message
container is empty.

In `testFlushBatch`, replace the `flush` call with the `flushAsync` call
and verify the callback is only called once after it's completed.

(cherry picked from commit 37ea769)

Author: BewareMyPower

lib/BatchMessageKeyBasedContainer.cc

@@ -77,11 +77,16 @@ std::vector<std::unique_ptr<OpSendMsg>> BatchMessageKeyBasedContainer::createOpS
     // Store raw pointers to use std::sort
     std::vector<OpSendMsg*> rawOpSendMsgs;
     for (auto& kv : batches_) {
-        rawOpSendMsgs.emplace_back(createOpSendMsgHelper(kv.second).release());
+        if (!kv.second.empty()) {
+            rawOpSendMsgs.emplace_back(createOpSendMsgHelper(kv.second).release());
+        }
     }
     std::sort(rawOpSendMsgs.begin(), rawOpSendMsgs.end(), [](const OpSendMsg* lhs, const OpSendMsg* rhs) {
         return lhs->sendArgs->sequenceId < rhs->sendArgs->sequenceId;
     });
+    if (rawOpSendMsgs.empty()) {
+        return {};
+    }
     rawOpSendMsgs.back()->addTrackerCallback(flushCallback);
 
     std::vector<std::unique_ptr<OpSendMsg>> opSendMsgs{rawOpSendMsgs.size()};

lib/ProducerImpl.cc

@@ -376,29 +376,37 @@ void ProducerImpl::setMessageMetadata(const Message& msg, const uint64_t& sequen
 
 void ProducerImpl::flushAsync(FlushCallback callback) {
     if (state_ != Ready) {
-        callback(ResultAlreadyClosed);
+        if (callback) {
+            callback(ResultAlreadyClosed);
+        }
         return;
     }
+
+    auto addCallbackToLastOp = [this, &callback] {
+        if (pendingMessagesQueue_.empty()) {
+            return false;
+        }
+        pendingMessagesQueue_.back()->addTrackerCallback(callback);
+        return true;
+    };
+
     if (batchMessageContainer_) {
         Lock lock(mutex_);
-        auto failures = batchMessageAndSend(callback);
-        if (!pendingMessagesQueue_.empty()) {
-            auto& opSendMsg = pendingMessagesQueue_.back();
-            lock.unlock();
-            failures.complete();
-            opSendMsg->addTrackerCallback(callback);
-        } else {
-            lock.unlock();
-            failures.complete();
-            callback(ResultOk);
+
+        if (batchMessageContainer_->isEmpty()) {
+            if (!addCallbackToLastOp() && callback) {
+                lock.unlock();
+                callback(ResultOk);
+            }
+            return;
         }
+
+        auto failures = batchMessageAndSend(callback);
+        lock.unlock();
+        failures.complete();
     } else {
         Lock lock(mutex_);
-        if (!pendingMessagesQueue_.empty()) {
-            auto& opSendMsg = pendingMessagesQueue_.back();
-            lock.unlock();
-            opSendMsg->addTrackerCallback(callback);
-        } else {
+        if (!addCallbackToLastOp() && callback) {
             lock.unlock();
             callback(ResultOk);
         }

tests/ProducerTest.cc

@@ -476,7 +476,23 @@ TEST_P(ProducerTest, testFlushBatch) {
         producer.sendAsync(msg, cb);
     }
 
-    producer.flush();
+    auto assertFlushCallbackOnce = [&producer] {
+        Latch latch{1};
+        std::mutex mutex;
+        std::vector<Result> results;
+        producer.flushAsync([&](Result result) {
+            {
+                std::lock_guard<std::mutex> lock{mutex};
+                results.emplace_back(result);
+            }
+            latch.countdown();
+        });
+        latch.wait();
+        std::lock_guard<std::mutex> lock{mutex};
+        ASSERT_EQ(results, (std::vector<Result>{ResultOk}));
+    };
+
+    assertFlushCallbackOnce();
     ASSERT_EQ(needCallBack.load(), 0);
     producer.close();
 
@@ -494,7 +510,7 @@ TEST_P(ProducerTest, testFlushBatch) {
         producer.sendAsync(msg, cb2);
     }
 
-    producer.flush();
+    assertFlushCallbackOnce();
     ASSERT_EQ(needCallBack2.load(), 0);
     producer.close();