From ae415e4cab69c2648b2c4382feb7e286f4c149cf Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Sun, 30 Jan 2022 18:40:33 +0800 Subject: [PATCH 1/2] [C++] Change AutoOldest to AutoAckOldest in consumer configuration --- pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h | 8 ++++---- pulsar-client-cpp/lib/ConsumerConfiguration.cc | 4 ++-- pulsar-client-cpp/lib/ConsumerImpl.cc | 2 +- pulsar-client-cpp/tests/ConsumerConfigurationTest.cc | 6 +++--- 4 files changed, 10 insertions(+), 10 deletions(-) diff --git a/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h b/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h index 2cdbf4707061e..812bea799c2bb 100644 --- a/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h +++ b/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h @@ -450,7 +450,7 @@ class PULSAR_PUBLIC ConsumerConfiguration { * Buffering large number of outstanding uncompleted chunked messages can create memory pressure and it * can be guarded by providing this maxPendingChunkedMessage threshold. Once, consumer reaches this * threshold, it drops the outstanding unchunked-messages by silently acking or asking broker to redeliver - * later by marking it unacked. See setAutoOldestChunkedMessageOnQueueFull. + * later by marking it unacked. See setAutoAckOldestChunkedMessageOnQueueFull. * * If it's zero, the pending chunked messages will not be limited. * @@ -475,13 +475,13 @@ class PULSAR_PUBLIC ConsumerConfiguration { * * @param autoAckOldestChunkedMessageOnQueueFull whether to ack the discarded chunked message */ - ConsumerConfiguration& setAutoOldestChunkedMessageOnQueueFull( + ConsumerConfiguration& setAutoAckOldestChunkedMessageOnQueueFull( bool autoAckOldestChunkedMessageOnQueueFull); /** - * The associated getter of setAutoOldestChunkedMessageOnQueueFull + * The associated getter of setAutoAckOldestChunkedMessageOnQueueFull */ - bool isAutoOldestChunkedMessageOnQueueFull() const; + bool isAutoAckOldestChunkedMessageOnQueueFull() const; friend class PulsarWrapper; diff --git a/pulsar-client-cpp/lib/ConsumerConfiguration.cc b/pulsar-client-cpp/lib/ConsumerConfiguration.cc index d13cb0e0c86b2..b755063d419b1 100644 --- a/pulsar-client-cpp/lib/ConsumerConfiguration.cc +++ b/pulsar-client-cpp/lib/ConsumerConfiguration.cc @@ -238,13 +238,13 @@ ConsumerConfiguration& ConsumerConfiguration::setMaxPendingChunkedMessage(size_t size_t ConsumerConfiguration::getMaxPendingChunkedMessage() const { return impl_->maxPendingChunkedMessage; } -ConsumerConfiguration& ConsumerConfiguration::setAutoOldestChunkedMessageOnQueueFull( +ConsumerConfiguration& ConsumerConfiguration::setAutoAckOldestChunkedMessageOnQueueFull( bool autoAckOldestChunkedMessageOnQueueFull) { impl_->autoAckOldestChunkedMessageOnQueueFull = autoAckOldestChunkedMessageOnQueueFull; return *this; } -bool ConsumerConfiguration::isAutoOldestChunkedMessageOnQueueFull() const { +bool ConsumerConfiguration::isAutoAckOldestChunkedMessageOnQueueFull() const { return impl_->autoAckOldestChunkedMessageOnQueueFull; } diff --git a/pulsar-client-cpp/lib/ConsumerImpl.cc b/pulsar-client-cpp/lib/ConsumerImpl.cc index fa817a094ce24..b5b5ceb046f99 100644 --- a/pulsar-client-cpp/lib/ConsumerImpl.cc +++ b/pulsar-client-cpp/lib/ConsumerImpl.cc @@ -65,7 +65,7 @@ ConsumerImpl::ConsumerImpl(const ClientImplPtr client, const std::string& topic, readCompacted_(conf.isReadCompacted()), startMessageId_(startMessageId), maxPendingChunkedMessage_(conf.getMaxPendingChunkedMessage()), - autoAckOldestChunkedMessageOnQueueFull_(conf.isAutoOldestChunkedMessageOnQueueFull()) { + autoAckOldestChunkedMessageOnQueueFull_(conf.isAutoAckOldestChunkedMessageOnQueueFull()) { std::stringstream consumerStrStream; consumerStrStream << "[" << topic_ << ", " << subscription_ << ", " << consumerId_ << "] "; consumerStr_ = consumerStrStream.str(); diff --git a/pulsar-client-cpp/tests/ConsumerConfigurationTest.cc b/pulsar-client-cpp/tests/ConsumerConfigurationTest.cc index 57ed0ec7c79c6..941924c0699ef 100644 --- a/pulsar-client-cpp/tests/ConsumerConfigurationTest.cc +++ b/pulsar-client-cpp/tests/ConsumerConfigurationTest.cc @@ -60,7 +60,7 @@ TEST(ConsumerConfigurationTest, testDefaultConfig) { ASSERT_EQ(conf.getProperties().empty(), true); ASSERT_EQ(conf.getPriorityLevel(), 0); ASSERT_EQ(conf.getMaxPendingChunkedMessage(), 100); - ASSERT_EQ(conf.isAutoOldestChunkedMessageOnQueueFull(), false); + ASSERT_EQ(conf.isAutoAckOldestChunkedMessageOnQueueFull(), false); } TEST(ConsumerConfigurationTest, testCustomConfig) { @@ -145,8 +145,8 @@ TEST(ConsumerConfigurationTest, testCustomConfig) { conf.setMaxPendingChunkedMessage(500); ASSERT_EQ(conf.getMaxPendingChunkedMessage(), 500); - conf.setAutoOldestChunkedMessageOnQueueFull(true); - ASSERT_TRUE(conf.isAutoOldestChunkedMessageOnQueueFull()); + conf.setAutoAckOldestChunkedMessageOnQueueFull(true); + ASSERT_TRUE(conf.isAutoAckOldestChunkedMessageOnQueueFull()); } TEST(ConsumerConfigurationTest, testReadCompactPersistentExclusive) { From b9d540f3cb682c91d7450867b1fafcf831965370 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Sun, 30 Jan 2022 18:59:49 +0800 Subject: [PATCH 2/2] Change the default value of MaxPendingChunkedMessages to 10 --- pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h | 2 +- pulsar-client-cpp/lib/ConsumerConfigurationImpl.h | 2 +- pulsar-client-cpp/tests/ConsumerConfigurationTest.cc | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h b/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h index 812bea799c2bb..70c90cff3d73d 100644 --- a/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h +++ b/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h @@ -454,7 +454,7 @@ class PULSAR_PUBLIC ConsumerConfiguration { * * If it's zero, the pending chunked messages will not be limited. * - * Default: 100 + * Default: 10 * * @param maxPendingChunkedMessage the number of max pending chunked messages */ diff --git a/pulsar-client-cpp/lib/ConsumerConfigurationImpl.h b/pulsar-client-cpp/lib/ConsumerConfigurationImpl.h index 9c2a4615fef43..1848f2dafad47 100644 --- a/pulsar-client-cpp/lib/ConsumerConfigurationImpl.h +++ b/pulsar-client-cpp/lib/ConsumerConfigurationImpl.h @@ -50,7 +50,7 @@ struct ConsumerConfigurationImpl { std::map properties; int priorityLevel{0}; KeySharedPolicy keySharedPolicy; - size_t maxPendingChunkedMessage{100}; + size_t maxPendingChunkedMessage{10}; bool autoAckOldestChunkedMessageOnQueueFull{false}; }; } // namespace pulsar diff --git a/pulsar-client-cpp/tests/ConsumerConfigurationTest.cc b/pulsar-client-cpp/tests/ConsumerConfigurationTest.cc index 941924c0699ef..fc67e867e3ddb 100644 --- a/pulsar-client-cpp/tests/ConsumerConfigurationTest.cc +++ b/pulsar-client-cpp/tests/ConsumerConfigurationTest.cc @@ -59,7 +59,7 @@ TEST(ConsumerConfigurationTest, testDefaultConfig) { ASSERT_EQ(conf.isReplicateSubscriptionStateEnabled(), false); ASSERT_EQ(conf.getProperties().empty(), true); ASSERT_EQ(conf.getPriorityLevel(), 0); - ASSERT_EQ(conf.getMaxPendingChunkedMessage(), 100); + ASSERT_EQ(conf.getMaxPendingChunkedMessage(), 10); ASSERT_EQ(conf.isAutoAckOldestChunkedMessageOnQueueFull(), false); }