Merge pull request #1718 from koo-taejin/#1711

Added UdpDataSender using NIO #1711
2026-08-27 13:46:45 +10:00 · 2016-04-26 18:53:37 +09:00
parent a6e497cc7d 590694b0e4
commit 7362bee2d4
15 changed files with 647 additions and 18 deletions
@@ -47,12 +47,14 @@ profiler.spandatasender.write.queue.size=5120
 #profiler.spandatasender.socket.sendbuffersize=1048576
 #profiler.spandatasender.socket.timeout=3000
 profiler.spandatasender.chunk.size=16384
+profiler.spandatasender.socket.type=OIO

 # Capacity of the StatDataSender write queue.
 profiler.statdatasender.write.queue.size=5120
 #profiler.statdatasender.socket.sendbuffersize=1048576
 #profiler.statdatasender.socket.timeout=3000
 profiler.statdatasender.chunk.size=16384
+profiler.statdatasender.socket.type=OIO

 # Interval to retry sending agent info. Unit is milliseconds.
 profiler.agentInfo.send.retry.interval=300000
@@ -16,6 +16,11 @@

 package com.navercorp.pinpoint.bootstrap.config;

+import com.navercorp.pinpoint.bootstrap.logging.JavaLoggerFactory;
+import com.navercorp.pinpoint.bootstrap.util.NumberUtils;
+import com.navercorp.pinpoint.bootstrap.util.spring.PropertyPlaceholderHelper;
+import com.navercorp.pinpoint.common.util.PropertyUtils;
+
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.util.Arrays;
@@ -28,11 +33,6 @@ import java.util.logging.Level;
 import java.util.logging.Logger;
 import java.util.regex.Pattern;

-import com.navercorp.pinpoint.bootstrap.logging.JavaLoggerFactory;
-import com.navercorp.pinpoint.bootstrap.util.NumberUtils;
-import com.navercorp.pinpoint.bootstrap.util.spring.PropertyPlaceholderHelper;
-import com.navercorp.pinpoint.common.util.PropertyUtils;
-
 /**
 * @author emeroad
 * @author netspider
@@ -101,12 +101,14 @@ public class DefaultProfilerConfig implements ProfilerConfig {
    private int spanDataSenderSocketSendBufferSize = 1024 * 64 * 16;
    private int spanDataSenderSocketTimeout = 1000 * 3;
    private int spanDataSenderChunkSize = 1024 * 16;
+    private String spanDataSenderSocketType = "OIO";

    private int statDataSenderWriteQueueSize = 1024 * 5;
    private int statDataSenderSocketSendBufferSize = 1024 * 64 * 16;
    private int statDataSenderSocketTimeout = 1000 * 3;
    private int statDataSenderChunkSize = 1024 * 16;
-    
+    private String statDataSenderSocketType = "OIO";
+
    private boolean tcpDataSenderCommandAcceptEnable = false;

    private boolean traceAgentActiveThread = true;
@@ -247,6 +249,11 @@ public class DefaultProfilerConfig implements ProfilerConfig {
        return statDataSenderSocketTimeout;
    }

+    @Override
+    public String getStatDataSenderSocketType() {
+        return statDataSenderSocketType;
+    }
+
    @Override
    public int getSpanDataSenderWriteQueueSize() {
        return spanDataSenderWriteQueueSize;
@@ -272,6 +279,11 @@ public class DefaultProfilerConfig implements ProfilerConfig {
        return spanDataSenderSocketTimeout;
    }

+    @Override
+    public String getSpanDataSenderSocketType() {
+        return spanDataSenderSocketType;
+    }
+
    @Override
    public int getSpanDataSenderChunkSize() {
        return spanDataSenderChunkSize;
@@ -568,11 +580,13 @@ public class DefaultProfilerConfig implements ProfilerConfig {
        this.spanDataSenderSocketSendBufferSize = readInt("profiler.spandatasender.socket.sendbuffersize", 1024 * 64 * 16);
        this.spanDataSenderSocketTimeout = readInt("profiler.spandatasender.socket.timeout", 1000 * 3);
        this.spanDataSenderChunkSize = readInt("profiler.spandatasender.chunk.size", 1024 * 16);
+        this.spanDataSenderSocketType = readString("profiler.spandatasender.socket.type", "OIO");

        this.statDataSenderWriteQueueSize = readInt("profiler.statdatasender.write.queue.size", 1024 * 5);
        this.statDataSenderSocketSendBufferSize = readInt("profiler.statdatasender.socket.sendbuffersize", 1024 * 64 * 16);
        this.statDataSenderSocketTimeout = readInt("profiler.statdatasender.socket.timeout", 1000 * 3);
        this.statDataSenderChunkSize = readInt("profiler.statdatasender.chunk.size", 1024 * 16);
+        this.statDataSenderSocketType = readString("profiler.statdatasender.socket.type", "OIO");

        this.tcpDataSenderCommandAcceptEnable = readBoolean("profiler.tcpdatasender.command.accept.enable", false);

@@ -817,6 +831,8 @@ public class DefaultProfilerConfig implements ProfilerConfig {
        builder.append(spanDataSenderSocketTimeout);
        builder.append(", spanDataSenderChunkSize=");
        builder.append(spanDataSenderChunkSize);
+        builder.append(", spanDataSenderSocketType=");
+        builder.append(spanDataSenderSocketType);
        builder.append(", statDataSenderWriteQueueSize=");
        builder.append(statDataSenderWriteQueueSize);
        builder.append(", statDataSenderSocketSendBufferSize=");
@@ -825,6 +841,8 @@ public class DefaultProfilerConfig implements ProfilerConfig {
        builder.append(statDataSenderSocketTimeout);
        builder.append(", statDataSenderChunkSize=");
        builder.append(statDataSenderChunkSize);
+        builder.append(", statDataSenderSocketType=");
+        builder.append(statDataSenderSocketType);
        builder.append(", tcpDataSenderCommandAcceptEnable=");
        builder.append(tcpDataSenderCommandAcceptEnable);
        builder.append(", traceAgentActiveThread=");
@@ -43,6 +43,8 @@ public interface ProfilerConfig {

    int getStatDataSenderSocketTimeout();

+    String getStatDataSenderSocketType();
+
    int getSpanDataSenderWriteQueueSize();

    int getSpanDataSenderSocketSendBufferSize();
@@ -53,6 +55,8 @@ public interface ProfilerConfig {

    int getSpanDataSenderSocketTimeout();

+    String getSpanDataSenderSocketType();
+
    int getSpanDataSenderChunkSize();

    int getStatDataSenderChunkSize();
@@ -16,15 +16,6 @@

 package com.navercorp.pinpoint.profiler;

-import java.lang.instrument.Instrumentation;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.Set;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import com.navercorp.pinpoint.ProductInfo;
 import com.navercorp.pinpoint.bootstrap.Agent;
 import com.navercorp.pinpoint.bootstrap.AgentOption;
@@ -61,13 +52,21 @@ import com.navercorp.pinpoint.profiler.sampler.SamplerFactory;
 import com.navercorp.pinpoint.profiler.sender.DataSender;
 import com.navercorp.pinpoint.profiler.sender.EnhancedDataSender;
 import com.navercorp.pinpoint.profiler.sender.TcpDataSender;
-import com.navercorp.pinpoint.profiler.sender.UdpDataSender;
+import com.navercorp.pinpoint.profiler.sender.UdpDataSenderFactory;
 import com.navercorp.pinpoint.profiler.util.ApplicationServerTypeResolver;
 import com.navercorp.pinpoint.profiler.util.RuntimeMXBeanUtils;
 import com.navercorp.pinpoint.rpc.ClassPreLoader;
 import com.navercorp.pinpoint.rpc.client.PinpointClient;
 import com.navercorp.pinpoint.rpc.client.PinpointClientFactory;
 import com.navercorp.pinpoint.rpc.util.ClientFactoryUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.instrument.Instrumentation;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;

 /**
 * @author emeroad
@@ -364,11 +363,13 @@ public class DefaultAgent implements Agent {
    }

    protected DataSender createUdpStatDataSender(int port, String threadName, int writeQueueSize, int timeout, int sendBufferSize) {
-        return new UdpDataSender(this.profilerConfig.getCollectorStatServerIp(), port, threadName, writeQueueSize, timeout, sendBufferSize);
+        UdpDataSenderFactory factory = new UdpDataSenderFactory(this.profilerConfig.getCollectorStatServerIp(), port, threadName, writeQueueSize, timeout, sendBufferSize);
+        return factory.create(profilerConfig.getStatDataSenderSocketType());
    }
    
    protected DataSender createUdpSpanDataSender(int port, String threadName, int writeQueueSize, int timeout, int sendBufferSize) {
-        return new UdpDataSender(this.profilerConfig.getCollectorSpanServerIp(), port, threadName, writeQueueSize, timeout, sendBufferSize);
+        UdpDataSenderFactory factory = new UdpDataSenderFactory(this.profilerConfig.getCollectorSpanServerIp(), port, threadName, writeQueueSize, timeout, sendBufferSize);
+        return factory.create(profilerConfig.getSpanDataSenderSocketType());
    }

    protected EnhancedDataSender getTcpDataSender() {
@@ -0,0 +1,191 @@
+package com.navercorp.pinpoint.profiler.sender;
+
+import com.navercorp.pinpoint.rpc.PinpointSocketException;
+import com.navercorp.pinpoint.rpc.buffer.ByteBufferFactory;
+import com.navercorp.pinpoint.rpc.buffer.ByteBufferFactoryLocator;
+import com.navercorp.pinpoint.rpc.buffer.ByteBufferType;
+import com.navercorp.pinpoint.thrift.io.ByteBufferOutputStream;
+import com.navercorp.pinpoint.thrift.io.HeaderTBaseSerializer2;
+import com.navercorp.pinpoint.thrift.io.HeaderTBaseSerializerFactory2;
+import com.navercorp.pinpoint.thrift.io.NetworkAvailabilityCheckPacket;
+import org.apache.thrift.TBase;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.DatagramSocket;
+import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.DatagramChannel;
+import java.util.Arrays;
+
+/**
+ * @Author Taejin Koo
+ */
+public class NioUDPDataSender extends AbstractDataSender implements DataSender {
+
+    protected final Logger logger = LoggerFactory.getLogger(this.getClass());
+    protected final boolean isDebug = logger.isDebugEnabled();
+
+    public static final int SOCKET_TIMEOUT = 1000 * 5;
+    public static final int SEND_BUFFER_SIZE = 1024 * 64 * 16;
+    public static final int UDP_MAX_PACKET_LENGTH = 65507;
+
+    private final DatagramChannel datagramChannel;
+    private final HeaderTBaseSerializer2 serializer;
+    private final ByteBufferOutputStream byteBufferOutputStream;
+
+    private final AsyncQueueingExecutor<Object> executor;
+
+    private volatile boolean closed = false;
+
+    public NioUDPDataSender(String host, int port, String threadName, int queueSize) {
+        this(host, port, threadName, queueSize, SOCKET_TIMEOUT, SEND_BUFFER_SIZE);
+    }
+
+    public NioUDPDataSender(String host, int port, String threadName, int queueSize, int timeout, int sendBufferSize) {
+        if (host == null ) {
+            throw new NullPointerException("host must not be null");
+        }
+        if (threadName == null) {
+            throw new NullPointerException("threadName must not be null");
+        }
+        if (queueSize <= 0) {
+            throw new IllegalArgumentException("queueSize");
+        }
+        if (timeout <= 0) {
+            throw new IllegalArgumentException("timeout");
+        }
+        if (sendBufferSize <= 0) {
+            throw new IllegalArgumentException("sendBufferSize");
+        }
+
+        // TODO If fail to create socket, stop agent start
+        logger.info("NioUDPDataSender initialized. host={}, port={}", host, port);
+        this.datagramChannel = createChannel(host, port, timeout, sendBufferSize);
+
+        HeaderTBaseSerializerFactory2 serializerFactory = new HeaderTBaseSerializerFactory2();
+        this.serializer = serializerFactory.createSerializer();
+
+        ByteBufferFactory bufferFactory = ByteBufferFactoryLocator.getFactory(ByteBufferType.DIRECT);
+        ByteBuffer byteBuffer = bufferFactory.getBuffer(UDP_MAX_PACKET_LENGTH);
+        this.byteBufferOutputStream = new ByteBufferOutputStream(byteBuffer);
+
+        this.executor = createAsyncQueueingExecutor(queueSize, threadName);
+    }
+
+    private DatagramChannel createChannel(String host, int port, int timeout, int sendBufferSize) {
+        DatagramChannel datagramChannel = null;
+        DatagramSocket socket = null;
+        try {
+            datagramChannel = DatagramChannel.open();
+            socket = datagramChannel.socket();
+            socket.setSoTimeout(timeout);
+            socket.setSendBufferSize(sendBufferSize);
+
+            if (logger.isWarnEnabled()) {
+                final int checkSendBufferSize = socket.getSendBufferSize();
+                if (sendBufferSize != checkSendBufferSize) {
+                    logger.warn("DatagramChannel.setSendBufferSize() error. {}!={}", sendBufferSize, checkSendBufferSize);
+                }
+            }
+
+            InetSocketAddress serverAddress = new InetSocketAddress(host, port);
+            datagramChannel.connect(serverAddress);
+
+            return datagramChannel;
+        } catch (IOException e) {
+            if (socket != null) {
+                socket.close();
+            }
+
+            if (datagramChannel != null) {
+                try {
+                    datagramChannel.close();
+                } catch (IOException ignored) {
+                }
+            }
+
+            throw new IllegalStateException("DatagramChannel create fail. Cause" + e.getMessage(), e);
+        }
+    }
+
+    @Override
+    public boolean send(TBase<?, ?> data) {
+        return executor.execute(data);
+    }
+
+    @Override
+    public void stop() {
+        try {
+            closed = true;
+            executor.stop();
+        } finally {
+            try {
+                byteBufferOutputStream.close();
+            } catch (IOException e) {
+                // ignore
+            }
+        }
+    }
+
+    public boolean isNetworkAvailable() {
+        final NetworkAvailabilityCheckPacket dto = new NetworkAvailabilityCheckPacket();
+        try {
+            sendPacket(dto);
+
+            if (logger.isInfoEnabled()) {
+                logger.info("Data sent. {}", dto);
+            }
+
+            final byte[] receiveData = new byte[NetworkAvailabilityCheckPacket.DATA_OK.length];
+            datagramChannel.read(ByteBuffer.wrap(receiveData));
+
+            if (logger.isInfoEnabled()) {
+                logger.info("Data received. {}", Arrays.toString(receiveData));
+            }
+
+            return Arrays.equals(NetworkAvailabilityCheckPacket.DATA_OK , receiveData);
+        } catch (IOException e) {
+            logger.warn("packet send error {}", dto, e);
+            return false;
+        }
+    }
+
+    protected void sendPacket(Object message) {
+        if (closed) {
+            throw new PinpointSocketException("NioUDPDataSender already closed.");
+        }
+
+        if (message instanceof TBase) {
+            byteBufferOutputStream.clear();
+
+            final TBase dto = (TBase) message;
+            // do not copy bytes because it's single threaded
+
+            try {
+                serializer.serialize(dto,  byteBufferOutputStream);
+            } catch (TException e) {
+                throw new PinpointSocketException("Serialize " + dto + " failed. Error:" +  e.getMessage(), e);
+            }
+            ByteBuffer byteBuffer = byteBufferOutputStream.getByteBuffer();
+            int bufferSize = byteBuffer.remaining();
+            try {
+                datagramChannel.write(byteBuffer);
+            } catch (IOException e) {
+                final Thread currentThread = Thread.currentThread();
+                if (currentThread.isInterrupted()) {
+                    logger.warn("{} thread interrupted.", currentThread.getName());
+                    throw new PinpointSocketException(currentThread.getName() + " thread interrupted.", e);
+                } else {
+                    throw new PinpointSocketException("packet send error. size:" + bufferSize + ", " +  dto, e);
+                }
+            }
+        } else {
+            logger.warn("sendPacket fail. invalid type:{}", message != null ? message.getClass() : null);
+            return;
+        }
+    }
+
+}
@@ -0,0 +1,40 @@
+package com.navercorp.pinpoint.profiler.sender;
+
+/**
+ * @Author Taejin Koo
+ */
+public final class UdpDataSenderFactory {
+
+//    String host, int port, String threadName, int queueSize, int timeout, int sendBufferSize
+
+    private final String host;
+    private final int port;
+    private final String threadName;
+    private final int queueSize;
+    private final int timeout;
+    private final int sendBufferSize;
+
+    public UdpDataSenderFactory(String host, int port, String threadName, int queueSize, int timeout, int sendBufferSize) {
+        this.host = host;
+        this.port = port;
+        this.threadName = threadName;
+        this.queueSize = queueSize;
+        this.timeout = timeout;
+        this.sendBufferSize = sendBufferSize;
+    }
+
+    public DataSender create(String typeName) {
+        return create(UdpDataSenderType.valueOf(typeName));
+    }
+
+    public DataSender create(UdpDataSenderType type) {
+        if (type == UdpDataSenderType.NIO) {
+            return new NioUDPDataSender(host, port, threadName, queueSize, timeout, sendBufferSize);
+        } else if (type == UdpDataSenderType.OIO) {
+            return new UdpDataSender(host, port, threadName, queueSize, timeout, sendBufferSize);
+        } else {
+            throw new IllegalArgumentException("Unknown type.");
+        }
+    }
+
+}
@@ -0,0 +1,11 @@
+package com.navercorp.pinpoint.profiler.sender;
+
+/**
+ * @Author Taejin Koo
+ */
+public enum UdpDataSenderType {
+
+    OIO,
+    NIO;
+
+}
@@ -0,0 +1,113 @@
+package com.navercorp.pinpoint.profiler.sender;
+
+import com.navercorp.pinpoint.thrift.dto.TAgentInfo;
+import org.apache.commons.lang3.RandomStringUtils;
+import org.apache.thrift.TBase;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.util.SocketUtils;
+
+import java.io.IOException;
+import java.net.DatagramPacket;
+import java.net.DatagramSocket;
+import java.net.SocketException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * @Author Taejin Koo
+ */
+public class NioUdpDataSenderTest {
+
+    // The correct maximum UDP message size is 65507, as determined by the following formula:
+    // 0xffff - (sizeof(IP Header) + sizeof(UDP Header)) = 65535-(20+8) = 65507
+    private static int AcceptedSize = 65507;
+    private final Logger logger = LoggerFactory.getLogger(this.getClass().getName());
+    // port conflict against base port. so increased 5
+    private int PORT = SocketUtils.findAvailableUdpPort(61112);
+    private DatagramSocket receiver;
+
+    @Before
+    public void setUp() throws SocketException {
+        receiver = new DatagramSocket(PORT);
+        receiver.setSoTimeout(1000);
+    }
+
+    @After
+    public void setDown() throws InterruptedException {
+        close(receiver);
+        // port conflict happens when testcases run continuously so port number is increased.
+        PORT = SocketUtils.findAvailableUdpPort(61112);
+    }
+
+    private void close(DatagramSocket socket) {
+        if (socket == null) {
+            return;
+        }
+        socket.close();
+
+    }
+
+    @Test
+    public void sendTest1() throws Exception {
+        NioUDPDataSender sender = new NioUDPDataSender("localhost", PORT, "test", 128, 1000, 1024 * 64 * 100);
+
+        int sendMessageCount = 10;
+        TAgentInfo agentInfo = new TAgentInfo();
+        for (int i = 0; i < 10; i++) {
+            sender.send(agentInfo);
+        }
+
+        try {
+            waitMessageReceived(sendMessageCount);
+        } finally {
+            sender.stop();
+        }
+    }
+
+    @Test(expected = IOException.class)
+    public void exceedMessageSendTest() throws InterruptedException, IOException {
+        String random = RandomStringUtils.randomAlphabetic(UdpDataSender.UDP_MAX_PACKET_LENGTH + 100);
+
+        TAgentInfo agentInfo = new TAgentInfo();
+        agentInfo.setAgentId(random);
+
+        NioUDPDataSender sender = new NioUDPDataSender("localhost", PORT, "test", 128, 1000, 1024 * 64 * 100);
+        sender.send(agentInfo);
+
+        waitMessageReceived(1);
+    }
+
+
+    private boolean sendMessage_getLimit(TBase tbase, long waitTimeMillis) throws InterruptedException {
+        final AtomicBoolean limitCounter = new AtomicBoolean(false);
+        final CountDownLatch latch = new CountDownLatch(1);
+
+        NioUDPDataSender sender = new NioUDPDataSender("localhost", PORT, "test", 128, 1000, 1024 * 64 * 100);
+        try {
+            sender.send(tbase);
+            latch.await(waitTimeMillis, TimeUnit.MILLISECONDS);
+        } finally {
+            sender.stop();
+        }
+        return limitCounter.get();
+    }
+
+
+    private void waitMessageReceived(int expectReceiveMessageCount) throws IOException {
+        byte[] receiveData = new byte[65535];
+        DatagramPacket datagramPacket = new DatagramPacket(receiveData, 0, receiveData.length);
+
+        int remainCount = expectReceiveMessageCount;
+
+        while (remainCount > 0) {
+            remainCount--;
+            receiver.receive(datagramPacket);
+        }
+    }
+
+}
@@ -39,11 +39,13 @@ profiler.spandatasender.write.queue.size=5120
 #profiler.spandatasender.socket.sendbuffersize=1048576
 #profiler.spandatasender.socket.timeout=3000
 profiler.spandatasender.chunk.size=16384
+profiler.spandatasender.socket.type=OIO

 profiler.statdatasender.write.queue.size=5120
 #profiler.statdatasender.socket.sendbuffersize=1048576
 #profiler.statdatasender.socket.timeout=3000
 profiler.statdatasender.chunk.size=16384
+profiler.statdatasender.socket.type=OIO

 profiler.agentInfo.send.retry.interval=300000

@@ -39,11 +39,13 @@ profiler.spandatasender.write.queue.size=5120
 #profiler.spandatasender.socket.sendbuffersize=1048576
 #profiler.spandatasender.socket.timeout=3000
 profiler.spandatasender.chunk.size=16384
+profiler.spandatasender.socket.type=OIO

 profiler.statdatasender.write.queue.size=5120
 #profiler.statdatasender.socket.sendbuffersize=1048576
 #profiler.statdatasender.socket.timeout=3000
 profiler.statdatasender.chunk.size=16384
+profiler.statdatasender.socket.type=OIO

 profiler.agentInfo.send.retry.interval=300000

@@ -0,0 +1,89 @@
+package com.navercorp.pinpoint.thrift.io;
+
+import sun.misc.Cleaner;
+import sun.nio.ch.DirectBuffer;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+
+/**
+ * Not Thread Safe.
+ *
+ * @Author Taejin Koo
+ */
+public class ByteBufferOutputStream extends OutputStream implements ResettableOutputStream {
+
+    private final ByteBuffer byteBuffer;
+
+    public ByteBufferOutputStream(ByteBuffer byteBuffer) {
+        this.byteBuffer = byteBuffer;
+    }
+
+    @Override
+    public void write(int b) throws IOException {
+        checkWriteAvailable(1);
+        byteBuffer.put((byte) b);
+    }
+
+    @Override
+    public void write(byte[] byteArray, int offset, int length) throws IOException {
+        checkWriteAvailable(length);
+        byteBuffer.put(byteArray, offset, length);
+    }
+
+    private void checkWriteAvailable(int size) {
+        if (byteBuffer.remaining() >= size) {
+            return;
+        }
+
+        byteBuffer.limit(byteBuffer.capacity());
+        if (byteBuffer.remaining() < size) {
+            throw new BufferOverflowException("write failed remaining-size:" + byteBuffer.remaining() + ", input-size:" + size + ".");
+        }
+    }
+
+    @Override
+    public void flush() throws IOException {
+        // ignore
+    }
+
+    @Override
+    public void close() throws IOException {
+        deallocate(byteBuffer);
+    }
+
+    private void deallocate(ByteBuffer byteBuffer) {
+        if (byteBuffer == null) {
+            return;
+        }
+
+        if (byteBuffer instanceof DirectBuffer) {
+            Cleaner cleaner = ((DirectBuffer) byteBuffer).cleaner();
+            if (cleaner != null) {
+                cleaner.clean();
+            }
+        }
+    }
+
+
+    public ByteBuffer getByteBuffer() {
+        byteBuffer.flip();
+        return byteBuffer;
+    }
+
+    public void clear() {
+        byteBuffer.clear();
+    }
+
+    @Override
+    public void mark() {
+        byteBuffer.mark();
+    }
+
+    @Override
+    public void resetToMarkIndex() {
+        byteBuffer.reset();
+    }
+
+}
@@ -0,0 +1,50 @@
+package com.navercorp.pinpoint.thrift.io;
+
+import org.apache.thrift.TBase;
+import org.apache.thrift.TException;
+import org.apache.thrift.protocol.TProtocol;
+import org.apache.thrift.protocol.TProtocolFactory;
+
+import java.io.OutputStream;
+
+/**
+ *
+ * Caution. not thread safe
+ *
+ * @Author Taejin Koo
+ */
+public class HeaderTBaseSerializer2 {
+
+    private static final String UTF8 = "UTF8";
+
+    private final TOutputStreamTransport tOutputStreamTransport;
+    private final TProtocol protocol;
+    private final TBaseLocator tBaseLocator;
+
+    public HeaderTBaseSerializer2(TProtocolFactory protocolFactory, TBaseLocator tBaseLocator) {
+        this.tOutputStreamTransport = new TOutputStreamTransport();
+        this.protocol = protocolFactory.getProtocol(tOutputStreamTransport);
+        this.tBaseLocator = tBaseLocator;
+    }
+
+    public void serialize(TBase<?, ?> base, OutputStream outputStream) throws TException {
+        tOutputStreamTransport.open(outputStream);
+        try {
+            final Header header = tBaseLocator.headerLookup(base);
+            writeHeader(header);
+            base.write(protocol);
+        } finally {
+            tOutputStreamTransport.close();
+        }
+    }
+
+    private void writeHeader(Header header) throws TException {
+        protocol.writeByte(header.getSignature());
+        protocol.writeByte(header.getVersion());
+        // fixed size regardless protocol
+        short type = header.getType();
+        protocol.writeByte(BytesUtils.writeShort1(type));
+        protocol.writeByte(BytesUtils.writeShort2(type));
+    }
+
+}
@@ -0,0 +1,39 @@
+package com.navercorp.pinpoint.thrift.io;
+
+import org.apache.thrift.protocol.TCompactProtocol;
+import org.apache.thrift.protocol.TProtocolFactory;
+
+/**
+ * @Author Taejin Koo
+ */
+public class HeaderTBaseSerializerFactory2 implements SerializerFactory<HeaderTBaseSerializer2> {
+
+    private static final TProtocolFactory DEFAULT_PROTOCOL_FACTORY = new TCompactProtocol.Factory();
+    private static final TBaseLocator DEFAULT_TBASE_LOCATOR = new DefaultTBaseLocator();
+
+    private final TProtocolFactory protocolFactory;
+    private final TBaseLocator tBaseLocator;
+
+    public HeaderTBaseSerializerFactory2() {
+        this(DEFAULT_PROTOCOL_FACTORY, DEFAULT_TBASE_LOCATOR);
+    }
+
+    public HeaderTBaseSerializerFactory2(TProtocolFactory protocolFactory) {
+        this(protocolFactory, DEFAULT_TBASE_LOCATOR);
+    }
+
+    public HeaderTBaseSerializerFactory2(TBaseLocator tBaseLocator) {
+        this(DEFAULT_PROTOCOL_FACTORY, tBaseLocator);
+    }
+
+    public HeaderTBaseSerializerFactory2(TProtocolFactory protocolFactory, TBaseLocator tBaseLocator) {
+        this.protocolFactory = protocolFactory;
+        this.tBaseLocator = tBaseLocator;
+    }
+
+    @Override
+    public HeaderTBaseSerializer2 createSerializer() {
+        return new HeaderTBaseSerializer2(protocolFactory, tBaseLocator);
+    }
+
+}
@@ -0,0 +1,12 @@
+package com.navercorp.pinpoint.thrift.io;
+
+/**
+ * @Author Taejin Koo
+ */
+public interface ResettableOutputStream {
+
+    void mark();
+
+    void resetToMarkIndex();
+
+}
@@ -0,0 +1,55 @@
+package com.navercorp.pinpoint.thrift.io;
+
+import org.apache.thrift.transport.TTransport;
+import org.apache.thrift.transport.TTransportException;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+/**
+ *
+ * Caution. not thread safe
+ *
+ * @Author Taejin Koo
+ */
+public class TOutputStreamTransport extends TTransport {
+
+    private OutputStream outputStream;
+
+    @Override
+    public boolean isOpen() {
+        return outputStream != null;
+    }
+
+    @Override
+    public void open() throws TTransportException {
+    }
+
+    public void open(OutputStream outputStream) throws TTransportException {
+        this.outputStream = outputStream;
+    }
+
+    @Override
+    public void close() {
+        this.outputStream = null;
+    }
+
+    @Override
+    public int read(byte[] bytes, int index, int length) throws TTransportException {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public void write(byte[] bytes, int index, int length) throws TTransportException {
+        if (this.outputStream == null) {
+            throw new TTransportException(1, "TOutputStreamTransport is not opend.");
+        } else {
+            try {
+                this.outputStream.write(bytes, index, length);
+            } catch (IOException var5) {
+                throw new TTransportException(0, var5);
+            }
+        }
+    }
+
+}