[fix][broker] Fix unack count when mixing non batch index and batch i…

…ndex acks (#21126)
apache · Sep 11, 2023 · 6e8bf93 · 6e8bf93
1 parent 9f12ace
commit 6e8bf93
Show file tree

Hide file tree

Showing 2 changed files with 38 additions and 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
@@ -487,7 +487,6 @@ public CompletableFuture<Void> messageAcked(CommandAck ack) {
     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;
@@ -508,19 +507,15 @@ private CompletableFuture<Long> individualAckNormal(CommandAck ack, Map<String,
                                 .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/...r/src/test/java/org/apache/pulsar/broker/service/BatchMessageWithBatchIndexLevelTest.java b/...r/src/test/java/org/apache/pulsar/broker/service/BatchMessageWithBatchIndexLevelTest.java
@@ -365,4 +365,40 @@ public void testNegativeAckAndLongAckDelayWillNotLeadRepeatConsume() throws Exce
         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);
+    }
 }