feat: add negativeack precision bit configuration and related tests

Author: syslucas

include/pulsar/ConsumerConfiguration.h

@@ -277,6 +277,21 @@ class PULSAR_PUBLIC ConsumerConfiguration {
      */
     long getNegativeAckRedeliveryDelayMs() const;
 
+    /**
+     * Set the precision bit count for negative ack redelivery delay.
+     * The lower bits of the redelivery time will be trimmed to reduce the memory occupation.
+     * @param negativeAckPrecisionBitCnt
+     *            negative ack precision bit count
+     */
+    void setNegativeAckPrecisionBitCnt(int negativeAckPrecisionBitCnt);
+
+    /**
+     * Get the configured precision bit count for negative ack redelivery delay.
+     *
+     * @return redelivery time precision bit count
+     */
+    int getNegativeAckPrecisionBitCnt() const;
+
     /**
      * Set time window in milliseconds for grouping message ACK requests. An ACK request is not sent
      * to broker until the time window reaches its end, or the number of grouped messages reaches

lib/ConsumerConfiguration.cc

@@ -134,6 +134,16 @@ void ConsumerConfiguration::setAckGroupingTimeMs(long ackGroupingMillis) {
     impl_->ackGroupingTimeMs = ackGroupingMillis;
 }
 
+int ConsumerConfiguration::getNegativeAckPrecisionBitCnt() const { return impl_->negativeAckPrecisionBitCnt; }
+
+void ConsumerConfiguration::setNegativeAckPrecisionBitCnt(int negativeAckPrecisionBitCnt) {
+    if (negativeAckPrecisionBitCnt < 0) {
+        throw std::invalid_argument(
+            "Consumer Config Exception: NegativeAckPrecisionBitCnt should be nonnegative number.");
+    }
+    impl_->negativeAckPrecisionBitCnt = negativeAckPrecisionBitCnt;
+}
+
 long ConsumerConfiguration::getAckGroupingTimeMs() const { return impl_->ackGroupingTimeMs; }
 
 void ConsumerConfiguration::setAckGroupingMaxSize(long maxGroupingSize) {

lib/ConsumerConfigurationImpl.h

@@ -26,6 +26,7 @@ struct ConsumerConfigurationImpl {
     long unAckedMessagesTimeoutMs{0};
     long tickDurationInMs{1000};
     long negativeAckRedeliveryDelayMs{60000};
+    int negativeAckPrecisionBitCnt{8};
     long ackGroupingTimeMs{100};
     long ackGroupingMaxSize{1000};
     long brokerConsumerStatsCacheTimeInMs{30 * 1000L};  // 30 seconds

lib/NegativeAcksTracker.cc

@@ -46,6 +46,7 @@ NegativeAcksTracker::NegativeAcksTracker(const ClientImplPtr &client, ConsumerIm
     nackDelay_ =
         std::chrono::milliseconds(std::max(conf.getNegativeAckRedeliveryDelayMs(), MIN_NACK_DELAY_MILLIS));
     timerInterval_ = std::chrono::milliseconds((long)(nackDelay_.count() / 3));
+    nackPrecisionBit_ = conf.getNegativeAckPrecisionBitCnt();
     LOG_DEBUG("Created negative ack tracker with delay: " << nackDelay_.count() << " ms - Timer interval: "
                                                           << timerInterval_.count());
 }
@@ -106,14 +107,27 @@ void NegativeAcksTracker::handleTimer(const ASIO_ERROR &ec) {
     scheduleTimer();
 }
 
+std::chrono::steady_clock::time_point trimLowerBit(const std::chrono::steady_clock::time_point &tp,
+                                                   int bits) {
+    // get origin timestamp in nanoseconds
+    auto timestamp = std::chrono::duration_cast<std::chrono::nanoseconds>(tp.time_since_epoch()).count();
+
+    // trim lower bits
+    auto trimmedTimestamp = timestamp & (~((1LL << bits) - 1));
+
+    return std::chrono::steady_clock::time_point(std::chrono::nanoseconds(trimmedTimestamp));
+}
+
 void NegativeAcksTracker::add(const MessageId &m) {
     auto msgId = discardBatch(m);
     auto now = Clock::now();
 
     {
         std::lock_guard<std::mutex> lock{mutex_};
+        auto trimmedTimestamp = trimLowerBit(now + nackDelay_, nackPrecisionBit_);
+        // If the timestamp is already in the map, we can just add the message to the existing entry
         // Erase batch id to group all nacks from same batch
-        nackedMessages_[now][msgId.ledgerId()].add((uint64_t)msgId.entryId());
+        nackedMessages_[trimmedTimestamp][msgId.ledgerId()].add((uint64_t)msgId.entryId());
     }
 
     scheduleTimer();

lib/NegativeAcksTracker.h

@@ -72,6 +72,7 @@ class NegativeAcksTracker : public std::enable_shared_from_this<NegativeAcksTrac
 
     std::chrono::milliseconds nackDelay_;
     std::chrono::milliseconds timerInterval_;
+    int nackPrecisionBit_;
     typedef typename std::chrono::steady_clock Clock;
     std::map<Clock::time_point, std::unordered_map<LedgerId, ConditionalRoaringMap>> nackedMessages_;
 

lib/c/c_ConsumerConfiguration.cc

@@ -121,6 +121,16 @@ long pulsar_configure_get_negative_ack_redelivery_delay_ms(
     return consumer_configuration->consumerConfiguration.getNegativeAckRedeliveryDelayMs();
 }
 
+void pulsar_configure_set_negative_ack_precision_bit_cnt(
+    pulsar_consumer_configuration_t *consumer_configuration, int negativeAckPrecisionBitCnt) {
+    consumer_configuration->consumerConfiguration.setNegativeAckPrecisionBitCnt(negativeAckPrecisionBitCnt);
+}
+
+int pulsar_configure_get_negative_ack_precision_bit_cnt(
+    pulsar_consumer_configuration_t *consumer_configuration) {
+    return consumer_configuration->consumerConfiguration.getNegativeAckPrecisionBitCnt();
+}
+
 void pulsar_configure_set_ack_grouping_time_ms(pulsar_consumer_configuration_t *consumer_configuration,
                                                long ackGroupingMillis) {
     consumer_configuration->consumerConfiguration.setAckGroupingTimeMs(ackGroupingMillis);

tests/BasicEndToEndTest.cc

@@ -3249,6 +3249,60 @@ TEST(BasicEndToEndTest, testNegativeAcksWithPartitions) {
     testNegativeAcks(topicName, true);
 }
 
+int64_t getCurrentTimeMs() {
+    return std::chrono::duration_cast<std::chrono::milliseconds>(
+               std::chrono::system_clock::now().time_since_epoch())
+        .count();
+}
+
+void testNegativeAckPrecisionBitCnt(const std::string &topic, int precisionBitCnt) {
+    constexpr int delayMs = 2000;
+    const int64_t timeDeviation = 1L << precisionBitCnt;
+
+    Client client(lookupUrl);
+
+    Consumer consumer;
+    ConsumerConfiguration conf;
+    conf.setNegativeAckRedeliveryDelayMs(delayMs);
+    conf.setNegativeAckPrecisionBitCnt(precisionBitCnt);
+
+    Result result = client.subscribe(topic, "sub1", conf, consumer);
+    ASSERT_EQ(ResultOk, result);
+
+    Producer producer;
+    ProducerConfiguration producerConf;
+    result = client.createProducer(topic, producerConf, producer);
+    ASSERT_EQ(ResultOk, result);
+
+    Message msg = MessageBuilder().setContent("test-0").build();
+    producer.sendAsync(msg, nullptr);
+    producer.flush();
+
+    // receive and trigger negative ack
+    Message received;
+    consumer.receive(received);
+    consumer.negativeAcknowledge(received);
+
+    int64_t expectedRedeliveryTime = getCurrentTimeMs() + delayMs;
+
+    Message redelivered;
+    consumer.receive(redelivered);
+    int64_t now = getCurrentTimeMs();
+    ASSERT_GE(now, expectedRedeliveryTime - timeDeviation);
+    ASSERT_EQ(redelivered.getDataAsString(), "test-0");
+
+    consumer.acknowledge(redelivered);
+    client.shutdown();
+}
+
+TEST(BasicEndToEndTest, testNegativeAckPrecisionBitCnt) {
+    for (int precisionBitCnt = 1; precisionBitCnt <= 12; precisionBitCnt++) {
+        std::string topic = "testNegativeAckPrecisionBitCnt-" + std::to_string(precisionBitCnt) + "-" +
+                            std::to_string(time(nullptr));
+        testNegativeAckPrecisionBitCnt(topic, precisionBitCnt);
+    }
+}
+
 static long regexTestMessagesReceived = 0;
 
 static void regexMessageListenerFunction(const Consumer &consumer, const Message &msg) {

tests/ConsumerConfigurationTest.cc

@@ -52,6 +52,7 @@ TEST(ConsumerConfigurationTest, testDefaultConfig) {
     ASSERT_EQ(conf.getUnAckedMessagesTimeoutMs(), 0);
     ASSERT_EQ(conf.getTickDurationInMs(), 1000);
     ASSERT_EQ(conf.getNegativeAckRedeliveryDelayMs(), 60000);
+    ASSERT_EQ(conf.getNegativeAckPrecisionBitCnt(), 8);
     ASSERT_EQ(conf.getAckGroupingTimeMs(), 100);
     ASSERT_EQ(conf.getAckGroupingMaxSize(), 1000);
     ASSERT_EQ(conf.getBrokerConsumerStatsCacheTimeInMs(), 30000);
@@ -114,6 +115,9 @@ TEST(ConsumerConfigurationTest, testCustomConfig) {
     conf.setNegativeAckRedeliveryDelayMs(10000);
     ASSERT_EQ(conf.getNegativeAckRedeliveryDelayMs(), 10000);
 
+    conf.setNegativeAckPrecisionBitCnt(4);
+    ASSERT_EQ(conf.getNegativeAckPrecisionBitCnt(), 4);
+
     conf.setAckGroupingTimeMs(200);
     ASSERT_EQ(conf.getAckGroupingTimeMs(), 200);