Ensure the timer tasks in the producer and consumer are cleanup after…

… closed (apache#7124)

### Motivation

Ensure the timer tasks in the producer and consumer are cleanup after closed

### Verifying this change

Unit test added

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java

@@ -74,11 +74,11 @@ public void testGetPolicy() throws ExecutionException, InterruptedException, Top
                 .maxConsumerPerTopic(10)
                 .build();
         systemTopicBasedTopicPoliciesService.updateTopicPoliciesAsync(TOPIC1, initPolicy).get();
-        Assert.assertTrue(systemTopicBasedTopicPoliciesService.getPoliciesCacheInit(TOPIC1.getNamespaceObject()));
 
         // Wait for all topic policies updated.
         Thread.sleep(3000);
 
+        Assert.assertTrue(systemTopicBasedTopicPoliciesService.getPoliciesCacheInit(TOPIC1.getNamespaceObject()));
         // Assert broker is cache all topic policies
         Assert.assertEquals(systemTopicBasedTopicPoliciesService.getTopicPolicies(TOPIC1).getMaxConsumerPerTopic().intValue(), 10);
 

pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerCleanupTest.java

@@ -0,0 +1,58 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.pulsar.client.api;
+
+import io.netty.util.HashedWheelTimer;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import java.util.UUID;
+
+public class ConsumerCleanupTest extends ProducerConsumerBase {
+
+    @BeforeClass
+    @Override
+    protected void setup() throws Exception {
+        super.internalSetup();
+        super.producerBaseSetup();
+    }
+
+    @AfterClass
+    @Override
+    protected void cleanup() throws Exception {
+        super.internalCleanup();
+    }
+
+    @Test
+    public void testAllTimerTaskShouldCanceledAfterConsumerClosed() throws PulsarClientException, InterruptedException {
+        PulsarClient pulsarClient = newPulsarClient(lookupUrl.toString(), 1);
+        Consumer<byte[]> consumer = pulsarClient.newConsumer()
+                .topic("persistent://public/default/" + UUID.randomUUID().toString())
+                .subscriptionName("test")
+                .subscribe();
+        consumer.close();
+        Thread.sleep(2000);
+        HashedWheelTimer timer = (HashedWheelTimer) ((PulsarClientImpl) pulsarClient).timer();
+        Assert.assertEquals(timer.pendingTimeouts(), 0);
+    }
+}

pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerCleanupTest.java

@@ -0,0 +1,59 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api;
+
+import io.netty.util.HashedWheelTimer;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+public class ProducerCleanupTest extends ProducerConsumerBase {
+
+    @BeforeMethod
+    @Override
+    protected void setup() throws Exception {
+        super.internalSetup();
+        super.producerBaseSetup();
+    }
+
+    @AfterMethod
+    @Override
+    protected void cleanup() throws Exception {
+        super.internalCleanup();
+    }
+
+    @Test
+    public void testAllTimerTaskShouldCanceledAfterProducerClosed() throws PulsarClientException, InterruptedException {
+        Producer<byte[]> producer = pulsarClient.newProducer()
+                .topic("persistent://public/default/" + UUID.randomUUID().toString())
+                .sendTimeout(1, TimeUnit.SECONDS)
+                .create();
+        producer.close();
+        Thread.sleep(2000);
+        HashedWheelTimer timer = (HashedWheelTimer) ((PulsarClientImpl) pulsarClient).timer();
+        Assert.assertEquals(timer.pendingTimeouts(), 0);
+    }
+}

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicDoesNotExistsTest.java

@@ -68,11 +68,20 @@ public void testCreateProducerOnNotExistsTopic() throws PulsarClientException, I
         pulsarClient.close();
     }
 
-    @Test(expectedExceptions = PulsarClientException.TopicDoesNotExistException.class)
-    public void testCreateConsumerOnNotExistsTopic() throws PulsarClientException {
-        pulsarClient.newConsumer()
-                .topic("persistent://public/default/" + UUID.randomUUID().toString())
-                .subscriptionName("test")
-                .subscribe();
+    @Test
+    public void testCreateConsumerOnNotExistsTopic() throws PulsarClientException, InterruptedException {
+        PulsarClient pulsarClient = newPulsarClient(lookupUrl.toString(), 1);
+        try {
+            pulsarClient.newConsumer()
+                    .topic("persistent://public/default/" + UUID.randomUUID().toString())
+                    .subscriptionName("test")
+                    .subscribe();
+            Assert.fail("Create consumer should failed while topic does not exists.");
+        } catch (PulsarClientException ignore) {
+        }
+        Thread.sleep(2000);
+        HashedWheelTimer timer = (HashedWheelTimer) ((PulsarClientImpl) pulsarClient).timer();
+        Assert.assertEquals(timer.pendingTimeouts(), 0);
+        Assert.assertEquals(((PulsarClientImpl) pulsarClient).consumersCount(), 0);
     }
 }

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

@@ -832,6 +832,7 @@ public void connectionOpened(final ClientCnx cnx) {
                 //    auto-topic-creation set to false
                 // No more retries are needed in this case.
                 setState(State.Failed);
+                closeConsumerTasks();
                 client.cleanupConsumer(this);
                 log.warn("[{}][{}] Closed consumer because topic does not exist anymore {}", topic, subscription, cnx.channel().remoteAddress());
             } else {
@@ -977,6 +978,10 @@ private void closeConsumerTasks() {
         }
 
         acknowledgmentsGroupingTracker.close();
+        if (batchReceiveTimeout != null) {
+            batchReceiveTimeout.cancel();
+        }
+        stats.getStatTimeout().ifPresent(Timeout::cancel);
     }
 
     private void failPendingReceive() {