[fix] callback of send batch message is error when flush (apache#303)

### Motivation

When sendAsync batch messages and then flush() and close(), found that the callback of message is ResultAlreadyClosed but not Ok. 

The root is if there is no batch messages remain in BatchMessageContainer when do flush,  the flush callback directly return Ok. Instead, it should return Ok until the lastSendOpFuture complete.  

### Modifications

1. fix the error code.
2. test two cases of flush batch messages.

Author: TakaHiR07

lib/BatchMessageContainerBase.cc

@@ -92,7 +92,7 @@ void BatchMessageContainerBase::processAndClear(
     std::function<void(Result, const OpSendMsg&)> opSendMsgCallback, FlushCallback flushCallback) {
     if (isEmpty()) {
         if (flushCallback) {
-            flushCallback(ResultOk);
+            // do nothing, flushCallback complete until the lastOpSend complete
         }
     } else {
         const auto numBatches = getNumBatches();

lib/ProducerImpl.cc

@@ -374,8 +374,16 @@ void ProducerImpl::flushAsync(FlushCallback callback) {
     if (batchMessageContainer_) {
         Lock lock(mutex_);
         auto failures = batchMessageAndSend(callback);
-        lock.unlock();
-        failures.complete();
+        if (!pendingMessagesQueue_.empty()) {
+            auto& opSendMsg = pendingMessagesQueue_.back();
+            lock.unlock();
+            failures.complete();
+            opSendMsg.addTrackerCallback(callback);
+        } else {
+            lock.unlock();
+            failures.complete();
+            callback(ResultOk);
+        }
     } else {
         Lock lock(mutex_);
         if (!pendingMessagesQueue_.empty()) {

tests/ProducerTest.cc

@@ -439,6 +439,68 @@ TEST_P(ProducerTest, testFlushNoBatch) {
     client.close();
 }
 
+TEST_P(ProducerTest, testFlushBatch) {
+    Client client(serviceUrl);
+
+    auto partitioned = GetParam();
+    const auto topicName = std::string("testFlushNoBatch") +
+                           (partitioned ? "partitioned-" : "-no-partitioned-") +
+                           std::to_string(time(nullptr));
+
+    if (partitioned) {
+        // call admin api to make it partitioned
+        std::string url = adminUrl + "admin/v2/persistent/public/default/" + topicName + "/partitions";
+        int res = makePutRequest(url, "5");
+        LOG_INFO("res = " << res);
+        ASSERT_FALSE(res != 204 && res != 409);
+    }
+
+    ProducerConfiguration producerConfiguration;
+    producerConfiguration.setBatchingEnabled(true);
+    producerConfiguration.setBatchingMaxMessages(10);
+    producerConfiguration.setBatchingMaxPublishDelayMs(1000);
+    producerConfiguration.setBatchingMaxAllowedSizeInBytes(4 * 1024 * 1024);
+
+    // test all messages in batch has been sent
+    Producer producer;
+    ASSERT_EQ(ResultOk, client.createProducer(topicName, producerConfiguration, producer));
+
+    std::atomic_int needCallBack(100);
+    auto cb = [&needCallBack](Result code, const MessageId& msgId) {
+        ASSERT_EQ(code, ResultOk);
+        needCallBack.fetch_sub(1);
+    };
+
+    for (int i = 0; i < 100; ++i) {
+        Message msg = MessageBuilder().setContent("content").build();
+        producer.sendAsync(msg, cb);
+    }
+
+    producer.flush();
+    ASSERT_EQ(needCallBack.load(), 0);
+    producer.close();
+
+    // test remain messages in batch not send
+    ASSERT_EQ(ResultOk, client.createProducer(topicName, producerConfiguration, producer));
+
+    std::atomic_int needCallBack2(105);
+    auto cb2 = [&needCallBack2](Result code, const MessageId& msgId) {
+        ASSERT_EQ(code, ResultOk);
+        needCallBack2.fetch_sub(1);
+    };
+
+    for (int i = 0; i < 105; ++i) {
+        Message msg = MessageBuilder().setContent("content").build();
+        producer.sendAsync(msg, cb2);
+    }
+
+    producer.flush();
+    ASSERT_EQ(needCallBack2.load(), 0);
+    producer.close();
+
+    client.close();
+}
+
 TEST(ProducerTest, testCloseSubProducerWhenFail) {
     Client client(serviceUrl);