You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2023/03/06 01:26:00 UTC
[pulsar] branch branch-2.11 updated: [fix] [client] fix memory leak if enabled pooled messages (#19585)
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.11 by this push:
new adee8901f04 [fix] [client] fix memory leak if enabled pooled messages (#19585)
adee8901f04 is described below
commit adee8901f0461818c25a053385565cd0e1551391
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Sun Mar 5 10:08:30 2023 +0800
[fix] [client] fix memory leak if enabled pooled messages (#19585)
---
.../client/impl/BrokerClientIntegrationTest.java | 7 ++++-
.../apache/pulsar/client/impl/ConsumerBase.java | 5 ++--
.../apache/pulsar/client/impl/ConsumerImpl.java | 12 ++++++++
.../collections/GrowableArrayBlockingQueue.java | 33 ++++++++++++++++++++++
4 files changed, 54 insertions(+), 3 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java
index 0039fc92cb8..1210e14b23f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java
@@ -945,9 +945,14 @@ public class BrokerClientIntegrationTest extends ProducerConsumerBase {
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);
}
/**
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
index 4a60dad8bfc..ebd6a2e0ebf 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
@@ -29,7 +29,6 @@ import java.util.List;
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;
@@ -80,7 +79,9 @@ public abstract class ConsumerBase<T> extends HandlerState implements Consumer<T
protected final ExecutorProvider executorProvider;
protected final ExecutorService externalPinnedExecutor;
protected final ExecutorService internalPinnedExecutor;
- final BlockingQueue<Message<T>> incomingMessages;
+
+ protected UnAckedMessageTracker unAckedMessageTracker;
+ final GrowableArrayBlockingQueue<Message<T>> incomingMessages;
protected ConcurrentOpenHashMap<MessageIdImpl, MessageIdImpl[]> unAckedChunkedMessageIdSequenceMap;
protected final ConcurrentLinkedQueue<CompletableFuture<Message<T>>> pendingReceives;
protected final int maxReceiverQueueSize;
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
index 4d011ce4977..67341898e0e 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -1116,6 +1116,18 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
}
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) {
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/GrowableArrayBlockingQueue.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/GrowableArrayBlockingQueue.java
index d4afe6ecf63..e3a2c4e2f46 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/GrowableArrayBlockingQueue.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/GrowableArrayBlockingQueue.java
@@ -32,6 +32,7 @@ import java.util.concurrent.locks.Condition;
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 class GrowableArrayBlockingQueue<T> extends AbstractQueue<T> implements B
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 class GrowableArrayBlockingQueue<T> extends AbstractQueue<T> implements B
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