(pulsar) branch branch-2.10 updated: [fix][broker] Fix unack count when mixing non batch index and batch index acks (#21126)

[zixuan]

This is an automated email from the ASF dual-hosted git repository.

zixuan pushed a commit to branch branch-2.10
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/branch-2.10 by this push:
     new 46d2ecde112 [fix][broker] Fix unack count when mixing non batch index 
and batch index acks (#21126)
46d2ecde112 is described below

commit 46d2ecde1127eb6a6c5cd2390d2061dbfa78c0b9
Author: erobot <ero...@users.noreply.github.com>
AuthorDate: Mon Sep 11 09:56:34 2023 +0800

    [fix][broker] Fix unack count when mixing non batch index and batch index 
acks (#21126)
    
    (cherry picked from commit 6e8bf93135b2e80c657e7d80f94a57f40e634f02)
---
 .../org/apache/pulsar/broker/service/Consumer.java |  9 ++----
 .../BatchMessageWithBatchIndexLevelTest.java       | 36 ++++++++++++++++++++++
 2 files changed, 38 insertions(+), 7 deletions(-)

diff --git 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
index a7c06d0c85d..1e6d9690c26 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
@@ -444,7 +444,6 @@ public class Consumer {
     private CompletableFuture<Long> individualAckNormal(CommandAck ack, 
Map<String, Long> properties) {
         List<Position> positionsAcked = new ArrayList<>();
         long totalAckCount = 0;
-        boolean individualAck = false;
         for (int i = 0; i < ack.getMessageIdsCount(); i++) {
             MessageIdData msgId = ack.getMessageIdAt(i);
             PositionImpl position;
@@ -465,19 +464,15 @@ public class Consumer {
                                 
.syncBatchPositionBitSetForPendingAck(position);
                     }
                 }
+                addAndGetUnAckedMsgs(ackOwnerConsumer, -(int) ackedCount);
             } else {
                 position = PositionImpl.get(msgId.getLedgerId(), 
msgId.getEntryId());
                 ackedCount = getAckedCountForMsgIdNoAckSets(batchSize, 
position, ackOwnerConsumer);
-                individualAck = true;
-            }
-
-            if (individualAck) {
                 if (checkCanRemovePendingAcksAndHandle(position, msgId)) {
                     addAndGetUnAckedMsgs(ackOwnerConsumer, -(int) ackedCount);
                 }
-            } else {
-                addAndGetUnAckedMsgs(ackOwnerConsumer, -(int) ackedCount);
             }
+
             positionsAcked.add(position);
 
             checkAckValidationError(ack, position);
diff --git 
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageWithBatchIndexLevelTest.java
 
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageWithBatchIndexLevelTest.java
index 769b611b0bd..344f8c34514 100644
--- 
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageWithBatchIndexLevelTest.java
+++ 
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageWithBatchIndexLevelTest.java
@@ -369,4 +369,40 @@ public class BatchMessageWithBatchIndexLevelTest extends 
BatchMessageTest {
         consumer.close();
         admin.topics().delete(topicName);
     }
+
+    @Test
+    public void testMixIndexAndNonIndexUnAckMessageCount() throws Exception {
+        final String topicName = 
"persistent://prop/ns-abc/testMixIndexAndNonIndexUnAckMessageCount-";
+
+        @Cleanup
+        Producer<byte[]> producer = pulsarClient.newProducer()
+                .topic(topicName)
+                .enableBatching(true)
+                .batchingMaxPublishDelay(100, TimeUnit.MILLISECONDS)
+                .create();
+        @Cleanup
+        Consumer<byte[]> consumer = pulsarClient.newConsumer()
+                .topic(topicName)
+                .subscriptionName("sub")
+                .subscriptionType(SubscriptionType.Shared)
+                .acknowledgmentGroupTime(100, TimeUnit.MILLISECONDS)
+                .enableBatchIndexAcknowledgment(true)
+                .isAckReceiptEnabled(true)
+                .subscribe();
+
+        // send two batch messages: [(1), (2,3)]
+        producer.send("1".getBytes());
+        producer.sendAsync("2".getBytes());
+        producer.send("3".getBytes());
+
+        Message<byte[]> message1 = consumer.receive();
+        Message<byte[]> message2 = consumer.receive();
+        Message<byte[]> message3 = consumer.receive();
+        consumer.acknowledgeAsync(message1);
+        consumer.acknowledge(message2);  // send group ack: non-index ack for 
1, index ack for 2
+        consumer.acknowledge(message3);  // index ack for 3
+
+        assertEquals(admin.topics().getStats(topicName).getSubscriptions()
+                .get("sub").getUnackedMessages(), 0);
+    }
 }