Consumer lost message ack due to race condition in acknowledge with b…

…atch message
Shawyeok · Mar 26, 2024 · d5de766 · d5de766
1 parent a52945b
commit d5de766
Showing 1 changed file with 7 additions and 5 deletions.
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageIdAdvUtils.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageIdAdvUtils.java
@@ -51,12 +51,14 @@ static boolean acknowledge(MessageIdAdv msgId, boolean individual) {
             return false;
         }
         int batchIndex = msgId.getBatchIndex();
-        if (individual) {
-            ackSet.clear(batchIndex);
-        } else {
-            ackSet.clear(0, batchIndex + 1);
+        synchronized (ackSet) {
+            if (individual) {
+                ackSet.clear(batchIndex);
+            } else {
+                ackSet.clear(0, batchIndex + 1);
+            }
+            return ackSet.isEmpty();
         }
-        return ackSet.isEmpty();
     }
 
     static boolean isBatch(MessageIdAdv msgId) {