[improve][client] PIP-393: Support configuring NegativeAckPrecisionBitCnt while building consumer. (#23804)

(cherry picked from commit f199e88)

Author: thetumbled

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/NegativeAcksTest.java

@@ -545,4 +545,51 @@ public void testMultiTopicConsumerConcurrentRedeliverAndReceive() throws Excepti
         consumer.close();
         admin.topics().deletePartitionedTopic("persistent://public/default/" + topic);
     }
+
+    @DataProvider(name = "negativeAckPrecisionBitCnt")
+    public Object[][] negativeAckPrecisionBitCnt() {
+        return new Object[][]{
+                {1}, {2}, {3}, {4}, {5}, {6}, {7}, {8}, {9}, {10}, {11}, {12}
+        };
+    }
+
+    /**
+     * When negativeAckPrecisionBitCnt is greater than 0, the lower bits of the redelivery time will be truncated
+     * to reduce the memory occupation. If set to k, the redelivery time will be bucketed by 2^k ms, resulting in
+     * the redelivery time could be earlier(no later) than the expected time no more than 2^k ms.
+     * @throws Exception if an error occurs
+     */
+    @Test(dataProvider = "negativeAckPrecisionBitCnt")
+    public void testConfigureNegativeAckPrecisionBitCnt(int negativeAckPrecisionBitCnt) throws Exception {
+        String topic = BrokerTestUtil.newUniqueName("testConfigureNegativeAckPrecisionBitCnt");
+        long timeDeviation = 1L << negativeAckPrecisionBitCnt;
+        long delayInMs = 2000;
+
+        @Cleanup
+        Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
+                .topic(topic)
+                .subscriptionName("sub1")
+                .acknowledgmentGroupTime(0, TimeUnit.SECONDS)
+                .subscriptionType(SubscriptionType.Shared)
+                .negativeAckRedeliveryDelay(delayInMs, TimeUnit.MILLISECONDS)
+                .negativeAckRedeliveryDelayPrecision(negativeAckPrecisionBitCnt)
+                .subscribe();
+
+        @Cleanup
+        Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+                .topic(topic)
+                .create();
+        producer.sendAsync("test-0");
+        producer.flush();
+
+        // receive the message and negative ack
+        consumer.negativeAcknowledge(consumer.receive());
+        long expectedTime = System.currentTimeMillis() + delayInMs;
+
+        // receive the redelivered message and calculate the time deviation
+        // assert that the redelivery time is no earlier than the `expected time - timeDeviation`
+        Message<String> msg1 = consumer.receive();
+        assertTrue(System.currentTimeMillis() >= expectedTime - timeDeviation);
+        assertNotNull(msg1);
+    }
 }

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java

@@ -243,6 +243,19 @@ public interface ConsumerBuilder<T> extends Cloneable {
      */
     ConsumerBuilder<T> negativeAckRedeliveryDelay(long redeliveryDelay, TimeUnit timeUnit);
 
+    /**
+     * Sets the redelivery time precision bit count. The lower bits of the redelivery time will be
+     * trimmed to reduce the memory occupation. The default value is 8, which means the redelivery time
+     * will be bucketed by 256ms, the redelivery time could be earlier(no later) than the expected time,
+     * but no more than 256ms. If set to k, the redelivery time will be bucketed by 2^k ms.
+     * If the value is 0, the redelivery time will be accurate to ms.
+     *
+     * @param negativeAckPrecisionBitCnt
+     *            The redelivery time precision bit count.
+     * @return the consumer builder instance
+     */
+    ConsumerBuilder<T> negativeAckRedeliveryDelayPrecision(int negativeAckPrecisionBitCount);
+
     /**
      * Select the subscription type to be used when subscribing to a topic.
      *

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java

@@ -281,6 +281,13 @@ public ConsumerBuilder<T> negativeAckRedeliveryDelay(long redeliveryDelay, TimeU
         return this;
     }
 
+    @Override
+    public ConsumerBuilder<T> negativeAckRedeliveryDelayPrecision(int negativeAckPrecisionBitCount) {
+        checkArgument(negativeAckPrecisionBitCount >= 0, "negativeAckPrecisionBitCount needs to be >= 0");
+        conf.setNegativeAckPrecisionBitCnt(negativeAckPrecisionBitCount);
+        return this;
+    }
+
     @Override
     public ConsumerBuilder<T> subscriptionType(@NonNull SubscriptionType subscriptionType) {
         conf.setSubscriptionType(subscriptionType);