[pulsar-client] clean up MultiTopicsConsumerImpl reference on consume…

…r creation failure (#11754)
apache · Aug 24, 2021 · f154de7 · f154de7
1 parent 4e60de6
commit f154de7
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 3 deletions.
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
@@ -103,7 +103,7 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
     private final ReadWriteLock lock = new ReentrantReadWriteLock();
 
     private final ConsumerStatsRecorder stats;
-    private final UnAckedMessageTracker unAckedMessageTracker;
+    private UnAckedMessageTracker unAckedMessageTracker;
     private final ConsumerConfigurationData<T> internalConfig;
 
     private volatile BatchMessageIdImpl startMessageId = null;
@@ -543,7 +543,9 @@ public CompletableFuture<Void> unsubscribeAsync() {
     @Override
     public CompletableFuture<Void> closeAsync() {
         if (getState() == State.Closing || getState() == State.Closed) {
-            unAckedMessageTracker.close();
+            if (unAckedMessageTracker != null) {
+                unAckedMessageTracker.close();
+            }
             return CompletableFuture.completedFuture(null);
         }
         setState(State.Closing);
@@ -580,7 +582,14 @@ public CompletableFuture<Void> closeAsync() {
     }
 
     private void cleanupMultiConsumer() {
-        unAckedMessageTracker.close();
+        if (unAckedMessageTracker != null) {
+            unAckedMessageTracker.close();
+            unAckedMessageTracker = null;
+        }
+        if (partitionsAutoUpdateTimeout != null) {
+            partitionsAutoUpdateTimeout.cancel();
+            partitionsAutoUpdateTimeout = null;
+        }
         client.cleanupConsumer(this);
     }
 

diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/UnAckedMessageTracker.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/UnAckedMessageTracker.java
@@ -248,6 +248,7 @@ private void stop() {
         try {
             if (timeout != null && !timeout.isCancelled()) {
                 timeout.cancel();
+                timeout = null;
             }
             this.clear();
         } finally {