From b7b678fe6c145d36186c8006bc86ce4d9d826ee9 Mon Sep 17 00:00:00 2001 From: saosir Date: Mon, 11 Jan 2021 15:09:09 +0800 Subject: [PATCH 1/2] [C++] UnAckedMessageTracker handles batch message incorrectly --- pulsar-client-cpp/lib/ConsumerImpl.h | 1 + .../lib/UnAckedMessageTrackerEnabled.cc | 16 +++-- .../lib/UnAckedMessageTrackerEnabled.h | 5 +- pulsar-client-cpp/tests/ConsumerTest.cc | 69 +++++++++++++++++++ 4 files changed, 82 insertions(+), 9 deletions(-) diff --git a/pulsar-client-cpp/lib/ConsumerImpl.h b/pulsar-client-cpp/lib/ConsumerImpl.h index ef60b946ee7a1..313f1d9eabed0 100644 --- a/pulsar-client-cpp/lib/ConsumerImpl.h +++ b/pulsar-client-cpp/lib/ConsumerImpl.h @@ -223,6 +223,7 @@ class ConsumerImpl : public ConsumerImplBase, FRIEND_TEST(ConsumerTest, testPartitionedConsumerUnAckedMessageRedelivery); FRIEND_TEST(ConsumerTest, testMultiTopicsConsumerUnAckedMessageRedelivery); + FRIEND_TEST(ConsumerTest, testBatchUnAckedMessageTracker); }; } /* namespace pulsar */ diff --git a/pulsar-client-cpp/lib/UnAckedMessageTrackerEnabled.cc b/pulsar-client-cpp/lib/UnAckedMessageTrackerEnabled.cc index 96dfd6c5c487c..3364c0eff9db8 100644 --- a/pulsar-client-cpp/lib/UnAckedMessageTrackerEnabled.cc +++ b/pulsar-client-cpp/lib/UnAckedMessageTrackerEnabled.cc @@ -90,12 +90,13 @@ UnAckedMessageTrackerEnabled::UnAckedMessageTrackerEnabled(long timeoutMs, long timeoutHandler(); } -bool UnAckedMessageTrackerEnabled::add(const MessageId& m) { +bool UnAckedMessageTrackerEnabled::add(const MessageId& msgId) { std::lock_guard acquire(lock_); - if (messageIdPartitionMap.count(m) == 0) { + MessageId id(msgId.partition(), msgId.ledgerId(), msgId.entryId(), -1); + if (messageIdPartitionMap.count(id) == 0) { std::set& partition = timePartitions.back(); - bool emplace = messageIdPartitionMap.emplace(m, partition).second; - bool insert = partition.insert(m).second; + bool emplace = messageIdPartitionMap.emplace(id, partition).second; + bool insert = partition.insert(id).second; return emplace && insert; } return false; @@ -106,13 +107,14 @@ bool UnAckedMessageTrackerEnabled::isEmpty() { return messageIdPartitionMap.empty(); } -bool UnAckedMessageTrackerEnabled::remove(const MessageId& m) { +bool UnAckedMessageTrackerEnabled::remove(const MessageId& msgId) { std::lock_guard acquire(lock_); + MessageId id(msgId.partition(), msgId.ledgerId(), msgId.entryId(), -1); bool removed = false; - std::map&>::iterator exist = messageIdPartitionMap.find(m); + std::map&>::iterator exist = messageIdPartitionMap.find(id); if (exist != messageIdPartitionMap.end()) { - removed = exist->second.erase(m); + removed = exist->second.erase(id); messageIdPartitionMap.erase(exist); } return removed; diff --git a/pulsar-client-cpp/lib/UnAckedMessageTrackerEnabled.h b/pulsar-client-cpp/lib/UnAckedMessageTrackerEnabled.h index 36753dca51298..12daf4565e903 100644 --- a/pulsar-client-cpp/lib/UnAckedMessageTrackerEnabled.h +++ b/pulsar-client-cpp/lib/UnAckedMessageTrackerEnabled.h @@ -29,8 +29,8 @@ class UnAckedMessageTrackerEnabled : public UnAckedMessageTrackerInterface { ~UnAckedMessageTrackerEnabled(); UnAckedMessageTrackerEnabled(long timeoutMs, const ClientImplPtr, ConsumerImplBase&); UnAckedMessageTrackerEnabled(long timeoutMs, long tickDuration, const ClientImplPtr, ConsumerImplBase&); - bool add(const MessageId& m); - bool remove(const MessageId& m); + bool add(const MessageId& msgId); + bool remove(const MessageId& msgId); void removeMessagesTill(const MessageId& msgId); void removeTopicMessage(const std::string& topic); void timeoutHandler(); @@ -52,6 +52,7 @@ class UnAckedMessageTrackerEnabled : public UnAckedMessageTrackerInterface { FRIEND_TEST(ConsumerTest, testPartitionedConsumerUnAckedMessageRedelivery); FRIEND_TEST(ConsumerTest, testMultiTopicsConsumerUnAckedMessageRedelivery); + FRIEND_TEST(ConsumerTest, testBatchUnAckedMessageTracker); }; } // namespace pulsar diff --git a/pulsar-client-cpp/tests/ConsumerTest.cc b/pulsar-client-cpp/tests/ConsumerTest.cc index 2278c05cda2bf..40e34d2f419e0 100644 --- a/pulsar-client-cpp/tests/ConsumerTest.cc +++ b/pulsar-client-cpp/tests/ConsumerTest.cc @@ -20,6 +20,8 @@ #include #include #include +#include +#include #include "gtest/gtest.h" @@ -345,4 +347,71 @@ TEST(ConsumerTest, testMultiTopicsConsumerUnAckedMessageRedelivery) { client.close(); } +TEST(ConsumerTest, testBatchUnAckedMessageTracker) { + Client client(lookupUrl); + const std::string topic = + "testBatchUnAckedMessageTracker" + std::to_string(time(nullptr)); + std::string subName = "sub-batch-un-acked-msg-tracker"; + constexpr int numOfMessages = 50; + constexpr int batchSize = 5; + constexpr int batchCount = numOfMessages / batchSize; + constexpr int unAckedMessagesTimeoutMs = 10000; + constexpr int tickDurationInMs = 1000; + + Consumer consumer; + ConsumerConfiguration consumerConfig; + consumerConfig.setUnAckedMessagesTimeoutMs(unAckedMessagesTimeoutMs); + consumerConfig.setTickDurationInMs(tickDurationInMs); + ASSERT_EQ(ResultOk, client.subscribe(topic, subName, consumerConfig, consumer)); + auto consumerImplPtr = PulsarFriend::getConsumerImplPtr(consumer); + auto tracker = + static_cast(consumerImplPtr->unAckedMessageTrackerPtr_.get()); + + // send messages + ProducerConfiguration producerConfig; + producerConfig.setBatchingEnabled(true); + producerConfig.setBlockIfQueueFull(true); + producerConfig.setBatchingMaxMessages(batchSize); + Producer producer; + ASSERT_EQ(ResultOk, client.createProducer(topic, producerConfig, producer)); + std::string prefix = "message-"; + for (int i = 0; i < numOfMessages; i++) { + std::string messageContent = prefix + std::to_string(i); + Message msg = MessageBuilder().setContent(messageContent).build(); + producer.sendAsync(msg, NULL); + } + producer.close(); + + std::map> msgIdInBatchMap; + std::vector messageIds; + for (auto i = 0; i < numOfMessages; ++i) { + Message msg; + ASSERT_EQ(ResultOk, consumer.receive(msg, 1000)); + MessageId msgId = msg.getMessageId(); + MessageId id(msgId.partition(), msgId.ledgerId(), msgId.entryId(), -1); + msgIdInBatchMap[id].emplace_back(msgId); + } + + ASSERT_EQ(batchCount, msgIdInBatchMap.size()); + ASSERT_EQ(batchCount, tracker->size()); + for (const auto& iter: msgIdInBatchMap) { + ASSERT_EQ(iter.second.size(), batchSize); + } + + int ackedBatchCount = 0; + for (auto iter = msgIdInBatchMap.begin(); iter != msgIdInBatchMap.end(); ++iter) { + ASSERT_EQ(batchSize, iter->second.size()); + for (auto i = 0; i < iter->second.size(); ++i) { + ASSERT_EQ(ResultOk, consumer.acknowledge(iter->second[i])); + } + ackedBatchCount++; + ASSERT_EQ(batchCount - ackedBatchCount, tracker->size()); + } + ASSERT_EQ(0, tracker->size()); + ASSERT_TRUE(tracker->isEmpty()); + + consumer.close(); + client.close(); +} + } // namespace pulsar From 7b9fd0e4cdcc0a170b030cf846a3fef50b7ad6cf Mon Sep 17 00:00:00 2001 From: saosir Date: Mon, 11 Jan 2021 15:52:01 +0800 Subject: [PATCH 2/2] make format --- pulsar-client-cpp/tests/ConsumerTest.cc | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/pulsar-client-cpp/tests/ConsumerTest.cc b/pulsar-client-cpp/tests/ConsumerTest.cc index 40e34d2f419e0..e0d4091ee5355 100644 --- a/pulsar-client-cpp/tests/ConsumerTest.cc +++ b/pulsar-client-cpp/tests/ConsumerTest.cc @@ -349,8 +349,7 @@ TEST(ConsumerTest, testMultiTopicsConsumerUnAckedMessageRedelivery) { TEST(ConsumerTest, testBatchUnAckedMessageTracker) { Client client(lookupUrl); - const std::string topic = - "testBatchUnAckedMessageTracker" + std::to_string(time(nullptr)); + const std::string topic = "testBatchUnAckedMessageTracker" + std::to_string(time(nullptr)); std::string subName = "sub-batch-un-acked-msg-tracker"; constexpr int numOfMessages = 50; constexpr int batchSize = 5; @@ -394,7 +393,7 @@ TEST(ConsumerTest, testBatchUnAckedMessageTracker) { ASSERT_EQ(batchCount, msgIdInBatchMap.size()); ASSERT_EQ(batchCount, tracker->size()); - for (const auto& iter: msgIdInBatchMap) { + for (const auto& iter : msgIdInBatchMap) { ASSERT_EQ(iter.second.size(), batchSize); }