Cancel timeout timer in ClientCnx (apache#3422)

leizhiyuan · Jan 25, 2019 · 14390da · 14390da
1 parent 9184c34
commit 14390da
Showing 1 changed file with 7 additions and 1 deletion.
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
@@ -29,6 +29,7 @@
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
@@ -122,6 +123,8 @@ public class ClientCnx extends PulsarHandler {
     private boolean isTlsHostnameVerificationEnable;
     private DefaultHostnameVerifier hostnameVerifier;
 
+    private final ScheduledFuture<?> timeoutTask;
+
     enum State {
         None, SentConnectFrame, Ready, Failed
     }
@@ -155,7 +158,8 @@ public ClientCnx(ClientConfigurationData conf, EventLoopGroup eventLoopGroup, in
         this.isTlsHostnameVerificationEnable = conf.isTlsHostnameVerificationEnable();
         this.hostnameVerifier = new DefaultHostnameVerifier();
         this.protocolVersion = protocolVersion;
-        this.eventLoopGroup.scheduleAtFixedRate(() -> checkRequestTimeout(), operationTimeoutMs, operationTimeoutMs, TimeUnit.MILLISECONDS);
+        this.timeoutTask = this.eventLoopGroup.scheduleAtFixedRate(() -> checkRequestTimeout(), operationTimeoutMs,
+                operationTimeoutMs, TimeUnit.MILLISECONDS);
     }
 
     @Override
@@ -223,6 +227,8 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
 
         producers.clear();
         consumers.clear();
+
+        timeoutTask.cancel(true);
     }
 
     // Command Handlers