Repository: pinpoint-apm/pinpoint
Issue: [PinpiontWeb] fix a ActiveThread concurrency problem
Labels: bug

If multiple threads invokes WebSocketSession's sendMessage method at the same time, ; 
then it is possible to makes IllegalStateException.


Issue comments:

Pull request review comments:

Patch that resolved the issue:
diff --git a/rpc/src/main/java/com/navercorp/pinpoint/rpc/stream/LoggingStreamChannelMessageListener.java b/rpc/src/main/java/com/navercorp/pinpoint/rpc/stream/LoggingStreamChannelMessageListener.java
index bee01d05c1bd..0cffad4c72a2 100644
--- a/rpc/src/main/java/com/navercorp/pinpoint/rpc/stream/LoggingStreamChannelMessageListener.java
+++ b/rpc/src/main/java/com/navercorp/pinpoint/rpc/stream/LoggingStreamChannelMessageListener.java
@@ -52,7 +52,7 @@ static class Client implements ClientStreamChannelMessageListener {
 
         @Override
         public void handleStreamData(ClientStreamChannelContext streamChannelContext, StreamResponsePacket packet) {
-            LOGGER.info("handleStreamData StreamChannel:{}, Packet:{}", streamChannelContext, packet);
+            LOGGER.debug("handleStreamData StreamChannel:{}, Packet:{}", streamChannelContext, packet);
         }
 
         @Override
diff --git a/web/src/main/java/com/navercorp/pinpoint/web/websocket/ActiveThreadCountHandler.java b/web/src/main/java/com/navercorp/pinpoint/web/websocket/ActiveThreadCountHandler.java
index a721c8047912..3b2fb7a187b0 100644
--- a/web/src/main/java/com/navercorp/pinpoint/web/websocket/ActiveThreadCountHandler.java
+++ b/web/src/main/java/com/navercorp/pinpoint/web/websocket/ActiveThreadCountHandler.java
@@ -33,7 +33,6 @@
 import org.springframework.web.socket.WebSocketSession;
 import org.springframework.web.socket.handler.TextWebSocketHandler;
 
-import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
@@ -374,7 +373,7 @@ public void run() {
         private void closeSession(WebSocketSession session, CloseStatus status) {
             try {
                 session.close(status);
-            } catch (IOException e) {
+            } catch (Exception e) {
                 logger.warn(e.getMessage(), e);
             }
         }
@@ -382,8 +381,9 @@ private void closeSession(WebSocketSession session, CloseStatus status) {
         private void sendPingMessage(WebSocketSession session, TextMessage pingMessage) {
             try {
                 session.sendMessage(pingMessage);
-            } catch (IOException e) {
+            } catch (Exception e) {
                 logger.warn(e.getMessage(), e);
+                closeSession(session, CloseStatus.SERVER_ERROR);
             }
         }
 
diff --git a/web/src/main/java/com/navercorp/pinpoint/web/websocket/ActiveThreadCountResponseAggregator.java b/web/src/main/java/com/navercorp/pinpoint/web/websocket/ActiveThreadCountResponseAggregator.java
index 3b4fcee83726..ef0658552a33 100644
--- a/web/src/main/java/com/navercorp/pinpoint/web/websocket/ActiveThreadCountResponseAggregator.java
+++ b/web/src/main/java/com/navercorp/pinpoint/web/websocket/ActiveThreadCountResponseAggregator.java
@@ -29,7 +29,6 @@
 import org.springframework.web.socket.TextMessage;
 import org.springframework.web.socket.WebSocketSession;
 
-import java.io.IOException;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -37,6 +36,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * @Author Taejin Koo
@@ -48,6 +48,7 @@ public class ActiveThreadCountResponseAggregator implements PinpointWebSocketRes
     private static final String TIME_STAMP = "timeStamp";
 
     private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
     private final String applicationName;
     private final AgentService agentService;
     private final Timer timer;
@@ -59,6 +60,9 @@ public class ActiveThreadCountResponseAggregator implements PinpointWebSocketRes
     private final Object aggregatorLock = new Object();
     private final PinpointWebSocketMessageConverter messageConverter;
 
+    private final AtomicInteger flushCount = new AtomicInteger(0);
+    private final int flushLogRecordRate = 60;
+
     private volatile boolean isStopped = false;
     private WorkerActiveManager workerActiveManager;
 
@@ -201,7 +205,9 @@ public void response(AgentActiveThreadCount activeThreadCount) {
 
     @Override
     public void flush() throws Exception {
-        logger.info("flush started. applicationName:{}", applicationName);
+        if ((flushCount.getAndIncrement() % flushLogRecordRate) == 0) {
+            logger.info("flush started. applicationName:{}", applicationName);
+        }
 
         if (isStopped) {
             return;
@@ -234,12 +240,12 @@ private void flush0(AgentActiveThreadCountList activeThreadCountList) {
                 try {
                     logger.debug("flush webSocketSession:{}, response:{}", webSocketSession, responseTextMessage);
                     webSocketSession.sendMessage(responseTextMessage);
-                } catch (IOException e) {
-                    logger.warn(e.getMessage(), e);
+                } catch (Exception e) {
+                    logger.warn("failed while flush message(applicationName:{}, session:{}). Error:{}", webSocketSession, applicationName, e.getMessage(), e);
                 }
             }
         } catch (JsonProcessingException e) {
-            logger.warn("json convert failed. original:{}, message:{}.", resultMap, e.getMessage(), e);
+            logger.warn("failed while convert message. applicationName:{}, original:{}, message:{}.", applicationName, resultMap, e.getMessage(), e);
         }
     }
 


Test patch: