Fix pending batchIndexAcks bitSet batchSize in PersistentAcknowledgmentsGroupingTracker (#7828)

### Motivation

The pending batchIndexAcks bitSet batchSize is not correct.

### Modifications

Fix the bitSet batchSize.

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

@@ -28,6 +28,7 @@ import org.apache.pulsar.client.api.ProducerConsumerBase;
 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 @@ public class BatchMessageIndexAckTest extends ProducerConsumerBase {
     }
 
     @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 class BatchMessageIndexAckTest extends ProducerConsumerBase {
         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));

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

@@ -169,8 +169,7 @@ public class PersistentAcknowledgmentsGroupingTracker implements Acknowledgments
             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);