[#9023] Polishing

batch/src/main/java/com/navercorp/pinpoint/batch/job/AgentCountProcessor.java

@@ -41,11 +41,9 @@ public AgentCountStatistics process(ApplicationAgentsList item) throws Exception
     }
 
     private int getAgentCount(List<ApplicationAgentList> applicationAgentLists) {
-        int agentCount = 0;
-        for (ApplicationAgentList applicationAgentList : applicationAgentLists) {
-            agentCount += applicationAgentList.getAgentInfos().size();
-        }
-        return agentCount;
+        return applicationAgentLists.stream()
+                .mapToInt(applicationAgentList -> applicationAgentList.getAgentInfos().size())
+                .sum();
     }
 
 }

collector/src/main/java/com/navercorp/pinpoint/collector/cluster/ClusterPoint.java

@@ -16,6 +16,7 @@
 
 package com.navercorp.pinpoint.collector.cluster;
 
+import com.navercorp.pinpoint.common.server.cluster.ClusterKey;
 import com.navercorp.pinpoint.rpc.Future;
 
 import com.navercorp.pinpoint.rpc.ResponseMessage;
@@ -28,7 +29,7 @@ public interface ClusterPoint<M> {
 
     Future<ResponseMessage> request(M request);
 
-    AgentInfo getDestAgentInfo();
+    ClusterKey getDestClusterKey();
 
     boolean isSupportCommand(TBase<?, ?> command);
 

collector/src/main/java/com/navercorp/pinpoint/collector/cluster/ClusterPointRepository.java

@@ -17,7 +17,7 @@
 package com.navercorp.pinpoint.collector.cluster;
 
 import com.navercorp.pinpoint.collector.receiver.grpc.PinpointGrpcServer;
-import com.navercorp.pinpoint.common.server.cluster.AgentInfoKey;
+import com.navercorp.pinpoint.common.server.cluster.ClusterKey;
 import com.navercorp.pinpoint.rpc.common.SocketStateCode;
 import com.navercorp.pinpoint.rpc.server.PinpointServer;
 
@@ -35,13 +35,12 @@ public class ClusterPointRepository<T extends ClusterPoint<?>> implements Cluste
 
     private final Logger logger = LogManager.getLogger(this.getClass());
 
-    private final Map<AgentInfoKey, Set<T>> clusterPointRepository = new HashMap<>();
+    private final Map<ClusterKey, Set<T>> clusterPointRepository = new HashMap<>();
 
     public boolean addAndIsKeyCreated(T clusterPoint) {
-        AgentInfo destAgentInfo = clusterPoint.getDestAgentInfo();
-        AgentInfoKey key = destAgentInfo.getAgentKey();
+        ClusterKey destClusterKey = clusterPoint.getDestClusterKey();
         synchronized (this) {
-            final Set<T> clusterPointSet = clusterPointRepository.get(key);
+            final Set<T> clusterPointSet = clusterPointRepository.get(destClusterKey);
             if (clusterPointSet != null) {
                 clusterPointSet.add(clusterPoint);
 
@@ -50,22 +49,21 @@ public boolean addAndIsKeyCreated(T clusterPoint) {
                 Set<T> newSet = new HashSet<>();
                 newSet.add(clusterPoint);
 
-                clusterPointRepository.put(key, newSet);
+                clusterPointRepository.put(destClusterKey, newSet);
                 return true;
             }
         }
     }
 
     public boolean removeAndGetIsKeyRemoved(T clusterPoint) {
-        AgentInfo destAgentInfo = clusterPoint.getDestAgentInfo();
-        AgentInfoKey key = destAgentInfo.getAgentKey();
+        ClusterKey destClusterKey = clusterPoint.getDestClusterKey();
         synchronized (this) {
-            final Set<T> clusterPointSet = clusterPointRepository.get(key);
+            final Set<T> clusterPointSet = clusterPointRepository.get(destClusterKey);
             if (clusterPointSet != null) {
                 clusterPointSet.remove(clusterPoint);
 
                 if (clusterPointSet.isEmpty()) {
-                    clusterPointRepository.remove(key);
+                    clusterPointRepository.remove(destClusterKey);
                     return true;
                 }
             }
@@ -85,12 +83,12 @@ public List<T> getClusterPointList() {
         }
     }
 
-    public Set<AgentInfoKey> getAvailableAgentKeyList() {
+    public Set<ClusterKey> getAvailableAgentKeyList() {
         synchronized (this) {
-            Set<AgentInfoKey> availableAgentKeySet = new HashSet<>(clusterPointRepository.size());
+            Set<ClusterKey> availableAgentKeySet = new HashSet<>(clusterPointRepository.size());
 
-            for (Map.Entry<AgentInfoKey, Set<T>> entry : clusterPointRepository.entrySet()) {
-                final AgentInfoKey key = entry.getKey();
+            for (Map.Entry<ClusterKey, Set<T>> entry : clusterPointRepository.entrySet()) {
+                final ClusterKey key = entry.getKey();
                 final Set<T> clusterPointSet = entry.getValue();
                 for (T clusterPoint : clusterPointSet) {
                     if (clusterPoint instanceof ThriftAgentConnection) {

collector/src/main/java/com/navercorp/pinpoint/collector/cluster/GrpcAgentConnection.java

@@ -18,6 +18,7 @@
 
 import com.google.protobuf.GeneratedMessageV3;
 import com.navercorp.pinpoint.collector.receiver.grpc.PinpointGrpcServer;
+import com.navercorp.pinpoint.common.server.cluster.ClusterKey;
 import com.navercorp.pinpoint.profiler.context.grpc.CommandThriftToGrpcMessageConverter;
 import com.navercorp.pinpoint.rpc.DefaultFuture;
 import com.navercorp.pinpoint.rpc.Future;
@@ -72,8 +73,8 @@ public ClientStreamChannel openStream(TBase<?, ?> request, ClientStreamChannelEv
     }
 
     @Override
-    public AgentInfo getDestAgentInfo() {
-        return pinpointGrpcServer.getAgentInfo();
+    public ClusterKey getDestClusterKey() {
+        return pinpointGrpcServer.getClusterKey();
     }
 
     @Override
@@ -92,7 +93,7 @@ public PinpointGrpcServer getPinpointGrpcServer() {
 
     @Override
     public int hashCode() {
-        return pinpointGrpcServer.getAgentInfo().hashCode();
+        return pinpointGrpcServer.getClusterKey().hashCode();
     }
 
     @Override

collector/src/main/java/com/navercorp/pinpoint/collector/cluster/ThriftAgentConnection.java

@@ -16,6 +16,7 @@
 
 package com.navercorp.pinpoint.collector.cluster;
 
+import com.navercorp.pinpoint.common.server.cluster.ClusterKey;
 import com.navercorp.pinpoint.rpc.Future;
 import com.navercorp.pinpoint.rpc.ResponseMessage;
 import com.navercorp.pinpoint.rpc.server.ChannelProperties;
@@ -34,32 +35,29 @@ public class ThriftAgentConnection implements ClusterPoint<byte[]> {
 
     private final PinpointServer pinpointServer;
 
-    private final AgentInfo agentInfo;
-
+    private final ClusterKey clusterKey;
+    private final TCommandTypeVersion commandTypeVersion;
     private final List<TCommandType> supportCommandList;
 
     public static ClusterPoint<byte[]> newClusterPoint(PinpointServer pinpointServer, ChannelProperties channelProperties) {
-        AgentInfo agentInfo = newAgentInfo(channelProperties);
+        ClusterKey agentInfo = newClusterKey(channelProperties);
+        TCommandTypeVersion commandTypeVersion = TCommandTypeVersion.getVersion(channelProperties.getAgentVersion());
         List<TCommandType> supportCommandList = SupportedCommandUtils.newSupportCommandList(channelProperties.getSupportCommand());
-        return new ThriftAgentConnection(pinpointServer, agentInfo, supportCommandList);
+        return new ThriftAgentConnection(pinpointServer, agentInfo, commandTypeVersion, supportCommandList);
     }
 
-    public ThriftAgentConnection(PinpointServer pinpointServer, AgentInfo agentInfo, List<TCommandType> supportCommandList) {
+    public ThriftAgentConnection(PinpointServer pinpointServer, ClusterKey clusterKey, TCommandTypeVersion commandTypeVersion, List<TCommandType> supportCommandList) {
         this.pinpointServer = Objects.requireNonNull(pinpointServer, "pinpointServer");
-        this.agentInfo = Objects.requireNonNull(agentInfo, "agentInfo");
+        this.clusterKey = Objects.requireNonNull(clusterKey, "clusterKey");
+        this.commandTypeVersion = Objects.requireNonNull(commandTypeVersion, "commandTypeVersion");
         this.supportCommandList = Objects.requireNonNull(supportCommandList, "supportCommandList");
     }
 
-    private static AgentInfo newAgentInfo(ChannelProperties channelProperties) {
+    private static ClusterKey newClusterKey(ChannelProperties channelProperties) {
         String applicationName = channelProperties.getApplicationName();
-
         String agentId = channelProperties.getAgentId();
-
         long startTimeStamp = channelProperties.getStartTime();
-
-        String version = channelProperties.getAgentVersion();
-
-        return new AgentInfo(applicationName, agentId, startTimeStamp, version);
+        return new ClusterKey(applicationName, agentId, startTimeStamp);
     }
 
     @Override
@@ -68,8 +66,8 @@ public Future<ResponseMessage> request(byte[] payload) {
     }
 
     @Override
-    public AgentInfo getDestAgentInfo() {
-        return agentInfo;
+    public ClusterKey getDestClusterKey() {
+        return clusterKey;
     }
 
     @Override
@@ -80,8 +78,7 @@ public boolean isSupportCommand(TBase<?, ?> command) {
             }
         }
 
-        TCommandTypeVersion commandVersion = TCommandTypeVersion.getVersion(agentInfo.getVersion());
-        if (commandVersion.isSupportCommand(command)) {
+        if (commandTypeVersion.isSupportCommand(command)) {
             return true;
         }
 
@@ -97,7 +94,7 @@ public String toString() {
         StringBuilder log = new StringBuilder(32);
         log.append(this.getClass().getSimpleName());
         log.append("(");
-        log.append(agentInfo);
+        log.append(clusterKey);
         log.append(", supportCommandList:");
         log.append(supportCommandList);
         log.append(", pinpointServer:");
@@ -109,7 +106,7 @@ public String toString() {
 
     @Override
     public int hashCode() {
-        return agentInfo.hashCode();
+        return clusterKey.hashCode();
     }
 
     @Override

collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/AbstractRouteHandler.java

@@ -16,9 +16,9 @@
 
 package com.navercorp.pinpoint.collector.cluster.route;
 
-import com.navercorp.pinpoint.collector.cluster.AgentInfo;
 import com.navercorp.pinpoint.collector.cluster.ClusterPoint;
 import com.navercorp.pinpoint.collector.cluster.ClusterPointLocator;
+import com.navercorp.pinpoint.common.server.cluster.ClusterKey;
 import com.navercorp.pinpoint.thrift.dto.command.TCommandTransfer;
 import com.navercorp.pinpoint.thrift.dto.command.TCommandTransferResponse;
 import com.navercorp.pinpoint.thrift.dto.command.TRouteResult;
@@ -46,12 +46,13 @@ protected ClusterPoint<?> findClusterPoint(TCommandTransfer deliveryCommand) {
         String applicationName = deliveryCommand.getApplicationName();
         String agentId = deliveryCommand.getAgentId();
         long startTimeStamp = deliveryCommand.getStartTime();
+        final ClusterKey sourceKey = new ClusterKey(applicationName, agentId, startTimeStamp);
 
         List<ClusterPoint<?>> result = new ArrayList<>();
 
         for (ClusterPoint<?> targetClusterPoint : targetClusterPointLocator.getClusterPointList()) {
-            AgentInfo destAgentInfo = targetClusterPoint.getDestAgentInfo();
-            if (destAgentInfo.equals(applicationName, agentId, startTimeStamp)) {
+            ClusterKey destAgentInfo = targetClusterPoint.getDestClusterKey();
+            if (destAgentInfo.equals(sourceKey)) {
                 result.add(targetClusterPoint);
             }
         }