Fixed accessing MessageImpl after it was enqueued on user queue (#11824)

apache · Aug 28, 2021 · 666ad3b · 666ad3b
1 parent c178564
commit 666ad3b
Showing 1 changed file with 4 additions and 5 deletions.
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
@@ -727,8 +727,11 @@ protected boolean canEnqueueMessage(Message<T> message) {
     }
 
     protected boolean enqueueMessageAndCheckBatchReceive(Message<T> message) {
+        int messageSize = message.size();
         if (canEnqueueMessage(message) && incomingMessages.offer(message)) {
-            increaseIncomingMessageSize(message);
+            // After we have enqueued the messages on `incomingMessages` queue, we cannot touch the message instance
+            // anymore, since for pooled messages, this instance was possibly already been released and recycled.
+            INCOMING_MESSAGES_SIZE_UPDATER.addAndGet(this, messageSize);
         }
         return hasEnoughMessagesForBatchReceive();
     }
@@ -970,10 +973,6 @@ protected boolean hasPendingBatchReceive() {
         return pendingBatchReceives != null && hasNextBatchReceive();
     }
 
-    protected void increaseIncomingMessageSize(final Message<?> message) {
-        INCOMING_MESSAGES_SIZE_UPDATER.addAndGet(this, message.size());
-    }
-
     protected void resetIncomingMessageSize() {
         INCOMING_MESSAGES_SIZE_UPDATER.set(this, 0);
     }