Repository: pinpoint-apm/pinpoint
Issue: Timer thread leak when client reconnect to server. 
Labels: bug

Timer thread leak when client reconnect to server. 

Client auto reconnect when socket disconnect from peer.
In this situation client does not clean up running timer thread. 

It can be problem when agent running for a long time or collector restart frequently.


Issue comments:

Pull request review comments:

Patch that resolved the issue:
diff --git a/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocketHandler.java b/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocketHandler.java
index f3018f3bc0dd..2f4761836b68 100644
--- a/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocketHandler.java
+++ b/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocketHandler.java
@@ -453,30 +453,6 @@ public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws
 
     }
 
-    @Override
-    public void channelClosed(final ChannelHandlerContext ctx, final ChannelStateEvent e) throws Exception {
-        final int currentState = state.getState();
-        if (currentState == State.CLOSED) {
-            logger.debug("channelClosed() normal. state:{} {}", state.getString(currentState), e.getChannel());
-            return;
-        } else if(currentState == State.INIT_RECONNECT){
-            logger.debug("channelClosed() reconnect fail. state:{} {}", state.getString(currentState), e.getChannel());
-        } else if (state.isRun(currentState) || currentState == State.RECONNECT) {
-            // abnormal closed from here
-            if (state.isRun(currentState)) {
-                logger.debug("change state=reconnect");
-                state.setState(State.RECONNECT);
-            }
-            logger.info("channelClosed() UnexpectedChannelClosed. state:{} try reconnect channel:{}, connectSocketAddress:{}", state.getString(), e.getChannel(), connectSocketAddress);
-
-            this.pinpointSocketFactory.reconnect(this.pinpointSocket, this.connectSocketAddress);
-            return;
-        } else {
-            logger.info("channelClosed() UnexpectedChannelClosed. state:{} {}", state.getString(currentState), e.getChannel());
-        }
-        releaseResource();
-    }
-
     private void ensureOpen() {
         final int currentState = state.getState();
         if (state.isRun(currentState)) {
@@ -503,8 +479,6 @@ boolean isClosed() {
     public void close() {
         logger.debug("close() call");
         
-        this.handshaker.handshakeAbort();
-        
         int currentState = this.state.getState();
         if (currentState == State.CLOSED) {
             logger.debug("already close()");
@@ -522,36 +496,44 @@ public void close() {
         // when you release resources, you need to clear messages about request/response and stream channel. need to handle reversely?
         // handling timer is unclear so just make and enhance later.
 
+        closeStreamChannelManager(channel);
         sendClosedPacket(channel);
-        releaseResource();
+        
         logger.debug("channel.close()");
-
-        ChannelFuture channelFuture = channel.close();
-        channelFuture.addListener(new WriteFailFutureListener(logger, "close() event failed.", "close() event success."));
-        channelFuture.awaitUninterruptibly();
+        ChannelFuture closeFuture = channel.close();
+        closeFuture.addListener(new WriteFailFutureListener(logger, "close() event failed.", "close() event success."));
+        closeFuture.awaitUninterruptibly();
         logger.debug("close() complete");
     }
 
+    private void closeStreamChannelManager(Channel channel) {
+        if (!channel.isConnected()) {
+            logger.debug("channel already closed. skip closeStreamChannelManager() {}", channel);
+            return;
+        }
+
+        // stream channel clear and send stream close packet 
+        SocketHandlerContext context = getChannelContext(channel);
+        if (context != null) {
+            context.getStreamChannelManager().close();
+        }
+    }
+
     private void releaseResource() {
         logger.debug("releaseResource()");
+
+        this.handshaker.handshakeAbort();
         this.requestManager.close();
-        
-        if (this.channel != null) {
-            SocketHandlerContext context = getChannelContext(channel);
-            if (context != null) {
-                context.getStreamChannelManager().close();
-            }
-        }
-        
         this.channelTimer.stop();
     }
-
+    
     private void sendClosedPacket(Channel channel) {
         if (!channel.isConnected()) {
             logger.debug("channel already closed. skip sendClosedPacket() {}", channel);
             return;
         }
-       logger.debug("write ClientClosePacket");
+        
+        logger.debug("write ClientClosePacket");
         ClientClosePacket clientClosePacket = new ClientClosePacket();
         ChannelFuture write = channel.write(clientClosePacket);
         write.addListener(new ChannelFutureListener() {
@@ -567,7 +549,28 @@ public void operationComplete(ChannelFuture future) throws Exception {
         write.awaitUninterruptibly(3000, TimeUnit.MILLISECONDS);
     }
 
+    @Override
+    public void channelClosed(final ChannelHandlerContext ctx, final ChannelStateEvent e) throws Exception {
+        final int currentState = state.getState();
+        if (State.CLOSED == currentState) {
+            logger.debug("channelClosed() normal. state:{} {}", state.getString(currentState), e.getChannel());
+        } else if(State.INIT_RECONNECT == currentState){
+            logger.debug("channelClosed() reconnect fail. state:{} {}", state.getString(currentState), e.getChannel());
+        } else if (state.isRun(currentState) || State.RECONNECT == currentState) {
+            // abnormal closed from here
+            if (state.isRun(currentState)) {
+                logger.debug("change state=reconnect");
+                state.setState(State.RECONNECT);
+            }
+            logger.info("channelClosed() UnexpectedChannelClosed. state:{} try reconnect channel:{}, connectSocketAddress:{}", state.getString(), e.getChannel(), connectSocketAddress);
 
+            this.pinpointSocketFactory.reconnect(this.pinpointSocket, this.connectSocketAddress);
+        } else {
+            logger.info("channelClosed() UnexpectedChannelClosed. state:{} {}", state.getString(currentState), e.getChannel());
+        }
+        releaseResource();
+    }
+    
     @Override
     public String toString() {
         final StringBuilder sb = new StringBuilder("PinpointSocketHandler{");


Test patch:
diff --git a/rpc/src/test/java/com/navercorp/pinpoint/rpc/client/ReconnectTest.java b/rpc/src/test/java/com/navercorp/pinpoint/rpc/client/ReconnectTest.java
index fe40921cb1ae..d307e0b620f6 100644
--- a/rpc/src/test/java/com/navercorp/pinpoint/rpc/client/ReconnectTest.java
+++ b/rpc/src/test/java/com/navercorp/pinpoint/rpc/client/ReconnectTest.java
@@ -17,11 +17,14 @@
 package com.navercorp.pinpoint.rpc.client;
 
 import java.io.IOException;
+import java.lang.management.ManagementFactory;
+import java.lang.management.ThreadMXBean;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.BeforeClass;
+import org.junit.Ignore;
 import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -110,6 +113,45 @@ public void reconnectPerformed(PinpointSocket socket) {
         
         Assert.assertTrue(reconnectPerformed.get());
     }
+    
+    // it takes very long time. 
+    // @Test
+    @Ignore
+    public void reconnectStressTest() throws IOException, InterruptedException {
+        int count = 3;
+        
+        ThreadMXBean tbean = ManagementFactory.getThreadMXBean();
+
+        int threadCount = tbean.getThreadCount();
+        for (int i = 0; i < count; i++) {
+            logger.info((i + 1) + "th's start.");
+            
+            PinpointServerSocket serverSocket = PinpointRPCTestUtils.createServerSocket(bindPort, new TestSeverMessageListener());
+            PinpointSocket socket = socketFactory.connect("localhost", bindPort);
+            PinpointRPCTestUtils.close(serverSocket);
+
+            logger.info("server.close()---------------------------");
+            Thread.sleep(10000);
+
+            serverSocket = PinpointRPCTestUtils.createServerSocket(bindPort, new TestSeverMessageListener());
+            logger.info("bind server---------------------------");
+
+            Thread.sleep(10000);
+            logger.info("request server---------------------------");
+            byte[] randomByte = TestByteUtils.createRandomByte(10);
+            byte[] response = PinpointRPCTestUtils.request(socket, randomByte);
+
+            Assert.assertArrayEquals(randomByte, response);
+
+            PinpointRPCTestUtils.close(socket);
+            PinpointRPCTestUtils.close(serverSocket);
+        }
+        
+        Thread.sleep(10000);
+
+        Assert.assertEquals(threadCount, tbean.getThreadCount());
+    }
+
 
     @Test
     public void scheduledConnect() throws IOException, InterruptedException {