Update Consumer.java

180254 · May 3, 2024 · 6dac4bf · 6dac4bf
1 parent 2d378d3
commit 6dac4bf
Showing 1 changed file with 4 additions and 4 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
@@ -516,17 +516,17 @@ 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);
-                if (checkCanRemovePendingAcksAndHandle(position, msgId)) {
-                    addAndGetUnAckedMsgs(ackOwnerConsumer, -(int) ackedCount);
-                }
             }
 
+            addAndGetUnAckedMsgs(ackOwnerConsumer, -(int) ackedCount);
+
             positionsAcked.add(position);
 
+            checkCanRemovePendingAcksAndHandle(position, msgId);
+
             checkAckValidationError(ack, position);
 
             totalAckCount += ackedCount;