[GitHub] sijie closed pull request #2087: Make ack timeout work in asynchronous implementation of C++ consumer

[GitBox <gi...@apache.org>]

sijie closed pull request #2087: Make ack timeout work in asynchronous implementation of C++ consumer
URL: https://github.com/apache/incubator-pulsar/pull/2087
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git a/pulsar-client-cpp/lib/ConsumerImpl.cc b/pulsar-client-cpp/lib/ConsumerImpl.cc
index df5cae5dd2..f0a9cca721 100644
--- a/pulsar-client-cpp/lib/ConsumerImpl.cc
+++ b/pulsar-client-cpp/lib/ConsumerImpl.cc
@@ -449,6 +449,7 @@ void ConsumerImpl::internalListener() {
         // This will only happen when the connection got reset and we cleared the queue
         return;
     }
+    unAckedMessageTrackerPtr_->add(msg.getMessageId());
     try {
         consumerStatsBasePtr_->receivedMessage(msg, ResultOk);
         messageListener_(Consumer(shared_from_this()), msg);
diff --git a/pulsar-client-cpp/tests/BasicEndToEndTest.cc b/pulsar-client-cpp/tests/BasicEndToEndTest.cc
index f3a9f1266f..023a58769d 100644
--- a/pulsar-client-cpp/tests/BasicEndToEndTest.cc
+++ b/pulsar-client-cpp/tests/BasicEndToEndTest.cc
@@ -22,6 +22,7 @@
 #include <lib/LogUtils.h>
 #include <pulsar/MessageBuilder.h>
 #include <lib/Commands.h>
+#include <lib/Latch.h>
 #include <sstream>
 #include "boost/date_time/posix_time/posix_time.hpp"
 #include "CustomRoutingPolicy.h"
@@ -50,6 +51,13 @@ static void messageListenerFunction(Consumer consumer, const Message& msg) {
     consumer.acknowledge(msg);
 }
 
+static void messageListenerFunctionWithoutAck(Consumer consumer, const Message& msg, Latch& latch,
+                                              const std::string& content) {
+    globalCount++;
+    ASSERT_EQ(content, msg.getDataAsString());
+    latch.countdown();
+}
+
 static void sendCallBack(Result r, const Message& msg, std::string prefix) {
     ASSERT_EQ(r, ResultOk);
     std::string messageContent = prefix + boost::lexical_cast<std::string>(globalTestBatchMessagesCounter++);
@@ -1427,3 +1435,76 @@ TEST(BasicEndToEndTest, testSeek) {
     ASSERT_EQ(ResultOk, producer.close());
     ASSERT_EQ(ResultOk, client.close());
 }
+
+TEST(BasicEndToEndTest, testUnAckedMessageTimeout) {
+    Client client(lookupUrl);
+    std::string topicName = "persistent://prop/unit/ns1/testUnAckedMessageTimeout";
+    std::string subName = "my-sub-name";
+    std::string content = "msg-content";
+
+    Producer producer;
+    Result result = client.createProducer(topicName, producer);
+    ASSERT_EQ(ResultOk, result);
+
+    Consumer consumer;
+    ConsumerConfiguration consConfig;
+    consConfig.setUnAckedMessagesTimeoutMs(10 * 1000);
+    result = client.subscribe(topicName, subName, consConfig, consumer);
+    ASSERT_EQ(ResultOk, result);
+
+    Message msg = MessageBuilder().setContent(content).build();
+    result = producer.send(msg);
+    ASSERT_EQ(ResultOk, result);
+
+    Message receivedMsg1;
+    MessageId msgId1;
+    consumer.receive(receivedMsg1);
+    msgId1 = receivedMsg1.getMessageId();
+    ASSERT_EQ(content, receivedMsg1.getDataAsString());
+
+    Message receivedMsg2;
+    MessageId msgId2;
+    consumer.receive(receivedMsg2, 30 * 1000);
+    msgId2 = receivedMsg2.getMessageId();
+    ASSERT_EQ(content, receivedMsg2.getDataAsString());
+
+    ASSERT_EQ(msgId1, msgId2);
+
+    consumer.unsubscribe();
+    consumer.close();
+    producer.close();
+    client.close();
+}
+
+TEST(BasicEndToEndTest, testUnAckedMessageTimeoutListener) {
+    Client client(lookupUrl);
+    std::string topicName = "persistent://prop/unit/ns1/testUnAckedMessageTimeoutListener";
+    std::string subName = "my-sub-name";
+    std::string content = "msg-content";
+
+    Producer producer;
+    Result result = client.createProducer(topicName, producer);
+    ASSERT_EQ(ResultOk, result);
+
+    Consumer consumer;
+    ConsumerConfiguration consConfig;
+    consConfig.setUnAckedMessagesTimeoutMs(10 * 1000);
+    Latch latch(2);
+    consConfig.setMessageListener(boost::bind(messageListenerFunctionWithoutAck, _1, _2, latch, content));
+    result = client.subscribe(topicName, subName, consConfig, consumer);
+    ASSERT_EQ(ResultOk, result);
+
+    globalCount = 0;
+
+    Message msg = MessageBuilder().setContent(content).build();
+    result = producer.send(msg);
+    ASSERT_EQ(ResultOk, result);
+
+    ASSERT_TRUE(latch.wait(milliseconds(30 * 1000)));
+    ASSERT_GE(globalCount, 2);
+
+    consumer.unsubscribe();
+    consumer.close();
+    producer.close();
+    client.close();
+}


 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services