Repository: pinpoint-apm/pinpoint
Issue: Allows  to set queueSize in the TcpDataSender
Labels: bug
module:agent

#### description 
Allows  to set queueSize on the TcpDataSender.

related issue #6144 


Issue comments:
resolved via #6153  #6158 

Pull request review comments:

Patch that resolved the issue:
diff --git a/profiler/src/main/java/com/navercorp/pinpoint/profiler/context/provider/SpanDataSenderProvider.java b/profiler/src/main/java/com/navercorp/pinpoint/profiler/context/provider/SpanDataSenderProvider.java
index e83689d3ef00..002e8f471731 100644
--- a/profiler/src/main/java/com/navercorp/pinpoint/profiler/context/provider/SpanDataSenderProvider.java
+++ b/profiler/src/main/java/com/navercorp/pinpoint/profiler/context/provider/SpanDataSenderProvider.java
@@ -78,7 +78,7 @@ public DataSender get() {
 
             PinpointClientFactory pinpointClientFactory = clientFactoryProvider.get();
             MessageSerializer<byte[]> messageSerializer = new ThriftMessageSerializer(messageConverter);
-            return new TcpDataSender("SpanDataSender", ip, port, pinpointClientFactory, messageSerializer);
+            return new TcpDataSender("SpanDataSender", ip, port, pinpointClientFactory, messageSerializer, writeQueueSize);
         } else {
             UdpDataSenderFactory factory = new UdpDataSenderFactory(ip, port, UDP_EXECUTOR_NAME, writeQueueSize, timeout, sendBufferSize, messageConverter);
             return factory.create(ioType);
diff --git a/profiler/src/main/java/com/navercorp/pinpoint/profiler/context/provider/StatDataSenderProvider.java b/profiler/src/main/java/com/navercorp/pinpoint/profiler/context/provider/StatDataSenderProvider.java
index 8b740c87abd7..95f44ff49569 100644
--- a/profiler/src/main/java/com/navercorp/pinpoint/profiler/context/provider/StatDataSenderProvider.java
+++ b/profiler/src/main/java/com/navercorp/pinpoint/profiler/context/provider/StatDataSenderProvider.java
@@ -80,7 +80,7 @@ public DataSender get() {
 
             PinpointClientFactory pinpointClientFactory = clientFactoryProvider.get();
             MessageSerializer<byte[]> messageSerializer = new ThriftMessageSerializer(messageConverter);
-            return new TcpDataSender("StatDataSender", ip, port, pinpointClientFactory, messageSerializer);
+            return new TcpDataSender("StatDataSender", ip, port, pinpointClientFactory, messageSerializer, writeQueueSize);
         } else {
             UdpDataSenderFactory factory = new UdpDataSenderFactory(ip, port, UDP_EXECUTOR_NAME, writeQueueSize, timeout, sendBufferSize, messageConverter);
             return factory.create(ioType);
diff --git a/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/TcpDataSender.java b/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/TcpDataSender.java
index 94c0986ce780..b0e02dc5b45f 100644
--- a/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/TcpDataSender.java
+++ b/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/TcpDataSender.java
@@ -53,6 +53,8 @@
  */
 public class TcpDataSender implements EnhancedDataSender<Object> {
 
+    private static final int DEFAULT_QUEUE_SIZE = 1024 * 5;
+
     private final Logger logger;
 
     static {
@@ -75,7 +77,7 @@ public class TcpDataSender implements EnhancedDataSender<Object> {
 
 
     public TcpDataSender(String name, String host, int port, PinpointClientFactory clientFactory) {
-        this(name, ClientFactoryUtils.newPinpointClientProvider(host, port, clientFactory), newDefaultMessageSerializer());
+        this(name, ClientFactoryUtils.newPinpointClientProvider(host, port, clientFactory), newDefaultMessageSerializer(), DEFAULT_QUEUE_SIZE);
     }
 
     private static ThriftMessageSerializer newDefaultMessageSerializer() {
@@ -84,21 +86,27 @@ private static ThriftMessageSerializer newDefaultMessageSerializer() {
     }
 
     public TcpDataSender(String name, String host, int port, PinpointClientFactory clientFactory, MessageSerializer<byte[]> messageSerializer) {
-        this(name, ClientFactoryUtils.newPinpointClientProvider(host, port, clientFactory), messageSerializer);
+        this(name, ClientFactoryUtils.newPinpointClientProvider(host, port, clientFactory), messageSerializer, DEFAULT_QUEUE_SIZE);
+    }
+
+    public TcpDataSender(String name, String host, int port, PinpointClientFactory clientFactory, MessageSerializer<byte[]> messageSerializer, int queueSize) {
+        this(name, ClientFactoryUtils.newPinpointClientProvider(host, port, clientFactory), messageSerializer, queueSize);
     }
 
-    private TcpDataSender(String name, ClientFactoryUtils.PinpointClientProvider clientProvider, MessageSerializer<byte[]> messageSerializer) {
+    private TcpDataSender(String name, ClientFactoryUtils.PinpointClientProvider clientProvider, MessageSerializer<byte[]> messageSerializer, int queueSize) {
         this.logger = newLogger(name);
 
-        Assert.requireNonNull(clientProvider, "clientProvider must not be null");
+        Assert.requireNonNull(clientProvider, "clientProvider");
         this.client = clientProvider.get();
 
-        this.messageSerializer = Assert.requireNonNull(messageSerializer, "messageSerializer must not be null");
+        Assert.isTrue(queueSize > 0, "queueSize must be 'queueSize > 0'");
+
+        this.messageSerializer = Assert.requireNonNull(messageSerializer, "messageSerializer");
         this.timer = createTimer(name);
         this.writeFailFutureListener = new WriteFailFutureListener(logger, "io write fail.", "host", -1);
 
         final String executorName = getExecutorName(name);
-        this.executor = createAsyncQueueingExecutor(1024 * 5, executorName);
+        this.executor = createAsyncQueueingExecutor(queueSize, executorName);
     }
 
     private AsyncQueueingExecutor<Object> createAsyncQueueingExecutor(int queueSize, String executorName) {


Test patch: