fix lock issue.

apache · Nov 19, 2022 · 6228045 · 6228045
1 parent f37ad7f
commit 6228045
Showing 1 changed file with 5 additions and 1 deletion.
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
@@ -847,7 +847,9 @@ protected boolean canEnqueueMessage(Message<T> message) {
 
     protected boolean enqueueMessageAndCheckBatchReceive(Message<T> message) {
         int messageSize = message.size();
-        synchronized (incomingQueueLock) {
+        // synchronize redeliverUnacknowledgedMessages().
+        incomingQueueLock.lock();
+        try {
             if (isValidConsumerEpoch((MessageImpl<T>) message) && canEnqueueMessage(message)
                     && incomingMessages.offer(message)) {
                 // After we have enqueued the messages on `incomingMessages` queue, we cannot touch the message
@@ -857,6 +859,8 @@ protected boolean enqueueMessageAndCheckBatchReceive(Message<T> message) {
                 getMemoryLimitController().ifPresent(limiter -> limiter.forceReserveMemory(messageSize));
                 updateAutoScaleReceiverQueueHint();
             }
+        } finally {
+            incomingQueueLock.unlock();
         }
         return hasEnoughMessagesForBatchReceive();
     }