Repository: pinpoint-apm/pinpoint
Issue: Need a way to refresh row timestamp for AgentInfo
Labels: enhancement

Data in AgentInfo table lives for the duration of TTL. If an agent stays alive for longer than the TTL value without a restart, it gets deleted.

There needs to be a way to refresh (or append) AgentInfo to stop this from happening for agents that are alive.


Issue comments:

Pull request review comments:

Patch that resolved the issue:
diff --git a/profiler/src/main/java/com/navercorp/pinpoint/profiler/AgentInfoSender.java b/profiler/src/main/java/com/navercorp/pinpoint/profiler/AgentInfoSender.java
index 0996060cd3db..75447968e480 100644
--- a/profiler/src/main/java/com/navercorp/pinpoint/profiler/AgentInfoSender.java
+++ b/profiler/src/main/java/com/navercorp/pinpoint/profiler/AgentInfoSender.java
@@ -18,13 +18,13 @@
 
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Map.Entry;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -39,56 +39,70 @@
 import com.navercorp.pinpoint.thrift.dto.TServerMetaData;
 import com.navercorp.pinpoint.thrift.dto.TServiceInfo;
 
-
 /**
  * @author emeroad
  * @author koo.taejin
- * @author hyungil.jeong
+ * @author HyunGil Jeong
  */
 public class AgentInfoSender implements ServerMetaDataListener {
-    private static final Logger LOGGER = LoggerFactory.getLogger(AgentInfoSender.class);
+    // refresh daily
+    public static final long DEFAULT_AGENT_INFO_REFRESH_INTERVAL_MS = 24 * 60 * 60 * 1000L;
+    // retry every 3 seconds
+    public static final long DEFAULT_AGENT_INFO_SEND_INTERVAL_MS = 3 * 1000L;
+    // retry 3 times per attempt
+    public static final int DEFAULT_MAX_TRY_COUNT_PER_ATTEMPT = 3;
 
+    private static final Logger LOGGER = LoggerFactory.getLogger(AgentInfoSender.class);
     private static final ThreadFactory THREAD_FACTORY = new PinpointThreadFactory("Pinpoint-agentInfo-sender", true);
-    
-    private final ScheduledExecutorService EXECUTOR_SERVICE = Executors.newSingleThreadScheduledExecutor(THREAD_FACTORY);
-    private final long agentInfoSendIntervalMs; 
+
+    private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(THREAD_FACTORY);
+
     private final EnhancedDataSender dataSender;
+    private final long refreshIntervalMs;
+    private final long sendIntervalMs;
+    private final int maxTryPerAttempt;
     private final AgentInformation agentInformation;
 
-    public AgentInfoSender(EnhancedDataSender dataSender, long agentInfoSendIntervalMs, AgentInformation agentInformation) {
-        if (dataSender == null) {
-            throw new NullPointerException("dataSender must not be null");
-        }
-        if (agentInformation == null) {
-            throw new NullPointerException("agentInformation must not be null");
-        }
-        this.agentInfoSendIntervalMs = agentInfoSendIntervalMs;
-        this.dataSender = dataSender;
-        this.agentInformation = agentInformation;
-    }
+    private volatile ServerMetaData serverMetaData;
 
-    public void start() {
-        final TAgentInfo agentInfo = createTAgentInfo();
-        if (LOGGER.isInfoEnabled()) {
-            LOGGER.info("AgentInfoSender started. Sending startup information to Pinpoint server via {}. agentInfo={}", dataSender.getClass().getSimpleName(), agentInfo);
-        }
-        send(agentInfo);
+    private AgentInfoSender(Builder builder) {
+        this.dataSender = builder.dataSender;
+        this.agentInformation = builder.agentInformation;
+        this.refreshIntervalMs = builder.refreshIntervalMs;
+        this.sendIntervalMs = builder.sendIntervalMs;
+        this.maxTryPerAttempt = builder.maxTryPerAttempt;
     }
-    
+
     @Override
     public void publishServerMetaData(ServerMetaData serverMetaData) {
-        final TAgentInfo agentInfo = createTAgentInfo(serverMetaData);
-        if (LOGGER.isInfoEnabled()) {
-            LOGGER.info("Sending AgentInfo with ServerMetaData. {}", agentInfo);
+        this.serverMetaData = serverMetaData;
+        submit(this.maxTryPerAttempt);
+    }
+
+    public void start() {
+        submit(Integer.MAX_VALUE);
+        this.executor.scheduleWithFixedDelay(new Runnable() {
+            @Override
+            public void run() {
+                submit(maxTryPerAttempt);
+            }
+        }, this.refreshIntervalMs, this.refreshIntervalMs, TimeUnit.MILLISECONDS);
+    }
+
+    public void stop() {
+        this.executor.shutdown();
+        try {
+            this.executor.awaitTermination(3000, TimeUnit.MILLISECONDS);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
         }
-        send(agentInfo);
+        LOGGER.info("AgentInfoSender stopped");
     }
-    
-    private void send(final TAgentInfo agentInfo) {
-        final AgentInfoSendRunnable agentInfoSendJob = new AgentInfoSendRunnable(agentInfo);
-        new AgentInfoSendRunnableWrapper(agentInfoSendJob).repeatWithFixedDelay(EXECUTOR_SERVICE, 0, this.agentInfoSendIntervalMs, TimeUnit.MILLISECONDS);
+
+    private void submit(final int maxTryPerAttempt) {
+        new AgentInfoSendRunnableWrapper(new AgentInfoSendRunnable(), maxTryPerAttempt).repeatWithFixedDelay(this.executor, 0, this.sendIntervalMs, TimeUnit.MILLISECONDS);
     }
-    
+
     private TAgentInfo createTAgentInfo() {
         final TAgentInfo agentInfo = new TAgentInfo();
         agentInfo.setIp(this.agentInformation.getHostIp());
@@ -101,31 +115,13 @@ private TAgentInfo createTAgentInfo() {
         agentInfo.setServiceType(this.agentInformation.getServerType().getCode());
         agentInfo.setVmVersion(this.agentInformation.getJvmVersion());
         agentInfo.setAgentVersion(Version.VERSION);
-        return agentInfo;
-    }
-    
-    private TAgentInfo createTAgentInfo(final ServerMetaData serverMetaData) {
-        final StringBuilder ports = new StringBuilder();
-        for (Entry<Integer, String> entry : serverMetaData.getConnectors().entrySet()) {
-            ports.append(" ");
-            ports.append(entry.getKey());
+        if (this.serverMetaData != null) {
+            agentInfo.setServerMetaData(createTServiceInfo());
         }
-        final TAgentInfo agentInfo = new TAgentInfo();
-        agentInfo.setIp(this.agentInformation.getHostIp());
-        agentInfo.setHostname(this.agentInformation.getMachineName());
-        agentInfo.setPorts(ports.toString());
-        agentInfo.setAgentId(this.agentInformation.getAgentId());
-        agentInfo.setApplicationName(this.agentInformation.getApplicationName());
-        agentInfo.setPid(this.agentInformation.getPid());
-        agentInfo.setStartTimestamp(this.agentInformation.getStartTime());
-        agentInfo.setServiceType(this.agentInformation.getServerType().getCode());
-        agentInfo.setVmVersion(this.agentInformation.getJvmVersion());
-        agentInfo.setAgentVersion(Version.VERSION);
-        agentInfo.setServerMetaData(createTServiceInfo(serverMetaData));
         return agentInfo;
     }
-    
-    private TServerMetaData createTServiceInfo(final ServerMetaData serverMetaData) {
+
+    private TServerMetaData createTServiceInfo() {
         TServerMetaData tServerMetaData = new TServerMetaData();
         tServerMetaData.setServerInfo(serverMetaData.getServerInfo());
         tServerMetaData.setVmArgs(serverMetaData.getVmArgs());
@@ -139,59 +135,106 @@ private TServerMetaData createTServiceInfo(final ServerMetaData serverMetaData)
         tServerMetaData.setServiceInfos(tServiceInfos);
         return tServerMetaData;
     }
-    
+
     private static class AgentInfoSendRunnableWrapper implements Runnable {
         private final AgentInfoSendRunnable delegate;
-        private ScheduledFuture<?> self;
-        
-        private AgentInfoSendRunnableWrapper(AgentInfoSendRunnable agentInfoSendRunnable) {
+        private final int maxTryCount;
+        private final AtomicInteger tryCount = new AtomicInteger();;
+        private volatile ScheduledFuture<?> self;
+
+        private AgentInfoSendRunnableWrapper(AgentInfoSendRunnable agentInfoSendRunnable, int maxTryCount) {
+            if (agentInfoSendRunnable == null) {
+                throw new NullPointerException("agentInfoSendRunnable must not be null");
+            }
+            if (maxTryCount < 0) {
+                throw new IllegalArgumentException("maxTryCount must not be less than 0");
+            }
             this.delegate = agentInfoSendRunnable;
+            this.maxTryCount = maxTryCount;
         }
 
         @Override
         public void run() {
-            // Cancel self when delegated runnable is completed successfully.
-            if (this.delegate.isSuccessful()) {
-                this.self.cancel(true);
+            // Cancel self when delegated runnable is completed successfully, or when max try count has been reached
+            if (this.delegate.isSuccessful() || this.tryCount.getAndIncrement() == this.maxTryCount) {
+                this.self.cancel(false);
             } else {
                 this.delegate.run();
             }
         }
-        
+
         private void repeatWithFixedDelay(ScheduledExecutorService scheduledExecutorService, long initialDelay, long delay, TimeUnit unit) {
             this.self = scheduledExecutorService.scheduleWithFixedDelay(this, initialDelay, delay, unit);
         }
     }
-    
+
     private class AgentInfoSendRunnable implements Runnable {
         private final AtomicBoolean isSuccessful = new AtomicBoolean(false);
         private final AgentInfoSenderListener agentInfoSenderListener = new AgentInfoSenderListener(this.isSuccessful);
         private final TAgentInfo agentInfo;
-        
-        private AgentInfoSendRunnable(TAgentInfo agentInfo) {
-            this.agentInfo = agentInfo;
+
+        private AgentInfoSendRunnable() {
+            this.agentInfo = createTAgentInfo();
         }
-        
+
         @Override
         public void run() {
             if (!isSuccessful.get()) {
-                dataSender.request(agentInfo, this.agentInfoSenderListener);
+                LOGGER.info("Sending AgentInfo {}", agentInfo);
+                dataSender.request(this.agentInfo, this.agentInfoSenderListener);
             }
         }
-        
+
         public boolean isSuccessful() {
             return this.isSuccessful.get();
         }
     }
 
-    public void stop() {
-        EXECUTOR_SERVICE.shutdown();
-        try {
-            EXECUTOR_SERVICE.awaitTermination(3000, TimeUnit.MILLISECONDS);
-        } catch (InterruptedException e) {
-            Thread.currentThread().interrupt();
+    public static class Builder {
+        private final EnhancedDataSender dataSender;
+        private final AgentInformation agentInformation;
+        private long refreshIntervalMs = DEFAULT_AGENT_INFO_REFRESH_INTERVAL_MS;
+        private long sendIntervalMs = DEFAULT_AGENT_INFO_SEND_INTERVAL_MS;
+        private int maxTryPerAttempt = DEFAULT_MAX_TRY_COUNT_PER_ATTEMPT;
+
+        public Builder(EnhancedDataSender dataSender, AgentInformation agentInformation) {
+            if (dataSender == null) {
+                throw new NullPointerException("enhancedDataSender must not be null");
+            }
+            if (agentInformation == null) {
+                throw new NullPointerException("agentInformation must not be null");
+            }
+            this.dataSender = dataSender;
+            this.agentInformation = agentInformation;
+        }
+
+        public Builder refreshInterval(long refreshIntervalMs) {
+            this.refreshIntervalMs = refreshIntervalMs;
+            return this;
+        }
+
+        public Builder sendInterval(long sendIntervalMs) {
+            this.sendIntervalMs = sendIntervalMs;
+            return this;
+        }
+
+        public Builder maxTryPerAttempt(int maxTryCountPerAttempt) {
+            this.maxTryPerAttempt = maxTryCountPerAttempt;
+            return this;
+        }
+
+        public AgentInfoSender build() {
+            if (this.refreshIntervalMs <= 0) {
+                throw new IllegalStateException("agentInfoRefreshIntervalMs must be greater than 0");
+            }
+            if (this.sendIntervalMs <= 0) {
+                throw new IllegalStateException("agentInfoSendIntervalMs must be greater than 0");
+            }
+            if (this.maxTryPerAttempt <= 0) {
+                throw new IllegalStateException("maxTryPerAttempt must be greater than 0");
+            }
+            return new AgentInfoSender(this);
         }
-        LOGGER.info("AgentInfoSender stopped");
     }
-    
+
 }
diff --git a/profiler/src/main/java/com/navercorp/pinpoint/profiler/DefaultAgent.java b/profiler/src/main/java/com/navercorp/pinpoint/profiler/DefaultAgent.java
index 25106c8a2545..5f3b013244ab 100644
--- a/profiler/src/main/java/com/navercorp/pinpoint/profiler/DefaultAgent.java
+++ b/profiler/src/main/java/com/navercorp/pinpoint/profiler/DefaultAgent.java
@@ -191,7 +191,7 @@ public DefaultAgent(AgentOption agentOption, final InterceptorRegistryBinder int
 
         addCommandService(commandDispatcher, traceContext);
 
-        this.agentInfoSender = new AgentInfoSender(tcpDataSender, profilerConfig.getAgentInfoSendRetryInterval(), this.agentInformation);
+        this.agentInfoSender = new AgentInfoSender.Builder(tcpDataSender, this.agentInformation).sendInterval(profilerConfig.getAgentInfoSendRetryInterval()).build();
         this.serverMetaDataHolder.addListener(this.agentInfoSender);
 
         AgentStatCollectorFactory agentStatCollectorFactory = new AgentStatCollectorFactory(this.getTransactionCounter(this.traceContext));


Test patch:
diff --git a/profiler/src/test/java/com/navercorp/pinpoint/profiler/AgentInfoSenderTest.java b/profiler/src/test/java/com/navercorp/pinpoint/profiler/AgentInfoSenderTest.java
index 6bd0e3c34073..453b7cb17d0b 100644
--- a/profiler/src/test/java/com/navercorp/pinpoint/profiler/AgentInfoSenderTest.java
+++ b/profiler/src/test/java/com/navercorp/pinpoint/profiler/AgentInfoSenderTest.java
@@ -29,6 +29,7 @@
 import com.navercorp.pinpoint.rpc.PinpointSocket;
 import com.navercorp.pinpoint.rpc.client.PinpointClient;
 import com.navercorp.pinpoint.rpc.client.PinpointClientFactory;
+import com.navercorp.pinpoint.rpc.client.PinpointClientReconnectEventListener;
 import com.navercorp.pinpoint.rpc.packet.*;
 import com.navercorp.pinpoint.rpc.server.PinpointServer;
 import com.navercorp.pinpoint.rpc.server.PinpointServerAcceptor;
@@ -37,6 +38,7 @@
 import com.navercorp.pinpoint.thrift.dto.TResult;
 import com.navercorp.pinpoint.thrift.io.HeaderTBaseSerializer;
 import com.navercorp.pinpoint.thrift.io.HeaderTBaseSerializerFactory;
+
 import org.apache.thrift.TException;
 import org.junit.Test;
 import org.slf4j.Logger;
@@ -60,7 +62,7 @@ public class AgentInfoSenderTest {
     public void agentInfoShouldBeSent() throws InterruptedException {
         final AtomicInteger requestCount = new AtomicInteger();
         final AtomicInteger successCount = new AtomicInteger();
-        final long agentInfoSendRetryIntervalMs = 1000L;
+        final long agentInfoSendRetryIntervalMs = 100L;
 
         ResponseServerMessageListener serverListener = new ResponseServerMessageListener(requestCount, successCount);
 
@@ -69,12 +71,12 @@ public void agentInfoShouldBeSent() throws InterruptedException {
         PinpointClientFactory clientFactory = createPinpointClientFactory();
         PinpointClient pinpointClient = ClientFactoryUtils.createPinpointClient(HOST, PORT, clientFactory);
 
-        TcpDataSender sender = new TcpDataSender(pinpointClient);
-        AgentInfoSender agentInfoSender = new AgentInfoSender(sender, agentInfoSendRetryIntervalMs, getAgentInfo());
+        TcpDataSender dataSender = new TcpDataSender(pinpointClient);
+        AgentInfoSender agentInfoSender = new AgentInfoSender.Builder(dataSender, getAgentInfo()).sendInterval(agentInfoSendRetryIntervalMs).build();
 
         try {
             agentInfoSender.start();
-            Thread.sleep(10000L);
+            Thread.sleep(1000L);
         } finally {
             closeAll(serverAcceptor, agentInfoSender, pinpointClient, clientFactory);
         }
@@ -86,8 +88,8 @@ public void agentInfoShouldBeSent() throws InterruptedException {
     public void agentInfoShouldRetryUntilSuccess() throws InterruptedException {
         final AtomicInteger requestCount = new AtomicInteger();
         final AtomicInteger successCount = new AtomicInteger();
-        final long agentInfoSendRetryIntervalMs = 1000L;
-        final int expectedTriesUntilSuccess = 5;
+        final long agentInfoSendRetryIntervalMs = 100L;
+        final int expectedTriesUntilSuccess = AgentInfoSender.DEFAULT_MAX_TRY_COUNT_PER_ATTEMPT;
 
         ResponseServerMessageListener serverListener = new ResponseServerMessageListener(requestCount, successCount, expectedTriesUntilSuccess);
 
@@ -97,7 +99,7 @@ public void agentInfoShouldRetryUntilSuccess() throws InterruptedException {
         PinpointClient pinpointClient = ClientFactoryUtils.createPinpointClient(HOST, PORT, socketFactory);
 
         TcpDataSender dataSender = new TcpDataSender(pinpointClient);
-        AgentInfoSender agentInfoSender = new AgentInfoSender(dataSender, agentInfoSendRetryIntervalMs, getAgentInfo());
+        AgentInfoSender agentInfoSender = new AgentInfoSender.Builder(dataSender, getAgentInfo()).sendInterval(agentInfoSendRetryIntervalMs).build();
 
         try {
             agentInfoSender.start();
@@ -109,11 +111,83 @@ public void agentInfoShouldRetryUntilSuccess() throws InterruptedException {
         assertEquals(1, successCount.get());
     }
 
+    @Test
+    public void agentInfoShouldInitiallyRetryIndefinitely() throws InterruptedException {
+        final AtomicInteger requestCount = new AtomicInteger();
+        final AtomicInteger successCount = new AtomicInteger();
+        final long agentInfoSendRetryIntervalMs = 100L;
+        final int expectedTriesUntilSuccess = 15;
+
+        ResponseServerMessageListener serverListener = new ResponseServerMessageListener(requestCount, successCount, expectedTriesUntilSuccess);
+
+        PinpointServerAcceptor serverAcceptor = createServerAcceptor(serverListener);
+
+        PinpointClientFactory socketFactory = createPinpointClientFactory();
+        PinpointClient pinpointClient = ClientFactoryUtils.createPinpointClient(HOST, PORT, socketFactory);
+
+        TcpDataSender dataSender = new TcpDataSender(pinpointClient);
+        AgentInfoSender agentInfoSender = new AgentInfoSender.Builder(dataSender, getAgentInfo()).sendInterval(agentInfoSendRetryIntervalMs).build();
+
+        try {
+            agentInfoSender.start();
+            Thread.sleep(agentInfoSendRetryIntervalMs * expectedTriesUntilSuccess);
+        } finally {
+            closeAll(serverAcceptor, agentInfoSender, pinpointClient, socketFactory);
+        }
+        assertEquals(expectedTriesUntilSuccess, requestCount.get());
+        assertEquals(1, successCount.get());
+    }
+    
+    @Test
+    public void agentInfoShouldRetryUntilAttemptsAreExhaustedWhenRefreshing() throws InterruptedException {
+        final AtomicInteger requestCount = new AtomicInteger();
+        final AtomicInteger successCount = new AtomicInteger();
+        final long agentInfoSendRetryIntervalMs = 100L;
+        final long agentInfoSendRefreshIntervalMs = 1000L;
+        final int expectedTries = AgentInfoSender.DEFAULT_MAX_TRY_COUNT_PER_ATTEMPT + 1;
+        final CountDownLatch agentReconnectLatch = new CountDownLatch(1);
+
+        ResponseServerMessageListener successServerListener = new ResponseServerMessageListener(requestCount, successCount);
+        ResponseServerMessageListener failServerListener = new ResponseServerMessageListener(requestCount, successCount, Integer.MAX_VALUE);
+
+        PinpointServerAcceptor successServerAcceptor = createServerAcceptor(successServerListener);
+        PinpointServerAcceptor failServerAcceptor = null;
+
+        PinpointClientFactory socketFactory = createPinpointClientFactory();
+        PinpointClient pinpointClient = ClientFactoryUtils.createPinpointClient(HOST, PORT, socketFactory);
+
+        TcpDataSender dataSender = new TcpDataSender(pinpointClient);
+        dataSender.addReconnectEventListener(new PinpointClientReconnectEventListener() {
+            @Override
+            public void reconnectPerformed(PinpointClient client) {
+                agentReconnectLatch.countDown();
+            }
+        });
+        AgentInfoSender agentInfoSender = new AgentInfoSender.Builder(dataSender, getAgentInfo())
+                .refreshInterval(agentInfoSendRefreshIntervalMs)
+                .sendInterval(agentInfoSendRetryIntervalMs)
+                .build();
+        try {
+            agentInfoSender.start();
+            Thread.sleep(agentInfoSendRetryIntervalMs);
+            successServerAcceptor.close();
+            // wait till agent reconnects
+            failServerAcceptor = createServerAcceptor(failServerListener);
+            agentReconnectLatch.await();
+            Thread.sleep(agentInfoSendRefreshIntervalMs);
+        } finally {
+            closeAll(failServerAcceptor, agentInfoSender, pinpointClient, socketFactory);
+        }
+        assertEquals(1, successCount.get());
+        assertEquals(expectedTries, requestCount.get());
+    }
+
     @Test
     public void agentInfoShouldBeSentOnlyOnceEvenAfterReconnect() throws InterruptedException {
         final AtomicInteger requestCount = new AtomicInteger();
         final AtomicInteger successCount = new AtomicInteger();
-        final long agentInfoSendRetryIntervalMs = 1000L;
+        final long agentInfoSendRetryIntervalMs = 100L;
+        final int maxTryCountPerAttempt = Integer.MAX_VALUE;
 
         ResponseServerMessageListener serverListener = new ResponseServerMessageListener(requestCount, successCount);
 
@@ -121,30 +195,33 @@ public void agentInfoShouldBeSentOnlyOnceEvenAfterReconnect() throws Interrupted
         PinpointClient pinpointClient = ClientFactoryUtils.createPinpointClient(HOST, PORT, clientFactory);
 
         TcpDataSender dataSender = new TcpDataSender(pinpointClient);
-        AgentInfoSender agentInfoSender = new AgentInfoSender(dataSender, agentInfoSendRetryIntervalMs, getAgentInfo());
+        AgentInfoSender agentInfoSender = new AgentInfoSender.Builder(dataSender, getAgentInfo())
+                .sendInterval(agentInfoSendRetryIntervalMs)
+                .maxTryPerAttempt(maxTryCountPerAttempt)
+                .build();
 
         try {
             agentInfoSender.start();
-            createAndDeleteServer(serverListener, 5000L);
-            Thread.sleep(1000L);
-            createAndDeleteServer(serverListener, 5000L);
-            Thread.sleep(1000L);
-            createAndDeleteServer(serverListener, 5000L);
+            createAndDeleteServer(serverListener, 1000L);
+            Thread.sleep(500L);
+            createAndDeleteServer(serverListener, 1000L);
+            Thread.sleep(500L);
+            createAndDeleteServer(serverListener, 1000L);
         } finally {
             closeAll(null, agentInfoSender, pinpointClient, clientFactory);
         }
-        assertEquals(1, requestCount.get());
         assertEquals(1, successCount.get());
     }
 
     @Test
-    public void agentInfoShouldKeepRetrying() throws InterruptedException {
+    public void agentInfoShouldKeepRefreshing() throws InterruptedException {
         final AtomicInteger requestCount = new AtomicInteger();
         final AtomicInteger successCount = new AtomicInteger();
-        final long agentInfoSendRetryIntervalMs = 1000L;
-        final long minimumAgentInfoSendRetryCount = 10;
+        final long agentInfoSendRetryIntervalMs = 100L;
+        final long agentInfoSendRefreshIntervalMs = 1000L;
+        final int expectedRefreshCount = 5;
 
-        ResponseServerMessageListener serverListener = new ResponseServerMessageListener(requestCount, successCount, Integer.MAX_VALUE);
+        ResponseServerMessageListener serverListener = new ResponseServerMessageListener(requestCount, successCount);
 
         PinpointServerAcceptor serverAcceptor = createServerAcceptor(serverListener);
 
@@ -152,18 +229,21 @@ public void agentInfoShouldKeepRetrying() throws InterruptedException {
         PinpointClient pinpointClient = ClientFactoryUtils.createPinpointClient(HOST, PORT, socketFactory);
 
         TcpDataSender dataSender = new TcpDataSender(pinpointClient);
-        AgentInfoSender agentInfoSender = new AgentInfoSender(dataSender, agentInfoSendRetryIntervalMs, getAgentInfo());
+        AgentInfoSender agentInfoSender = new AgentInfoSender.Builder(dataSender, getAgentInfo())
+                .refreshInterval(agentInfoSendRefreshIntervalMs)
+                .sendInterval(agentInfoSendRetryIntervalMs)
+                .build();
 
         try {
             agentInfoSender.start();
-            Thread.sleep(agentInfoSendRetryIntervalMs * minimumAgentInfoSendRetryCount);
+            Thread.sleep(agentInfoSendRefreshIntervalMs * expectedRefreshCount);
         } finally {
             closeAll(serverAcceptor, agentInfoSender, pinpointClient, socketFactory);
         }
-        assertTrue(requestCount.get() >= minimumAgentInfoSendRetryCount);
-        assertEquals(0, successCount.get());
+        assertTrue(requestCount.get() >= expectedRefreshCount);
+        assertTrue(successCount.get() >= expectedRefreshCount);
     }
-    
+
     @Test
     public void serverMetaDataShouldBeSentOnPublish() throws InterruptedException {
         // Given
@@ -178,14 +258,14 @@ public void serverMetaDataShouldBeSentOnPublish() throws InterruptedException {
         PinpointClientFactory clientFactory = createPinpointClientFactory();
         PinpointClient pinpointClient = ClientFactoryUtils.createPinpointClient(HOST, PORT, clientFactory);
 
-        TcpDataSender sender = new TcpDataSender(pinpointClient);
-        AgentInfoSender agentInfoSender = new AgentInfoSender(sender, agentInfoSendRetryIntervalMs, getAgentInfo());
+        TcpDataSender dataSender = new TcpDataSender(pinpointClient);
+        AgentInfoSender agentInfoSender = new AgentInfoSender.Builder(dataSender, getAgentInfo()).sendInterval(agentInfoSendRetryIntervalMs).build();
         final List<ServerMetaData> serverMetaDataObjects = new ArrayList<ServerMetaData>();
-        serverMetaDataObjects.add(new DefaultServerMetaData("server1", Collections.<String>emptyList(), Collections.<Integer, String>emptyMap(), Collections.<ServiceInfo>emptyList()));
-        serverMetaDataObjects.add(new DefaultServerMetaData("server2", Collections.<String>emptyList(), Collections.<Integer, String>emptyMap(), Collections.<ServiceInfo>emptyList()));
-        serverMetaDataObjects.add(new DefaultServerMetaData("server3", Collections.<String>emptyList(), Collections.<Integer, String>emptyMap(), Collections.<ServiceInfo>emptyList()));
-        serverMetaDataObjects.add(new DefaultServerMetaData("server4", Collections.<String>emptyList(), Collections.<Integer, String>emptyMap(), Collections.<ServiceInfo>emptyList()));
-        serverMetaDataObjects.add(new DefaultServerMetaData("server5", Collections.<String>emptyList(), Collections.<Integer, String>emptyMap(), Collections.<ServiceInfo>emptyList()));
+        serverMetaDataObjects.add(new DefaultServerMetaData("server1", Collections.<String> emptyList(), Collections.<Integer, String> emptyMap(), Collections.<ServiceInfo> emptyList()));
+        serverMetaDataObjects.add(new DefaultServerMetaData("server2", Collections.<String> emptyList(), Collections.<Integer, String> emptyMap(), Collections.<ServiceInfo> emptyList()));
+        serverMetaDataObjects.add(new DefaultServerMetaData("server3", Collections.<String> emptyList(), Collections.<Integer, String> emptyMap(), Collections.<ServiceInfo> emptyList()));
+        serverMetaDataObjects.add(new DefaultServerMetaData("server4", Collections.<String> emptyList(), Collections.<Integer, String> emptyMap(), Collections.<ServiceInfo> emptyList()));
+        serverMetaDataObjects.add(new DefaultServerMetaData("server5", Collections.<String> emptyList(), Collections.<Integer, String> emptyMap(), Collections.<ServiceInfo> emptyList()));
         // When
         try {
             for (ServerMetaData serverMetaData : serverMetaDataObjects) {
@@ -199,7 +279,7 @@ public void serverMetaDataShouldBeSentOnPublish() throws InterruptedException {
         assertEquals(5, requestCount.get());
         assertEquals(5, successCount.get());
     }
-    
+
     @Test
     public void serverMetaDataCouldBePublishedFromMultipleThreads() throws InterruptedException {
         // Given
@@ -220,9 +300,9 @@ public void serverMetaDataCouldBePublishedFromMultipleThreads() throws Interrupt
         PinpointClientFactory clientFactory = createPinpointClientFactory();
         PinpointClient pinpointClient = ClientFactoryUtils.createPinpointClient(HOST, PORT, clientFactory);
 
-        TcpDataSender sender = new TcpDataSender(pinpointClient);
-        AgentInfoSender agentInfoSender = new AgentInfoSender(sender, agentInfoSendRetryIntervalMs, getAgentInfo());
-        final ServerMetaDataHolder metaDataContext = new DefaultServerMetaDataHolder(Collections.<String>emptyList());
+        TcpDataSender dataSender = new TcpDataSender(pinpointClient);
+        AgentInfoSender agentInfoSender = new AgentInfoSender.Builder(dataSender, getAgentInfo()).sendInterval(agentInfoSendRetryIntervalMs).build();
+        final ServerMetaDataHolder metaDataContext = new DefaultServerMetaDataHolder(Collections.<String> emptyList());
         metaDataContext.addListener(agentInfoSender);
         // When
         for (int i = 0; i < threadCount; ++i) {
@@ -263,7 +343,8 @@ public void reconnectionStressTest() throws InterruptedException {
         final long stressTestTime = 60 * 1000L;
         final int randomMaxTime = 3000;
         final long agentInfoSendRetryIntervalMs = 1000L;
-        final int expectedTriesUntilSuccess = (int)stressTestTime / (randomMaxTime * 2);
+        final int maxTryPerAttempt = Integer.MAX_VALUE;
+        final int expectedTriesUntilSuccess = (int) stressTestTime / (randomMaxTime * 2);
 
         ResponseServerMessageListener serverListener = new ResponseServerMessageListener(requestCount, successCount, expectedTriesUntilSuccess);
 
@@ -271,7 +352,10 @@ public void reconnectionStressTest() throws InterruptedException {
         PinpointClient pinpointClient = ClientFactoryUtils.createPinpointClient(HOST, PORT, clientFactory);
 
         TcpDataSender dataSender = new TcpDataSender(pinpointClient);
-        AgentInfoSender agentInfoSender = new AgentInfoSender(dataSender, agentInfoSendRetryIntervalMs, getAgentInfo());
+        AgentInfoSender agentInfoSender = new AgentInfoSender.Builder(dataSender, getAgentInfo())
+                .sendInterval(agentInfoSendRetryIntervalMs)
+                .maxTryPerAttempt(maxTryPerAttempt)
+                .build();
 
         long startTime = System.currentTimeMillis();
 
@@ -393,11 +477,11 @@ public void handlePing(PingPacket pingPacket, PinpointServer pinpointServer) {
             logger.info("ping received {} {} ", pingPacket, pinpointServer);
         }
     }
-    
+
     private PinpointClientFactory createPinpointClientFactory() {
         PinpointClientFactory clientFactory = new PinpointClientFactory();
         clientFactory.setTimeoutMillis(1000 * 5);
-        clientFactory.setProperties(Collections.<String, Object>emptyMap());
+        clientFactory.setProperties(Collections.<String, Object> emptyMap());
 
         return clientFactory;
     }