[fix][clients]Check pendingIndividualBatchIndexAcks size in doIndivid…

…ualBatchAckAsync (apache#15877) * Check pendingIndividualBatchIndexAcks size and flush in doIndividualBatchAckAsync * put logic into method doIndividualBatchAck
felixYyu · Jun 10, 2022 · de9928a · de9928a
1 parent 6722417
commit de9928a
Showing 1 changed file with 6 additions and 6 deletions.
diff --git a/...src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java b/...src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java
@@ -312,9 +312,15 @@ private CompletableFuture<Void> doIndividualBatchAck(BatchMessageIdImpl batchMes
                 return this.currentIndividualAckFuture;
             } finally {
                 this.lock.readLock().unlock();
+                if (pendingIndividualBatchIndexAcks.size() >= MAX_ACK_GROUP_SIZE) {
+                    flush();
+                }
             }
         } else {
             doIndividualBatchAckAsync(batchMessageId);
+            if (pendingIndividualBatchIndexAcks.size() >= MAX_ACK_GROUP_SIZE) {
+                flush();
+            }
             return CompletableFuture.completedFuture(null);
         }
     }
@@ -337,15 +343,9 @@ private CompletableFuture<Void> doCumulativeAck(MessageIdImpl messageId, Map<Str
                     return this.currentCumulativeAckFuture;
                 } finally {
                     this.lock.readLock().unlock();
-                    if (pendingIndividualBatchIndexAcks.size() >= MAX_ACK_GROUP_SIZE) {
-                        flush();
-                    }
                 }
             } else {
                 doCumulativeAckAsync(messageId, bitSet);
-                if (pendingIndividualBatchIndexAcks.size() >= MAX_ACK_GROUP_SIZE) {
-                    flush();
-                }
                 return CompletableFuture.completedFuture(null);
             }
         }