Fix ack non-persistent topic will be blocked. (#240)

Author: shibd

lib/AckGroupingTracker.h

@@ -71,21 +71,25 @@ class AckGroupingTracker : public std::enable_shared_from_this<AckGroupingTracke
      * @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, ResultCallback callback) {}
+    virtual void addAcknowledge(const MessageId& msgId, ResultCallback callback) { callback(ResultOk); }
 
     /**
      * 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, ResultCallback callback) {}
+    virtual void addAcknowledgeList(const MessageIdList& msgIds, ResultCallback callback) {
+        callback(ResultOk);
+    }
 
     /**
      * 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, ResultCallback callback) {}
+    virtual void addAcknowledgeCumulative(const MessageId& msgId, ResultCallback callback) {
+        callback(ResultOk);
+    }
 
     /**
      * Flush all the pending grouped ACKs (as flush() does), and stop period ACKs sending.

tests/ConsumerTest.cc

@@ -1118,6 +1118,29 @@ TEST(ConsumerTest, testNegativeAcksTrackerClose) {
     client.close();
 }
 
+TEST(ConsumerTest, testAckNotPersistentTopic) {
+    Client client(lookupUrl);
+    auto topicName = "non-persistent://public/default/testAckNotPersistentTopic";
+
+    Consumer consumer;
+    client.subscribe(topicName, "test-sub", consumer);
+
+    Producer producer;
+    client.createProducer(topicName, producer);
+
+    for (int i = 0; i < 10; ++i) {
+        producer.send(MessageBuilder().setContent(std::to_string(i)).build());
+    }
+
+    Message msg;
+    for (int i = 0; i < 10; ++i) {
+        ASSERT_EQ(ResultOk, consumer.receive(msg));
+        ASSERT_EQ(ResultOk, consumer.acknowledge(msg));
+    }
+
+    client.close();
+}
+
 INSTANTIATE_TEST_CASE_P(Pulsar, ConsumerSeekTest, ::testing::Values(true, false));
 
 }  // namespace pulsar