Fix pending batchIndexAcks bitSet batchSize in PersistentAcknowledgme…

…ntsGroupingTracker (#7828) ### Motivation The pending batchIndexAcks bitSet batchSize is not correct. ### Modifications Fix the bitSet batchSize.
apache · Aug 18, 2020 · 23d795c · 23d795c
1 parent 8de22bd
commit 23d795c
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 4 deletions.
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BatchMessageIndexAckTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BatchMessageIndexAckTest.java
@@ -28,6 +28,7 @@
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
 import org.apache.pulsar.common.util.FutureUtil;
 import org.testng.Assert;
 import org.testng.annotations.AfterMethod;
@@ -58,13 +59,14 @@ protected void cleanup() throws Exception {
     }
 
     @Test
-    public void testBatchMessageIndexAckForSharedSubscription() throws PulsarClientException, ExecutionException, InterruptedException {
+    public void testBatchMessageIndexAckForSharedSubscription() throws Exception {
         final String topic = "testBatchMessageIndexAckForSharedSubscription";
+        final String subscriptionName = "sub";
 
         @Cleanup
         Consumer<Integer> consumer = pulsarClient.newConsumer(Schema.INT32)
             .topic(topic)
-            .subscriptionName("sub")
+            .subscriptionName(subscriptionName)
             .receiverQueueSize(100)
             .subscriptionType(SubscriptionType.Shared)
             .enableBatchIndexAcknowledgment(true)
@@ -115,6 +117,12 @@ public void testBatchMessageIndexAckForSharedSubscription() throws PulsarClientE
         Message<Integer> moreMessage = consumer.receive(2, TimeUnit.SECONDS);
         Assert.assertNull(moreMessage);
 
+        // check the mark delete position was changed
+        BatchMessageIdImpl ackedMessageId = (BatchMessageIdImpl) received.get(0);
+        PersistentTopicInternalStats stats = admin.topics().getInternalStats(topic);
+        String markDeletePosition = stats.cursors.get(subscriptionName).markDeletePosition;
+        Assert.assertEquals(ackedMessageId.ledgerId + ":" + ackedMessageId.entryId, markDeletePosition);
+
         futures.clear();
         for (int i = 0; i < 50; i++) {
             futures.add(producer.sendAsync(i));

diff --git a/...src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java b/...src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java
@@ -169,8 +169,7 @@ public void addBatchIndexAcknowledgment(BatchMessageIdImpl msgId, int batchIndex
             ConcurrentBitSetRecyclable bitSet = pendingIndividualBatchIndexAcks.computeIfAbsent(
                 new MessageIdImpl(msgId.getLedgerId(), msgId.getEntryId(), msgId.getPartitionIndex()), (v) -> {
                     ConcurrentBitSetRecyclable value = ConcurrentBitSetRecyclable.create();
-                    value.set(0, batchSize + 1);
-                    value.clear(batchIndex);
+                    value.set(0, batchSize);
                     return value;
                 });
             bitSet.clear(batchIndex);