Repository: pinpoint-apm/pinpoint
Issue: Fix executing fail networktest script
Labels: bug


#### Description
Sometimes "icmp port unreachble error" occure when executing networktest script.


Issue comments:

Pull request review comments:

Patch that resolved the issue:
diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/BaseUDPHandlerFactory.java b/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/BaseUDPHandlerFactory.java
index 3b40b32f1add..df6aad145185 100644
--- a/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/BaseUDPHandlerFactory.java
+++ b/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/BaseUDPHandlerFactory.java
@@ -27,6 +27,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.net.DatagramPacket;
+import java.net.DatagramSocket;
 import java.net.InetAddress;
 import java.net.SocketAddress;
 import java.net.UnknownHostException;
@@ -103,7 +104,7 @@ private DispatchPacket() {
         }
 
         @Override
-        public void receive(T packet) {
+        public void receive(DatagramSocket localSocket, T packet) {
             if (isIgnoreAddress(packet.getAddress())) {
                 return;
             }
@@ -114,7 +115,7 @@ public void receive(T packet) {
             
             try {
                 tBase = deserializer.deserialize(packet.getData());
-                if (filter.filter(tBase, socketAddress) == TBaseFilter.BREAK) {
+                if (filter.filter(localSocket, tBase, socketAddress) == TBaseFilter.BREAK) {
                     return;
                 }
                 // dispatch signifies business logic execution
diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/ChunkedUDPPacketHandlerFactory.java b/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/ChunkedUDPPacketHandlerFactory.java
index ef20dd7a87b2..db96f377e57a 100644
--- a/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/ChunkedUDPPacketHandlerFactory.java
+++ b/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/ChunkedUDPPacketHandlerFactory.java
@@ -64,7 +64,7 @@ private DispatchPacket() {
         }
 
         @Override
-        public void receive(T packet) {
+        public void receive(DatagramSocket localSocket, T packet) {
             final ChunkHeaderTBaseDeserializer deserializer = deserializerFactory.createDeserializer();
             try {
                 List<TBase<?, ?>> list = deserializer.deserialize(packet.getData(), packet.getOffset(), packet.getLength());
@@ -73,7 +73,7 @@ public void receive(T packet) {
                 }
 
                 for (TBase<?, ?> tBase : list) {
-                    if (filter.filter(tBase, packet.getSocketAddress()) == TBaseFilter.BREAK) {
+                    if (filter.filter(localSocket, tBase, packet.getSocketAddress()) == TBaseFilter.BREAK) {
                         return;
                     }
                     // dispatch signifies business logic execution
diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/NetworkAvailabilityCheckPacketFilter.java b/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/NetworkAvailabilityCheckPacketFilter.java
index 7b5bf4bb0a55..aebf1c3fff8c 100644
--- a/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/NetworkAvailabilityCheckPacketFilter.java
+++ b/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/NetworkAvailabilityCheckPacketFilter.java
@@ -31,31 +31,23 @@
 public class NetworkAvailabilityCheckPacketFilter<T extends SocketAddress> implements TBaseFilter<T>, DisposableBean {
     private final Logger logger = LoggerFactory.getLogger(this.getClass());
 
-    private final DatagramSocket socket;
-
     public NetworkAvailabilityCheckPacketFilter() {
-        try {
-            this.socket = new DatagramSocket();
-            logger.info("port:{}", this.socket.getLocalAddress());
-        } catch (SocketException ex) {
-            throw new RuntimeException("socket create fail. error:" + ex.getMessage(), ex);
-        }
     }
 
     @Override
-    public boolean filter(TBase<?, ?> tBase, T remoteHostAddress) {
+    public boolean filter(DatagramSocket localSocket, TBase<?, ?> tBase, T remoteHostAddress) {
         // Network port availability check packet
         if (tBase instanceof NetworkAvailabilityCheckPacket) {
             if (logger.isInfoEnabled()) {
                 logger.info("received udp network availability check packet. remoteAddress:{}", remoteHostAddress);
             }
-            responseOK(remoteHostAddress);
+            responseOK(localSocket, remoteHostAddress);
             return BREAK;
         }
         return CONTINUE;
     }
 
-    private void responseOK(T remoteHostAddress) {
+    private void responseOK(DatagramSocket socket, T remoteHostAddress) {
         try {
             byte[] okBytes = NetworkAvailabilityCheckPacket.DATA_OK;
             DatagramPacket pongPacket = new DatagramPacket(okBytes, okBytes.length, remoteHostAddress);
@@ -70,12 +62,6 @@ private void responseOK(T remoteHostAddress) {
 
     @Override
     public void destroy() throws Exception {
-        if (socket!= null) {
-            try {
-                socket.close();
-            } catch (Exception e) {
-                logger.warn("socket.close() error:" + e.getMessage(), e);
-            }
-        }
     }
+
 }
diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/PacketHandler.java b/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/PacketHandler.java
index 88fbf1dee902..5bb26a0f557c 100644
--- a/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/PacketHandler.java
+++ b/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/PacketHandler.java
@@ -16,9 +16,13 @@
 
 package com.navercorp.pinpoint.collector.receiver.udp;
 
+import java.net.DatagramSocket;
+
 /**
  * @author emeroad
  */
 public interface PacketHandler<T> {
-    void receive(T packet);
+
+    void receive(DatagramSocket localSocket, T packet);
+
 }
diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/PooledPacketWrap.java b/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/PooledPacketWrap.java
index 57afcead1ff6..a7f57923d04a 100644
--- a/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/PooledPacketWrap.java
+++ b/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/PooledPacketWrap.java
@@ -19,21 +19,27 @@
 import com.navercorp.pinpoint.collector.util.PooledObject;
 
 import java.net.DatagramPacket;
+import java.net.DatagramSocket;
 
 /**
  * @author emeroad
  */
 public class PooledPacketWrap implements Runnable {
+    private final DatagramSocket localSocket;
     private final PacketHandler<DatagramPacket> packetHandler;
     private final PooledObject<DatagramPacket> pooledObject;
 
-    public PooledPacketWrap(PacketHandler<DatagramPacket> packetHandler, PooledObject<DatagramPacket> pooledObject) {
+    public PooledPacketWrap(DatagramSocket localSocket, PacketHandler<DatagramPacket> packetHandler, PooledObject<DatagramPacket> pooledObject) {
+        if (localSocket == null) {
+            throw new NullPointerException("localSocket may not be null");
+        }
         if (packetHandler == null) {
             throw new NullPointerException("packetReceiveHandler must not be null");
         }
         if (pooledObject == null) {
             throw new NullPointerException("pooledObject must not be null");
         }
+        this.localSocket = localSocket;
         this.packetHandler = packetHandler;
         this.pooledObject = pooledObject;
     }
@@ -42,7 +48,7 @@ public PooledPacketWrap(PacketHandler<DatagramPacket> packetHandler, PooledObjec
     public void run() {
         final DatagramPacket packet = pooledObject.getObject();
         try {
-            packetHandler.receive(packet);
+            packetHandler.receive(localSocket, packet);
         } finally {
             pooledObject.returnObject();
         }
diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/SpanStreamUDPPacketHandlerFactory.java b/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/SpanStreamUDPPacketHandlerFactory.java
index 58f11d111052..aa83ac1bb4d0 100644
--- a/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/SpanStreamUDPPacketHandlerFactory.java
+++ b/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/SpanStreamUDPPacketHandlerFactory.java
@@ -27,6 +27,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.net.DatagramPacket;
+import java.net.DatagramSocket;
 import java.net.SocketAddress;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
@@ -70,7 +71,7 @@ private DispatchPacket() {
         }
 
         @Override
-        public void receive(DatagramPacket packet) {
+        public void receive(DatagramSocket localSocket, DatagramPacket packet) {
             final HeaderTBaseDeserializer deserializer = deserializerFactory.createDeserializer();
 
             ByteBuffer requestBuffer = ByteBuffer.wrap(packet.getData());
@@ -104,7 +105,7 @@ public void receive(DatagramPacket packet) {
                     }
                     
                     if (tbaseList.size() == 1) {
-                        if (filter.filter(tbaseList.get(0), socketAddress) == TBaseFilter.BREAK) {
+                        if (filter.filter(localSocket, tbaseList.get(0), socketAddress) == TBaseFilter.BREAK) {
                             continue;
                         }
                     }
diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/TBaseFilter.java b/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/TBaseFilter.java
index 0eb686911d45..82125213f5f8 100644
--- a/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/TBaseFilter.java
+++ b/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/TBaseFilter.java
@@ -18,6 +18,7 @@
 
 import org.apache.thrift.TBase;
 
+import java.net.DatagramSocket;
 import java.net.SocketAddress;
 
 /**
@@ -27,13 +28,15 @@ public interface TBaseFilter<T> {
     boolean CONTINUE = true;
     boolean BREAK = false;
 
-    boolean filter(TBase<?, ?> tBase, T remoteHostAddress);
+    boolean filter(DatagramSocket localSocket, TBase<?, ?> tBase, T remoteHostAddress);
 
     // TODO fix generic type
     TBaseFilter CONTINUE_FILTER = new TBaseFilter<SocketAddress>() {
+
         @Override
-        public boolean filter(TBase<?, ?> tBase, SocketAddress remoteHostAddress) {
+        public boolean filter(DatagramSocket localSocket, TBase<?, ?> tBase, SocketAddress remoteHostAddress) {
             return CONTINUE;
         }
+
     };
 }
diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/TBaseFilterChain.java b/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/TBaseFilterChain.java
index c0de0d64d90f..a7ebef410210 100644
--- a/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/TBaseFilterChain.java
+++ b/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/TBaseFilterChain.java
@@ -18,6 +18,7 @@
 
 import org.apache.thrift.TBase;
 
+import java.net.DatagramSocket;
 import java.net.SocketAddress;
 import java.util.ArrayList;
 import java.util.List;
@@ -37,12 +38,13 @@ public TBaseFilterChain(List<TBaseFilter<T>> tBaseFilter) {
     }
 
     @Override
-    public boolean filter(TBase<?, ?> tBase, T remoteHostAddress) {
+    public boolean filter(DatagramSocket localSocket, TBase<?, ?> tBase, T remoteHostAddress) {
         for (TBaseFilter tBaseFilter : filterChain) {
-            if (tBaseFilter.filter(tBase, remoteHostAddress) == TBaseFilter.BREAK) {
+            if (tBaseFilter.filter(localSocket, tBase, remoteHostAddress) == TBaseFilter.BREAK) {
                 return BREAK;
             }
         }
         return TBaseFilter.CONTINUE;
     }
+
 }
diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/UDPReceiver.java b/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/UDPReceiver.java
index f635613a56cd..e428c9b43453 100644
--- a/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/UDPReceiver.java
+++ b/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/udp/UDPReceiver.java
@@ -16,6 +16,25 @@
 
 package com.navercorp.pinpoint.collector.receiver.udp;
 
+import com.codahale.metrics.MetricRegistry;
+import com.navercorp.pinpoint.collector.monitor.MonitoredExecutorService;
+import com.navercorp.pinpoint.collector.receiver.DataReceiver;
+import com.navercorp.pinpoint.collector.receiver.WorkerOption;
+import com.navercorp.pinpoint.collector.util.DatagramPacketFactory;
+import com.navercorp.pinpoint.collector.util.DefaultObjectPool;
+import com.navercorp.pinpoint.collector.util.ObjectPool;
+import com.navercorp.pinpoint.collector.util.PacketUtils;
+import com.navercorp.pinpoint.collector.util.PooledObject;
+import com.navercorp.pinpoint.common.util.ExecutorFactory;
+import com.navercorp.pinpoint.common.util.PinpointThreadFactory;
+import com.navercorp.pinpoint.rpc.util.CpuUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.util.Assert;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
 import java.io.IOException;
 import java.net.DatagramPacket;
 import java.net.DatagramSocket;
@@ -31,27 +50,6 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import javax.annotation.PostConstruct;
-import javax.annotation.PreDestroy;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.util.Assert;
-
-import com.codahale.metrics.MetricRegistry;
-import com.navercorp.pinpoint.collector.monitor.MonitoredExecutorService;
-import com.navercorp.pinpoint.collector.receiver.DataReceiver;
-import com.navercorp.pinpoint.collector.receiver.WorkerOption;
-import com.navercorp.pinpoint.collector.util.DatagramPacketFactory;
-import com.navercorp.pinpoint.collector.util.DefaultObjectPool;
-import com.navercorp.pinpoint.collector.util.ObjectPool;
-import com.navercorp.pinpoint.collector.util.PacketUtils;
-import com.navercorp.pinpoint.collector.util.PooledObject;
-import com.navercorp.pinpoint.common.util.ExecutorFactory;
-import com.navercorp.pinpoint.common.util.PinpointThreadFactory;
-import com.navercorp.pinpoint.rpc.util.CpuUtils;
-
 /**
  * @author emeroad
  * @author netspider
@@ -183,7 +181,7 @@ private Runnable wrapDispatchTask(final PooledObject<DatagramPacket> pooledPacke
             @Override
             public void run() {
                 PacketHandler<DatagramPacket> dispatchPacket = packetHandlerFactory.createPacketHandler();
-                PooledPacketWrap pooledPacketWrap = new PooledPacketWrap(dispatchPacket, pooledPacket);
+                PooledPacketWrap pooledPacketWrap = new PooledPacketWrap(socket, dispatchPacket, pooledPacket);
                 Runnable execution = pooledPacketWrap;
                 execution.run();
             }


Test patch:
diff --git a/collector/src/test/java/com/navercorp/pinpoint/collector/receiver/udp/NetworkAvailabilityCheckPacketFilterTest.java b/collector/src/test/java/com/navercorp/pinpoint/collector/receiver/udp/NetworkAvailabilityCheckPacketFilterTest.java
index 9ad0e614eed4..c25578201734 100644
--- a/collector/src/test/java/com/navercorp/pinpoint/collector/receiver/udp/NetworkAvailabilityCheckPacketFilterTest.java
+++ b/collector/src/test/java/com/navercorp/pinpoint/collector/receiver/udp/NetworkAvailabilityCheckPacketFilterTest.java
@@ -38,33 +38,42 @@ public class NetworkAvailabilityCheckPacketFilterTest {
     private final Logger logger = LoggerFactory.getLogger(this.getClass());
 
     private NetworkAvailabilityCheckPacketFilter filter;
-    private DatagramSocket datagramSocket;
+    private DatagramSocket senderSocket;
+    private DatagramSocket receiverSocket;
 
     @Before
     public void setUp() throws Exception {
         filter = new NetworkAvailabilityCheckPacketFilter();
-        datagramSocket = new DatagramSocket(0);
+        senderSocket = new DatagramSocket(0);
+        receiverSocket = new DatagramSocket(0);
     }
 
     @After
     public void tearDown() throws Exception {
         filter.destroy();
-        datagramSocket.close();
+        try {
+            senderSocket.close();
+        } catch (Exception e) {
+        }
+        try {
+            receiverSocket.close();
+        } catch (Exception e) {
+        }
     }
 
     @Test
     public void testFilter() throws Exception {
 
-        SocketAddress localSocketAddress = datagramSocket.getLocalSocketAddress();
+        SocketAddress localSocketAddress = senderSocket.getLocalSocketAddress();
         logger.debug("localSocket:{}", localSocketAddress);
 
         NetworkAvailabilityCheckPacket  packet = new NetworkAvailabilityCheckPacket();
-        boolean skipResult = filter.filter(packet, new InetSocketAddress("localhost", datagramSocket.getLocalPort()));
+        boolean skipResult = filter.filter(receiverSocket, packet, new InetSocketAddress("localhost", senderSocket.getLocalPort()));
 
         Assert.assertEquals(skipResult, TBaseFilter.BREAK);
 
         DatagramPacket receivePacket = new DatagramPacket(new byte[100], 100);
-        datagramSocket.receive(receivePacket);
+        senderSocket.receive(receivePacket);
 
         Assert.assertEquals(receivePacket.getLength(), NetworkAvailabilityCheckPacket.DATA_OK.length);
         Assert.assertArrayEquals(Arrays.copyOf(receivePacket.getData(), NetworkAvailabilityCheckPacket.DATA_OK.length), NetworkAvailabilityCheckPacket.DATA_OK);
@@ -74,11 +83,11 @@ public void testFilter() throws Exception {
     @Test
     public void testFilter_Continue() throws Exception {
 
-        SocketAddress localSocketAddress = datagramSocket.getLocalSocketAddress();
+        SocketAddress localSocketAddress = senderSocket.getLocalSocketAddress();
         logger.debug("localSocket:{}", localSocketAddress);
 
         TSpan skip = new TSpan();
-        boolean skipResult = filter.filter(skip, null);
+        boolean skipResult = filter.filter(receiverSocket, skip, null);
 
         Assert.assertEquals(skipResult, TBaseFilter.CONTINUE);
 
diff --git a/collector/src/test/java/com/navercorp/pinpoint/collector/receiver/udp/SpanStreamUDPSenderTest.java b/collector/src/test/java/com/navercorp/pinpoint/collector/receiver/udp/SpanStreamUDPSenderTest.java
index 4691a5baab28..c2464836c021 100644
--- a/collector/src/test/java/com/navercorp/pinpoint/collector/receiver/udp/SpanStreamUDPSenderTest.java
+++ b/collector/src/test/java/com/navercorp/pinpoint/collector/receiver/udp/SpanStreamUDPSenderTest.java
@@ -44,6 +44,7 @@
 import org.springframework.util.SocketUtils;
 
 import java.io.IOException;
+import java.net.DatagramSocket;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -234,7 +235,7 @@ static class TestTBaseFilter<T> implements TBaseFilter<T> {
         private final Logger logger = LoggerFactory.getLogger(this.getClass());
 
         @Override
-        public boolean filter(TBase<?, ?> tBase, T remoteHostAddress) {
+        public boolean filter(DatagramSocket localSocket, TBase<?, ?> tBase, T remoteHostAddress) {
             logger.debug("filter");
             return false;
         }
diff --git a/collector/src/test/java/com/navercorp/pinpoint/collector/receiver/udp/TestUDPReceiver.java b/collector/src/test/java/com/navercorp/pinpoint/collector/receiver/udp/TestUDPReceiver.java
index 18f97e7f2ed5..f7b76080dc18 100644
--- a/collector/src/test/java/com/navercorp/pinpoint/collector/receiver/udp/TestUDPReceiver.java
+++ b/collector/src/test/java/com/navercorp/pinpoint/collector/receiver/udp/TestUDPReceiver.java
@@ -16,6 +16,21 @@
 
 package com.navercorp.pinpoint.collector.receiver.udp;
 
+import com.navercorp.pinpoint.collector.receiver.DataReceiver;
+import com.navercorp.pinpoint.collector.util.DatagramPacketFactory;
+import com.navercorp.pinpoint.collector.util.DefaultObjectPool;
+import com.navercorp.pinpoint.collector.util.ObjectPool;
+import com.navercorp.pinpoint.collector.util.PacketUtils;
+import com.navercorp.pinpoint.collector.util.PooledObject;
+import com.navercorp.pinpoint.common.util.ExecutorFactory;
+import com.navercorp.pinpoint.common.util.PinpointThreadFactory;
+import com.navercorp.pinpoint.rpc.util.CpuUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.util.Assert;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
 import java.io.IOException;
 import java.net.DatagramPacket;
 import java.net.DatagramSocket;
@@ -31,23 +46,6 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import javax.annotation.PostConstruct;
-import javax.annotation.PreDestroy;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.util.Assert;
-
-import com.navercorp.pinpoint.collector.receiver.DataReceiver;
-import com.navercorp.pinpoint.collector.util.DatagramPacketFactory;
-import com.navercorp.pinpoint.collector.util.DefaultObjectPool;
-import com.navercorp.pinpoint.collector.util.ObjectPool;
-import com.navercorp.pinpoint.collector.util.PacketUtils;
-import com.navercorp.pinpoint.collector.util.PooledObject;
-import com.navercorp.pinpoint.common.util.ExecutorFactory;
-import com.navercorp.pinpoint.common.util.PinpointThreadFactory;
-import com.navercorp.pinpoint.rpc.util.CpuUtils;
-
 public class TestUDPReceiver implements DataReceiver {
 
     private final Logger logger;
@@ -155,7 +153,7 @@ private Runnable wrapDispatchTask(final PooledObject<DatagramPacket> pooledPacke
             @Override
             public void run() {
                 PacketHandler<DatagramPacket> dispatchPacket = packetHandlerFactory.createPacketHandler();
-                PooledPacketWrap pooledPacketWrap = new PooledPacketWrap(dispatchPacket, pooledPacket);
+                PooledPacketWrap pooledPacketWrap = new PooledPacketWrap(socket, dispatchPacket, pooledPacket);
                 Runnable execution = pooledPacketWrap;
                 execution.run();
             }