From e97d53283a222645b9b2c5821f669975a68fe449 Mon Sep 17 00:00:00 2001 From: Lari Hotari 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 93c2d2547ab16..ab4a6f5552c32 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 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(); } }