[fix] [client] fix memory leak if enabled pooled messages (apache#19585)

Author: poorbarcode

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java

@@ -947,9 +947,14 @@ public void testPooledMessageWithAckTimeout(boolean isBatchingEnabled) throws Ex
         ByteBuf payload = ((MessageImpl) msg).getPayload();
         assertNotEquals(payload.refCnt(), 0);
         consumer.redeliverUnacknowledgedMessages();
-        assertEquals(payload.refCnt(), 0);
+        Awaitility.await().untilAsserted(() -> {
+            assertTrue(consumer.incomingMessages.size() >= 100);
+        });
         consumer.close();
         producer.close();
+        admin.topics().delete(topic, false);
+        assertEquals(consumer.incomingMessages.size(), 0);
+        assertEquals(payload.refCnt(), 0);
     }
 
     /**

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java

@@ -29,7 +29,6 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
-import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.ExecutionException;
@@ -82,7 +81,7 @@ public abstract class ConsumerBase<T> extends HandlerState implements Consumer<T
     protected final ExecutorService externalPinnedExecutor;
     protected final ExecutorService internalPinnedExecutor;
     protected UnAckedMessageTracker unAckedMessageTracker;
-    final BlockingQueue<Message<T>> incomingMessages;
+    final GrowableArrayBlockingQueue<Message<T>> incomingMessages;
     protected ConcurrentOpenHashMap<MessageIdImpl, MessageIdImpl[]> unAckedChunkedMessageIdSequenceMap;
     protected final ConcurrentLinkedQueue<CompletableFuture<Message<T>>> pendingReceives;
     protected final int maxReceiverQueueSize;

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

@@ -1075,6 +1075,18 @@ private void closeConsumerTasks() {
         }
         negativeAcksTracker.close();
         stats.getStatTimeout().ifPresent(Timeout::cancel);
+        if (poolMessages) {
+            releasePooledMessagesAndStopAcceptNew();
+        }
+    }
+
+    /**
+     * If enabled pooled messages, we should release the messages after closing consumer and stop accept the new
+     * messages.
+     */
+    private void releasePooledMessagesAndStopAcceptNew() {
+        incomingMessages.terminate(message -> message.release());
+        clearIncomingMessages();
     }
 
     void activeConsumerChanged(boolean isActive) {

pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/GrowableArrayBlockingQueue.java

@@ -32,6 +32,7 @@
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.concurrent.locks.StampedLock;
 import java.util.function.Consumer;
+import javax.annotation.Nullable;
 
 /**
  * This implements a {@link BlockingQueue} backed by an array with no fixed capacity.
@@ -53,6 +54,10 @@ public class GrowableArrayBlockingQueue<T> extends AbstractQueue<T> implements B
             .newUpdater(GrowableArrayBlockingQueue.class, "size");
     private volatile int size = 0;
 
+    private volatile boolean terminated = false;
+
+    private volatile Consumer<T> itemAfterTerminatedHandler;
+
     public GrowableArrayBlockingQueue() {
         this(64);
     }
@@ -132,6 +137,13 @@ public void put(T e) {
         boolean wasEmpty = false;
 
         try {
+            if (terminated){
+                if (itemAfterTerminatedHandler != null) {
+                    itemAfterTerminatedHandler.accept(e);
+                }
+                return;
+            }
+
             if (SIZE_UPDATER.get(this) == data.length) {
                 expandArray();
             }
@@ -401,6 +413,27 @@ public String toString() {
         return sb.toString();
     }
 
+    /**
+     * Make the queue not accept new items. if there are still new data trying to enter the queue, it will be handed
+     * by {@param itemAfterTerminatedHandler}.
+     */
+    public void terminate(@Nullable Consumer<T> itemAfterTerminatedHandler) {
+        // After wait for the in-flight item enqueue, it means the operation of terminate is finished.
+        long stamp = tailLock.writeLock();
+        try {
+            terminated = true;
+            if (itemAfterTerminatedHandler != null) {
+                this.itemAfterTerminatedHandler = itemAfterTerminatedHandler;
+            }
+        } finally {
+            tailLock.unlockWrite(stamp);
+        }
+    }
+
+    public boolean isTerminated() {
+        return terminated;
+    }
+
     @SuppressWarnings("unchecked")
     private void expandArray() {
         // We already hold the tailLock