[Broker] Call .release() when discarding entry to prevent direct memory leak
@@ -175,7 +175,9 @@ public class PersistentDispatcherSingleActiveConsumer extends AbstractDispatcher
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();
}
}
