Add close method in the class of NegativeAcksTracker (apache#12469)

AlphaWang · Oct 31, 2021 · 3694aa1 · 3694aa1
1 parent 7f60bcd
commit 3694aa1
Show file tree

Hide file tree

Showing 3 changed files with 30 additions and 1 deletion.
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
@@ -978,6 +978,7 @@ private void closeConsumerTasks() {
         if (batchReceiveTimeout != null) {
             batchReceiveTimeout.cancel();
         }
+        negativeAcksTracker.close();
         stats.getStatTimeout().ifPresent(Timeout::cancel);
     }
 

diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/NegativeAcksTracker.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/NegativeAcksTracker.java
@@ -21,6 +21,7 @@
 import io.netty.util.Timeout;
 import io.netty.util.Timer;
 
+import java.io.Closeable;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Set;
@@ -30,7 +31,7 @@
 import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
 import static org.apache.pulsar.client.impl.UnAckedMessageTracker.addChunkedMessageIdsAndRemoveFromSequenceMap;
 
-class NegativeAcksTracker {
+class NegativeAcksTracker implements Closeable {
 
     private HashMap<MessageId, Long> nackedMessages = null;
 
@@ -93,4 +94,17 @@ public synchronized void add(MessageId messageId) {
             this.timeout = timer.newTimeout(this::triggerRedelivery, timerIntervalNanos, TimeUnit.NANOSECONDS);
         }
     }
+
+    @Override
+    public synchronized void close() {
+        if (timeout != null && !timeout.isCancelled()) {
+            timeout.cancel();
+            timeout = null;
+        }
+
+        if (nackedMessages != null) {
+            nackedMessages.clear();
+            nackedMessages = null;
+        }
+    }
 }
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerImplTest.java
@@ -187,4 +187,18 @@ public void testBatchReceiveAsyncCanBeCancelled() {
         // then
         Assert.assertFalse(consumer.hasPendingBatchReceive());
     }
+
+    @Test
+    public void testClose() {
+        Exception checkException = null;
+        try {
+            if (consumer != null) {
+                consumer.negativeAcknowledge(new MessageIdImpl(-1, -1, -1));
+                consumer.close();
+            }
+        } catch (Exception e) {
+            checkException = e;
+        }
+        Assert.assertNull(checkException);
+    }
 }