BewareMyPower · BewareMyPower · Mar 24, 2023 · Mar 16, 2023 · Mar 28, 2023 · Mar 28, 2023
diff --git a/include/pulsar/ConsumerConfiguration.h b/include/pulsar/ConsumerConfiguration.h
@@ -629,6 +629,22 @@ class PULSAR_PUBLIC ConsumerConfiguration {
 
     const std::vector<ConsumerInterceptorPtr>& getInterceptors() const;
 
+    /**
+     * Whether to receive the ACK receipt from broker.
+     *
+     * By default, when Consumer::acknowledge is called, it won't wait until the corresponding response from
+     * broker. After it's enabled, the `acknowledge` method will return a Result that indicates if the
+     * acknowledgment succeeded.
+     *
+     * Default: false
+     */
+    ConsumerConfiguration& setAckReceiptEnabled(bool ackReceiptEnabled);
+
+    /**
+     * The associated getter of setAckReceiptEnabled.
+     */
+    bool isAckReceiptEnabled() const;
+
     friend class PulsarWrapper;
     friend class PulsarFriend;
 

diff --git a/lib/AckGroupingTracker.cc b/lib/AckGroupingTracker.cc
@@ -19,6 +19,9 @@
 
 #include "AckGroupingTracker.h"
 
+#include <atomic>
+#include <limits>
+
 #include "BitSet.h"
 #include "ClientConnection.h"
 #include "Commands.h"
@@ -29,24 +32,33 @@ namespace pulsar {
 
 DECLARE_LOG_OBJECT();
 
-inline void sendAck(ClientConnectionPtr cnx, uint64_t consumerId, const MessageId& msgId,
-                    CommandAck_AckType ackType) {
-    const auto& bitSet = Commands::getMessageIdImpl(msgId)->getBitSet();
-    auto cmd = Commands::newAck(consumerId, msgId.ledgerId(), msgId.entryId(), bitSet, ackType, -1);
-    cnx->sendCommand(cmd);
-    LOG_DEBUG("ACK request is sent for message - [" << msgId.ledgerId() << ", " << msgId.entryId() << "]");
-}
-
-bool AckGroupingTracker::doImmediateAck(ClientConnectionWeakPtr connWeakPtr, uint64_t consumerId,
-                                        const MessageId& msgId, CommandAck_AckType ackType) {
-    auto cnx = connWeakPtr.lock();
-    if (cnx == nullptr) {
-        LOG_DEBUG("Connection is not ready, ACK failed for message - [" << msgId.ledgerId() << ", "
-                                                                        << msgId.entryId() << "]");
-        return false;
+void AckGroupingTracker::doImmediateAck(const MessageId& msgId, ResultCallback callback,
+                                        CommandAck_AckType ackType) const {
+    const auto cnx = connectionSupplier_();
+    if (!cnx) {
+        LOG_DEBUG("Connection is not ready, ACK failed for " << msgId);
+        if (callback) {
+            callback(ResultAlreadyClosed);
+        }
+        return;
+    }
+    const auto& ackSet = Commands::getMessageIdImpl(msgId)->getBitSet();
+    if (waitResponse_) {
+        const auto requestId = requestIdSupplier_();
+        cnx->sendRequestWithId(
+               Commands::newAck(consumerId_, msgId.ledgerId(), msgId.entryId(), ackSet, ackType, requestId),
+               requestId)
+            .addListener([callback](Result result, const ResponseData&) {
+                if (callback) {
+                    callback(result);
+                }
+            });
+    } else {
+        cnx->sendCommand(Commands::newAck(consumerId_, msgId.ledgerId(), msgId.entryId(), ackSet, ackType));
+        if (callback) {
+            callback(ResultOk);
+        }
     }
-    sendAck(cnx, consumerId, msgId, ackType);
-    return true;
 }
 
 static std::ostream& operator<<(std::ostream& os, const std::set<MessageId>& msgIds) {
@@ -62,25 +74,42 @@ static std::ostream& operator<<(std::ostream& os, const std::set<MessageId>& msg
     return os;
 }
 
-bool AckGroupingTracker::doImmediateAck(ClientConnectionWeakPtr connWeakPtr, uint64_t consumerId,
-                                        const std::set<MessageId>& msgIds) {
-    auto cnx = connWeakPtr.lock();
-    if (cnx == nullptr) {
-        LOG_DEBUG("Connection is not ready, ACK failed.");
-        return false;
+void AckGroupingTracker::doImmediateAck(const std::set<MessageId>& msgIds, ResultCallback callback) const {
+    const auto cnx = connectionSupplier_();
+    if (!cnx) {
+        LOG_DEBUG("Connection is not ready, ACK failed for " << msgIds);
+        if (callback) {
+            callback(ResultAlreadyClosed);
+        }
+        return;
     }
 
     if (Commands::peerSupportsMultiMessageAcknowledgement(cnx->getServerProtocolVersion())) {
-        auto cmd = Commands::newMultiMessageAck(consumerId, msgIds);
-        cnx->sendCommand(cmd);
-        LOG_DEBUG("ACK request is sent for " << msgIds.size() << " messages: " << msgIds);
+        if (waitResponse_) {
+            const auto requestId = requestIdSupplier_();
+            cnx->sendRequestWithId(Commands::newMultiMessageAck(consumerId_, msgIds, requestId), requestId)
+                .addListener([callback](Result result, const ResponseData&) {
+                    if (callback) {
+                        callback(result);
+                    }
+                });
+        } else {
+            cnx->sendCommand(Commands::newMultiMessageAck(consumerId_, msgIds));
+            if (callback) {
+                callback(ResultOk);
+            }
+        }
     } else {
-        // Broker does not support multi-message ACK, use multiple individual ACKs instead.
-        for (const auto& msgId : msgIds) {
-            sendAck(cnx, consumerId, msgId, CommandAck_AckType_Individual);
+        auto count = std::make_shared<std::atomic<size_t>>(msgIds.size());
+        auto wrappedCallback = [callback, count](Result result) {
+            if (--*count == 0 && callback) {
+                callback(result);
+            }
+        };
+        for (auto&& msgId : msgIds) {
+            doImmediateAck(msgId, wrappedCallback, CommandAck_AckType_Individual);
         }
     }
-    return true;
 }
 
 }  // namespace pulsar
diff --git a/lib/AckGroupingTracker.h b/lib/AckGroupingTracker.h
@@ -20,16 +20,20 @@
 #define LIB_ACKGROUPINGTRACKER_H_
 
 #include <pulsar/MessageId.h>
+#include <pulsar/Result.h>
 
 #include <cstdint>
+#include <functional>
 #include <set>
 
 #include "ProtoApiEnums.h"
 
 namespace pulsar {
 
 class ClientConnection;
+using ClientConnectionPtr = std::shared_ptr<ClientConnection>;
 using ClientConnectionWeakPtr = std::weak_ptr<ClientConnection>;
+using ResultCallback = std::function<void(Result)>;
 
 /**
  * @class AckGroupingTracker
@@ -38,7 +42,13 @@ using ClientConnectionWeakPtr = std::weak_ptr<ClientConnection>;
  */
 class AckGroupingTracker : public std::enable_shared_from_this<AckGroupingTracker> {
    public:
-    AckGroupingTracker() = default;
+    AckGroupingTracker(std::function<ClientConnectionPtr()> connectionSupplier,
+                       std::function<uint64_t()> requestIdSupplier, uint64_t consumerId, bool waitResponse)
+        : connectionSupplier_(connectionSupplier),
+          requestIdSupplier_(requestIdSupplier),
+          consumerId_(consumerId),
+          waitResponse_(waitResponse) {}
+
     virtual ~AckGroupingTracker() = default;
 
     /**
@@ -59,20 +69,23 @@ class AckGroupingTracker : public std::enable_shared_from_this<AckGroupingTracke
     /**
      * Adding message ID into ACK group for individual ACK.
      * @param[in] msgId ID of the message to be ACKed.
+     * @param[in] callback the callback that is triggered when the message is acknowledged
      */
-    virtual void addAcknowledge(const MessageId& msgId) {}
+    virtual void addAcknowledge(const MessageId& msgId, ResultCallback callback) {}
 
     /**
      * Adding message ID list into ACK group for individual ACK.
      * @param[in] msgIds of the message to be ACKed.
+     * @param[in] callback the callback that is triggered when the messages are acknowledged
      */
-    virtual void addAcknowledgeList(const MessageIdList& msgIds) {}
+    virtual void addAcknowledgeList(const MessageIdList& msgIds, ResultCallback callback) {}
 
     /**
      * Adding message ID into ACK group for cumulative ACK.
      * @param[in] msgId ID of the message to be ACKed.
+     * @param[in] callback the callback that is triggered when the message is acknowledged
      */
-    virtual void addAcknowledgeCumulative(const MessageId& msgId) {}
+    virtual void addAcknowledgeCumulative(const MessageId& msgId, ResultCallback callback) {}
 
     /**
      * Flush all the pending grouped ACKs (as flush() does), and stop period ACKs sending.
@@ -91,27 +104,17 @@ class AckGroupingTracker : public std::enable_shared_from_this<AckGroupingTracke
     virtual void flushAndClean() {}
 
    protected:
-    /**
-     * Immediately send ACK request to broker.
-     * @param[in] connWeakPtr weak pointer of the client connection.
-     * @param[in] consumerId ID of the consumer that performs this ACK.
-     * @param[in] msgId message ID to be ACKed.
-     * @param[in] ackType ACK type, e.g. cumulative.
-     * @return true if the ACK is sent successfully, otherwise false.
-     */
-    bool doImmediateAck(ClientConnectionWeakPtr connWeakPtr, uint64_t consumerId, const MessageId& msgId,
-                        CommandAck_AckType ackType);
+    void doImmediateAck(const MessageId& msgId, ResultCallback callback, CommandAck_AckType ackType) const;
+    void doImmediateAck(const std::set<MessageId>& msgIds, ResultCallback callback) const;
+
+   private:
+    const std::function<ClientConnectionPtr()> connectionSupplier_;
+    const std::function<uint64_t()> requestIdSupplier_;
+    const uint64_t consumerId_;
+
+   protected:
+    const bool waitResponse_;
 
-    /**
-     * Immediately send a set of ACK requests one by one to the broker, it only supports individual
-     * ACK.
-     * @param[in] connWeakPtr weak pointer of the client connection.
-     * @param[in] consumerId ID of the consumer that performs this ACK.
-     * @param[in] msgIds message IDs to be ACKed.
-     * @return true if the ACK is sent successfully, otherwise false.
-     */
-    bool doImmediateAck(ClientConnectionWeakPtr connWeakPtr, uint64_t consumerId,
-                        const std::set<MessageId>& msgIds);
 };  // class AckGroupingTracker
 
 using AckGroupingTrackerPtr = std::shared_ptr<AckGroupingTracker>;

diff --git a/lib/AckGroupingTrackerDisabled.cc b/lib/AckGroupingTrackerDisabled.cc
@@ -20,32 +20,24 @@
 #include "AckGroupingTrackerDisabled.h"
 
 #include "HandlerBase.h"
-#include "LogUtils.h"
 #include "ProtoApiEnums.h"
 
 namespace pulsar {
 
-DECLARE_LOG_OBJECT();
-
-AckGroupingTrackerDisabled::AckGroupingTrackerDisabled(HandlerBase& handler, uint64_t consumerId)
-    : AckGroupingTracker(), handler_(handler), consumerId_(consumerId) {
-    LOG_INFO("ACK grouping is disabled.");
-}
-
-void AckGroupingTrackerDisabled::addAcknowledge(const MessageId& msgId) {
-    this->doImmediateAck(this->handler_.getCnx(), this->consumerId_, msgId, CommandAck_AckType_Individual);
+void AckGroupingTrackerDisabled::addAcknowledge(const MessageId& msgId, ResultCallback callback) {
+    doImmediateAck(msgId, callback, CommandAck_AckType_Individual);
 }
 
-void AckGroupingTrackerDisabled::addAcknowledgeList(const MessageIdList& msgIds) {
+void AckGroupingTrackerDisabled::addAcknowledgeList(const MessageIdList& msgIds, ResultCallback callback) {
     std::set<MessageId> msgIdSet;
     for (auto&& msgId : msgIds) {
         msgIdSet.emplace(msgId);
     }
-    this->doImmediateAck(this->handler_.getCnx(), this->consumerId_, msgIdSet);
+    doImmediateAck(msgIdSet, callback);
 }
 
-void AckGroupingTrackerDisabled::addAcknowledgeCumulative(const MessageId& msgId) {
-    this->doImmediateAck(this->handler_.getCnx(), this->consumerId_, msgId, CommandAck_AckType_Cumulative);
+void AckGroupingTrackerDisabled::addAcknowledgeCumulative(const MessageId& msgId, ResultCallback callback) {
+    doImmediateAck(msgId, callback, CommandAck_AckType_Cumulative);
 }
 
 }  // namespace pulsar
diff --git a/lib/AckGroupingTrackerDisabled.h b/lib/AckGroupingTrackerDisabled.h
@@ -34,25 +34,12 @@ class HandlerBase;
  */
 class AckGroupingTrackerDisabled : public AckGroupingTracker {
    public:
+    using AckGroupingTracker::AckGroupingTracker;
     virtual ~AckGroupingTrackerDisabled() = default;
 
-    /**
-     * Constructing ACK grouping tracker for peresistent topics that disabled ACK grouping.
-     * @param[in] handler the connection handler.
-     * @param[in] consumerId consumer ID that this tracker belongs to.
-     */
-    AckGroupingTrackerDisabled(HandlerBase& handler, uint64_t consumerId);
-
-    void addAcknowledge(const MessageId& msgId) override;
-    void addAcknowledgeList(const MessageIdList& msgIds) override;
-    void addAcknowledgeCumulative(const MessageId& msgId) override;
-
-   private:
-    //! The connection handler.
-    HandlerBase& handler_;
-
-    //! ID of the consumer that this tracker belongs to.
-    uint64_t consumerId_;
+    void addAcknowledge(const MessageId& msgId, ResultCallback callback) override;
+    void addAcknowledgeList(const MessageIdList& msgIds, ResultCallback callback) override;
+    void addAcknowledgeCumulative(const MessageId& msgId, ResultCallback callback) override;
 };  // class AckGroupingTrackerDisabled
 
 }  // namespace pulsar