From 1ed33f207dc12fa81f680c9eda212f8e9c3d0f4e Mon Sep 17 00:00:00 2001
From: Lari Hotari <lhotari@users.noreply.github.com>
Date: Mon, 23 Aug 2021 21:15:51 +0300
Subject: [PATCH] [Broker] Call .release() when discarding entry to prevent
 direct memory leak (#11748)

(cherry picked from commit 7906bb5990a918adfc47ed0ab52063caabac2801)
---
 .../persistent/PersistentDispatcherSingleActiveConsumer.java    | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
index 4bae5b02a40f0..c4bea81b27077 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
@@ -175,7 +175,9 @@ public synchronized void internalReadEntriesComplete(final List<Entry> entries,
                 Entry entry = iterator.next();
                 byte[] key = peekStickyKey(entry.getDataBuffer());
                 Consumer consumer = stickyKeyConsumerSelector.select(key);
+                // Skip the entry if it's not for current active consumer.
                 if (consumer == null || currentConsumer != consumer) {
+                    entry.release();
                     iterator.remove();
                 }
             }