From b9d4383fdd9ead429ba065cb1e1763d0ba0116bb 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 5904c96342dfa..00b549d31c60d 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 @@ -224,7 +224,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(); } }