refactor(canal): 升级JVM参数 (#5544)

- 重点升级JVM参数

- 重构SessionHandler代码, 处理错误请求会响应两次包；

client-adapter/launcher/src/main/bin/startup.sh

@@ -59,24 +59,34 @@ in
   exit;;
 esac
 
-JavaVersion=`$JAVA -version 2>&1 |awk 'NR==1{ gsub(/"/,""); print $3 }' | awk  -F '.' '{print $1}'`
-str=`file -L $JAVA | grep 64-bit`
+JAVA_VERSION=`$JAVA -version 2>&1 |awk 'NR==1{ gsub(/"/,""); print $3 }' | awk  -F '.' '{print $1}'`
 
-JAVA_OPTS="$JAVA_OPTS -Xss1m -XX:+AggressiveOpts -XX:-UseBiasedLocking -XX:-OmitStackTraceInFastThrow -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=$base/logs"
-if [ $JavaVersion -ge 11 ] ; then
+JAVA_OPTS="$JAVA_OPTS -Xss1m -XX:+AggressiveOpts -XX:-UseBiasedLocking -XX:-OmitStackTraceInFastThrow"
+JAVA_OPTS="$JAVA_OPTS -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=$base/logs"
+
+if [ $JAVA_VERSION -ge 11 ] ; then
   #JAVA_OPTS="$JAVA_OPTS -Xlog:gc*:$base_log/gc.log:time "
-  JAVA_OPTS="$JAVA_OPTS"
+  JAVA_OPTS="$JAVA_OPTS -Xlog:gc*:file=$base/logs/adapter/gc.log::filecount=5,filesize=32M"
 else
-  #JAVA_OPTS="$JAVA_OPTS -Xloggc:$base/logs/canal/gc.log -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+PrintGCApplicationStoppedTime"
-  JAVA_OPTS="$JAVA_OPTS -XX:+UseFastAccessorMethods -XX:+PrintAdaptiveSizePolicy -XX:+PrintTenuringDistribution"
+  JAVA_OPTS="$JAVA_OPTS -Xloggc:$base/logs/canal/gc.log -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=5 -XX:GCLogFileSize=32M"
+  JAVA_OPTS="$JAVA_OPTS -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:-PrintHeapAtGC -XX:+PrintGCApplicationStoppedTime"
+  JAVA_OPTS="$JAVA_OPTS -XX:+UseFastAccessorMethods -XX:-PrintReferenceGC -XX:+PrintAdaptiveSizePolicy -XX:+PrintTenuringDistribution"
+  JAVA_OPTS="$JAVA_OPTS -XX:+PrintSafepointStatistics -XX:PrintSafepointStatisticsCount=1 -XX:+SafepointTimeout -XX:SafepointTimeoutDelay=2000"
+  JAVA_OPTS="$JAVA_OPTS -XX:+UseCountedLoopSafepoints -XX:+DisableExplicitGC -XX:+ExplicitGCInvokesConcurrent -XX:+ExplicitGCInvokesConcurrentAndUnloadsClasses"
 fi
 
-if [ -n "$str" ]; then
-  if [ $JavaVersion -ge 11 ] ; then
+OS_ARCH=`file -L $JAVA | grep 64-bit`
+if [ -n "$OS_ARCH" ]; then
+  if [ $JAVA_VERSION -ge 11 ] ; then
     # For G1
     JAVA_OPTS="-server -Xms2g -Xmx3g -XX:+UseG1GC -XX:MaxGCPauseMillis=250 -XX:+UseGCOverheadLimit -XX:+ExplicitGCInvokesConcurrent $JAVA_OPTS"
   else
-	  JAVA_OPTS="-server -Xms2g -Xmx3g -Xmn1g -XX:SurvivorRatio=2 -XX:PermSize=96m -XX:MaxPermSize=256m -XX:MaxTenuringThreshold=15 -XX:+DisableExplicitGC $JAVA_OPTS"
+	  # JAVA_OPTS="-server -Xms2g -Xmx3g -Xmn1g -XX:SurvivorRatio=2 -XX:PermSize=96m -XX:MaxPermSize=256m -XX:MaxTenuringThreshold=15 -XX:+DisableExplicitGC $JAVA_OPTS"
+	  ## JDK 1.8 2c4g [-XX:ConcGCThreads=1 -XX:ParallelGCThreads=2], 4c8g [-XX:ConcGCThreads=2 -XX:ParallelGCThreads=4]
+    JAVA_OPTS="$JAVA_OPTS -XX:InitialRAMPercentage=45.0 -XX:MinRAMPercentage=45.0 -XX:MaxRAMPercentage=45.0"
+    JAVA_OPTS="$JAVA_OPTS -XX:-UseParallelGC -XX:+UseConcMarkSweepGC -XX:ConcGCThreads=2 -XX:ParallelGCThreads=4"
+    JAVA_OPTS="$JAVA_OPTS -XX:+UseCMSInitiatingOccupancyOnly -XX:CMSInitiatingOccupancyFraction=70"
+    JAVA_OPTS="-server -XX:NewRatio=1 -XX:SurvivorRatio=2 -XX:MetaspaceSize=96m -XX:MaxMetaspaceSize=128m "
 	fi
 else
 	JAVA_OPTS="-server -Xms1024m -Xmx1024m -XX:NewSize=256m -XX:MaxNewSize=256m -XX:MaxPermSize=128m $JAVA_OPTS"

deployer/src/main/bin/startup.sh

@@ -83,24 +83,35 @@ in
 	exit;;
 esac
 
-JavaVersion=`$JAVA -version 2>&1 |awk 'NR==1{ gsub(/"/,""); print $3 }' | awk  -F '.' '{print $1}'`
-str=`file -L $JAVA | grep 64-bit`
-JAVA_OPTS="$JAVA_OPTS -Xss1m -XX:-UseBiasedLocking -XX:-OmitStackTraceInFastThrow -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=$base/logs"
+JAVA_VERSION=`$JAVA -version 2>&1 |awk 'NR==1{ gsub(/"/,""); print $3 }' | awk  -F '.' '{print $1}'`
 
-if [ $JavaVersion -ge 11 ] ; then
+JAVA_OPTS="$JAVA_OPTS -Xss1m -XX:+AggressiveOpts -XX:-UseBiasedLocking -XX:-OmitStackTraceInFastThrow"
+JAVA_OPTS="$JAVA_OPTS -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=$base/logs"
+
+if [ $JAVA_VERSION -ge 11 ] ; then
   #JAVA_OPTS="$JAVA_OPTS -Xlog:gc*:$base_log/gc.log:time "
-  JAVA_OPTS="$JAVA_OPTS"
+  JAVA_OPTS="$JAVA_OPTS -Xlog:gc*:file=$base/logs/adapter/gc.log::filecount=5,filesize=32M"
 else
-  #JAVA_OPTS="$JAVA_OPTS -Xloggc:$base/logs/canal/gc.log -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+PrintGCApplicationStoppedTime"
-  JAVA_OPTS="$JAVA_OPTS -XX:+AggressiveOpts -XX:+UseFastAccessorMethods -XX:+PrintAdaptiveSizePolicy -XX:+PrintTenuringDistribution"
+  JAVA_OPTS="$JAVA_OPTS -Xloggc:$base/logs/canal/gc.log -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=5 -XX:GCLogFileSize=32M"
+  JAVA_OPTS="$JAVA_OPTS -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:-PrintHeapAtGC -XX:+PrintGCApplicationStoppedTime"
+  JAVA_OPTS="$JAVA_OPTS -XX:+UseFastAccessorMethods -XX:-PrintReferenceGC -XX:+PrintAdaptiveSizePolicy -XX:+PrintTenuringDistribution"
+  JAVA_OPTS="$JAVA_OPTS -XX:+PrintSafepointStatistics -XX:PrintSafepointStatisticsCount=1 -XX:+SafepointTimeout -XX:SafepointTimeoutDelay=2000"
+  JAVA_OPTS="$JAVA_OPTS -XX:+UseCountedLoopSafepoints -XX:+DisableExplicitGC -XX:+ExplicitGCInvokesConcurrent -XX:+ExplicitGCInvokesConcurrentAndUnloadsClasses"
 fi
 
-if [ -n "$str" ]; then
-  if [ $JavaVersion -ge 11 ] ; then
+OS_ARCH=`file -L $JAVA | grep 64-bit`
+
+if [ -n "$OS_ARCH" ]; then
+  if [ $JAVA_VERSION -ge 11 ] ; then
     # For G1
     JAVA_OPTS="-server -Xms2g -Xmx3g -XX:+UseG1GC -XX:MaxGCPauseMillis=250 -XX:+UseGCOverheadLimit -XX:+ExplicitGCInvokesConcurrent $JAVA_OPTS"
   else
-	  JAVA_OPTS="-server -Xms2g -Xmx3g -Xmn1g -XX:SurvivorRatio=2 -XX:PermSize=96m -XX:MaxPermSize=256m -XX:MaxTenuringThreshold=15 -XX:+DisableExplicitGC $JAVA_OPTS"
+	  # JAVA_OPTS="-server -Xms2g -Xmx3g -Xmn1g -XX:SurvivorRatio=2 -XX:PermSize=96m -XX:MaxPermSize=256m -XX:MaxTenuringThreshold=15 -XX:+DisableExplicitGC $JAVA_OPTS"
+	  ## JDK 1.8 2c4g [-XX:ConcGCThreads=1 -XX:ParallelGCThreads=2], 4c8g [-XX:ConcGCThreads=2 -XX:ParallelGCThreads=4]
+	  JAVA_OPTS="$JAVA_OPTS -XX:InitialRAMPercentage=45.0 -XX:MinRAMPercentage=45.0 -XX:MaxRAMPercentage=45.0"
+	  JAVA_OPTS="$JAVA_OPTS -XX:-UseParallelGC -XX:+UseConcMarkSweepGC -XX:ConcGCThreads=2 -XX:ParallelGCThreads=4"
+	  JAVA_OPTS="$JAVA_OPTS -XX:+UseCMSInitiatingOccupancyOnly -XX:CMSInitiatingOccupancyFraction=70"
+	  JAVA_OPTS="-server -XX:NewRatio=1 -XX:SurvivorRatio=2 -XX:MetaspaceSize=96m -XX:MaxMetaspaceSize=128m "
 	fi
 else
 	JAVA_OPTS="-server -Xms1024m -Xmx1024m -XX:NewSize=256m -XX:MaxNewSize=256m -XX:MaxPermSize=128m $JAVA_OPTS"

server/src/main/java/com/alibaba/otter/canal/admin/handler/SessionHandler.java

@@ -31,127 +31,114 @@ public class SessionHandler extends SimpleChannelHandler {
     }
 
     public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
-        logger.info("message receives in session handler...");
         ChannelBuffer buffer = (ChannelBuffer) e.getMessage();
         Packet packet = Packet.parseFrom(buffer.readBytes(buffer.readableBytes()).array());
         try {
-            String action = null;
-            String message = null;
-            String destination = null;
             switch (packet.getType()) {
                 case SERVER:
-                    ServerAdmin serverAdmin = ServerAdmin.parseFrom(packet.getBody());
-                    action = serverAdmin.getAction();
-                    switch (action) {
+                    ServerAdmin sa = ServerAdmin.parseFrom(packet.getBody());
+                    logger.info("received {} SERVER request in session handler...", sa.getAction());
+                    byte[] svrResp = null;
+                    switch (sa.getAction()) {
                         case "check":
-                            message = canalAdmin.check() ? "1" : "0";
+                            svrResp = AdminNettyUtils.ackPacket(canalAdmin.check() ? "1" : "0");
                             break;
                         case "start":
-                            message = canalAdmin.start() ? "1" : "0";
+                            svrResp = AdminNettyUtils.ackPacket(canalAdmin.start() ? "1" : "0");
                             break;
                         case "stop":
-                            message = canalAdmin.stop() ? "1" : "0";
+                            svrResp = AdminNettyUtils.ackPacket(canalAdmin.stop() ? "1" : "0");
                             break;
                         case "restart":
-                            message = canalAdmin.restart() ? "1" : "0";
+                            svrResp = AdminNettyUtils.ackPacket(canalAdmin.restart() ? "1" : "0");
                             break;
                         case "list":
-                            message = canalAdmin.getRunningInstances();
+                            svrResp = AdminNettyUtils.ackPacket(canalAdmin.getRunningInstances());
                             break;
                         default:
-                            byte[] errorBytes = AdminNettyUtils.errorPacket(301,
-                                MessageFormatter.format("ServerAdmin action={} is unknown", action).getMessage());
-                            AdminNettyUtils.write(ctx.getChannel(), errorBytes);
+                            svrResp = AdminNettyUtils.errorPacket(301, "ServerAdmin action: " + sa.getAction() + "  is unknown");
                             break;
                     }
-                    AdminNettyUtils.write(ctx.getChannel(), AdminNettyUtils.ackPacket(message));
+                    AdminNettyUtils.write(ctx.getChannel(), svrResp);
                     break;
+
                 case INSTANCE:
-                    InstanceAdmin instanceAdmin = InstanceAdmin.parseFrom(packet.getBody());
-                    destination = instanceAdmin.getDestination();
-                    action = instanceAdmin.getAction();
-                    switch (action) {
+                    InstanceAdmin ia = InstanceAdmin.parseFrom(packet.getBody());
+                    logger.info("received {} INSTANCE request in session handler...", ia.getAction());
+                    byte[] instResp = null;
+                    switch (ia.getAction()) {
                         case "check":
-                            message = canalAdmin.checkInstance(destination) ? "1" : "0";
+                            instResp = AdminNettyUtils.ackPacket(canalAdmin.checkInstance(ia.getDestination()) ? "1" : "0");
                             break;
                         case "start":
-                            message = canalAdmin.startInstance(destination) ? "1" : "0";
+                            instResp = AdminNettyUtils.ackPacket(canalAdmin.startInstance(ia.getDestination()) ? "1" : "0");
                             break;
                         case "stop":
-                            message = canalAdmin.stopInstance(destination) ? "1" : "0";
+                            instResp = AdminNettyUtils.ackPacket(canalAdmin.stopInstance(ia.getDestination()) ? "1" : "0");
                             break;
                         case "release":
-                            message = canalAdmin.releaseInstance(destination) ? "1" : "0";
+                            instResp = AdminNettyUtils.ackPacket(canalAdmin.releaseInstance(ia.getDestination()) ? "1" : "0");
                             break;
                         case "restart":
-                            message = canalAdmin.restartInstance(destination) ? "1" : "0";
+                            instResp = AdminNettyUtils.ackPacket(canalAdmin.restartInstance(ia.getDestination()) ? "1" : "0");
                             break;
                         default:
-                            byte[] errorBytes = AdminNettyUtils.errorPacket(301,
-                                MessageFormatter.format("InstanceAdmin action={} is unknown", action).getMessage());
-                            AdminNettyUtils.write(ctx.getChannel(), errorBytes);
+                            instResp = AdminNettyUtils.errorPacket(301, "InstanceAdmin action: " + ia.getAction() + " is unknown");
                             break;
                     }
-                    AdminNettyUtils.write(ctx.getChannel(), AdminNettyUtils.ackPacket(message));
+                    AdminNettyUtils.write(ctx.getChannel(), instResp);
                     break;
+
                 case LOG:
-                    LogAdmin logAdmin = LogAdmin.parseFrom(packet.getBody());
-                    action = logAdmin.getAction();
-                    destination = logAdmin.getDestination();
-                    String type = logAdmin.getType();
-                    String file = logAdmin.getFile();
-                    int count = logAdmin.getCount();
-                    switch (type) {
+                    LogAdmin la = LogAdmin.parseFrom(packet.getBody());
+                    logger.info("received {} LOG request in session handler...", la.getType());
+                    byte[] logResp = null;
+                    switch (la.getType()) {
                         case "server":
-                            if ("list".equalsIgnoreCase(action)) {
-                                message = canalAdmin.listCanalLog();
+                            if ("list".equalsIgnoreCase(la.getAction())) {
+                                logResp = AdminNettyUtils.ackPacket(canalAdmin.listCanalLog());
                             } else {
-                                message = canalAdmin.canalLog(count);
+                                logResp = AdminNettyUtils.ackPacket(canalAdmin.canalLog(la.getCount()));
                             }
                             break;
                         case "instance":
-                            if ("list".equalsIgnoreCase(action)) {
-                                message = canalAdmin.listInstanceLog(destination);
+                            if ("list".equalsIgnoreCase(la.getAction())) {
+                                logResp = AdminNettyUtils.ackPacket(canalAdmin.listInstanceLog(la.getDestination()));
                             } else {
-                                message = canalAdmin.instanceLog(destination, file, count);
+                                logResp = AdminNettyUtils.ackPacket(canalAdmin.instanceLog(la.getDestination(), la.getFile(), la.getCount()));
                             }
                             break;
                         default:
-                            byte[] errorBytes = AdminNettyUtils.errorPacket(301,
-                                MessageFormatter.format("LogAdmin type={} is unknown", type).getMessage());
-                            AdminNettyUtils.write(ctx.getChannel(), errorBytes);
+                            logResp = AdminNettyUtils.errorPacket(301, "LogAdmin type: " + la.getType() + " is unknown");
                             break;
                     }
-                    AdminNettyUtils.write(ctx.getChannel(), AdminNettyUtils.ackPacket(message));
+                    AdminNettyUtils.write(ctx.getChannel(), logResp);
                     break;
+
                 default:
-                    byte[] errorBytes = AdminNettyUtils.errorPacket(300,
-                        MessageFormatter.format("packet type={} is NOT supported!", packet.getType()).getMessage());
-                    AdminNettyUtils.write(ctx.getChannel(), errorBytes);
+                    logResp = AdminNettyUtils.errorPacket(300, "packet type: " + packet.getType() + " is NOT supported!");
+                    AdminNettyUtils.write(ctx.getChannel(), logResp);
                     break;
             }
+
         } catch (Throwable exception) {
-            byte[] errorBytes = AdminNettyUtils.errorPacket(400,
-                MessageFormatter.format("something goes wrong with channel:{}, exception={}",
-                    ctx.getChannel(),
-                    ExceptionUtils.getStackTrace(exception)).getMessage());
-            AdminNettyUtils.write(ctx.getChannel(), errorBytes);
+            String error = "something goes wrong with channel: " + ctx.getChannel() + ", exception: "+ ExceptionUtils.getStackTrace(exception);
+            AdminNettyUtils.write(ctx.getChannel(), AdminNettyUtils.errorPacket(400, error));
         }
     }
 
-    public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
-        logger.error("something goes wrong with channel:{}, exception={}",
-            ctx.getChannel(),
-            ExceptionUtils.getStackTrace(e.getCause()));
-
-        ctx.getChannel().close();
-    }
-
-    public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
-    }
-
-    public void setCanalAdmin(CanalAdmin canalAdmin) {
-        this.canalAdmin = canalAdmin;
-    }
+//    public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
+//        logger.error("something goes wrong with channel:{}, exception={}",
+//            ctx.getChannel(),
+//            ExceptionUtils.getStackTrace(e.getCause()));
+//        ctx.getChannel().close();
+//    }
+//
+//    public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
+//    }
+//
+//    public void setCanalAdmin(CanalAdmin canalAdmin) {
+//        this.canalAdmin = canalAdmin;
+//    }
 
 }

server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java

@@ -52,19 +52,15 @@ public class SessionHandler extends SimpleChannelHandler {
 
     @SuppressWarnings({ "deprecation" })
     public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
-        logger.info("message receives in session handler...");
         long start = System.nanoTime();
         ChannelBuffer buffer = (ChannelBuffer) e.getMessage();
         Packet packet = Packet.parseFrom(buffer.readBytes(buffer.readableBytes()).array());
-        ClientIdentity clientIdentity = null;
         try {
             switch (packet.getType()) {
                 case SUBSCRIPTION:
                     Sub sub = Sub.parseFrom(packet.getBody());
                     if (StringUtils.isNotEmpty(sub.getDestination()) && StringUtils.isNotEmpty(sub.getClientId())) {
-                        clientIdentity = new ClientIdentity(sub.getDestination(),
-                            Short.valueOf(sub.getClientId()),
-                            sub.getFilter());
+                        ClientIdentity clientIdentity = new ClientIdentity(sub.getDestination(), Short.parseShort(sub.getClientId()), sub.getFilter());
                         MDC.put("destination", clientIdentity.getDestination());
 
                         // 尝试启动，如果已经启动，忽略
@@ -84,8 +80,7 @@ public class SessionHandler extends SimpleChannelHandler {
                             ackBytes.length,
                             System.nanoTime() - start));
                     } else {
-                        byte[] errorBytes = NettyUtils.errorPacket(401,
-                            MessageFormatter.format("destination or clientId is null", sub.toString()).getMessage());
+                        byte[] errorBytes = NettyUtils.errorPacket(401, "destination or clientId is null. Sub: " + sub);
                         NettyUtils.write(ctx.getChannel(),
                             errorBytes,
                             new ChannelFutureAggregator(sub.getDestination(),
@@ -99,12 +94,12 @@ public class SessionHandler extends SimpleChannelHandler {
                 case UNSUBSCRIPTION:
                     Unsub unsub = Unsub.parseFrom(packet.getBody());
                     if (StringUtils.isNotEmpty(unsub.getDestination()) && StringUtils.isNotEmpty(unsub.getClientId())) {
-                        clientIdentity = new ClientIdentity(unsub.getDestination(),
-                            Short.valueOf(unsub.getClientId()),
-                            unsub.getFilter());
+                        ClientIdentity clientIdentity = new ClientIdentity(unsub.getDestination(), Short.parseShort(unsub.getClientId()), unsub.getFilter());
                         MDC.put("destination", clientIdentity.getDestination());
+
                         embeddedServer.unsubscribe(clientIdentity);
                         stopCanalInstanceIfNecessary(clientIdentity);// 尝试关闭
+
                         byte[] ackBytes = NettyUtils.ackPacket();
                         NettyUtils.write(ctx.getChannel(),
                             ackBytes,
@@ -114,8 +109,8 @@ public class SessionHandler extends SimpleChannelHandler {
                                 ackBytes.length,
                                 System.nanoTime() - start));
                     } else {
-                        byte[] errorBytes = NettyUtils.errorPacket(401,
-                            MessageFormatter.format("destination or clientId is null", unsub.toString()).getMessage());
+                        String error = "destination or clientId is null. Unsub: " + unsub;
+                        byte[] errorBytes = NettyUtils.errorPacket(401, error);
                         NettyUtils.write(ctx.getChannel(),
                             errorBytes,
                             new ChannelFutureAggregator(unsub.getDestination(),
@@ -129,8 +124,9 @@ public class SessionHandler extends SimpleChannelHandler {
                 case GET:
                     Get get = CanalPacket.Get.parseFrom(packet.getBody());
                     if (StringUtils.isNotEmpty(get.getDestination()) && StringUtils.isNotEmpty(get.getClientId())) {
-                        clientIdentity = new ClientIdentity(get.getDestination(), Short.valueOf(get.getClientId()));
+                        ClientIdentity clientIdentity = new ClientIdentity(get.getDestination(), Short.parseShort(get.getClientId()));
                         MDC.put("destination", clientIdentity.getDestination());
+
                         Message message = null;
 
                         // if (get.getAutoAck()) {
@@ -147,10 +143,7 @@ public class SessionHandler extends SimpleChannelHandler {
                             message = embeddedServer.getWithoutAck(clientIdentity, get.getFetchSize());
                         } else {
                             TimeUnit unit = convertTimeUnit(get.getUnit());
-                            message = embeddedServer.getWithoutAck(clientIdentity,
-                                get.getFetchSize(),
-                                get.getTimeout(),
-                                unit);
+                            message = embeddedServer.getWithoutAck(clientIdentity, get.getFetchSize(), get.getTimeout(), unit);
                         }
                         // }
 
@@ -165,7 +158,9 @@ public class SessionHandler extends SimpleChannelHandler {
                                 dataSize += CodedOutputStream.computeBytesSizeNoTag(rowEntry);
                             }
                             messageSize += dataSize;
-                            messageSize += 1 * rowEntries.size();
+                            // messageSize += 1 * rowEntries.size();
+                            messageSize += rowEntries.size();
+
                             // packet size
                             int size = 0;
                             size += com.google.protobuf.CodedOutputStream.computeEnumSize(3,
@@ -173,6 +168,7 @@ public class SessionHandler extends SimpleChannelHandler {
                             size += com.google.protobuf.CodedOutputStream.computeTagSize(5)
                                     + com.google.protobuf.CodedOutputStream.computeRawVarint32Size(messageSize)
                                     + messageSize;
+
                             // recyle bytes
                             // ByteBuffer byteBuffer = (ByteBuffer)
                             // ctx.getAttachment();
@@ -210,9 +206,6 @@ public class SessionHandler extends SimpleChannelHandler {
                             // NettyUtils.write(ctx.getChannel(), byteBuffer,
                             // null);
                         } else {
-                            Packet.Builder packetBuilder = CanalPacket.Packet.newBuilder();
-                            packetBuilder.setType(PacketType.MESSAGES).setVersion(NettyUtils.VERSION);
-
                             Messages.Builder messageBuilder = CanalPacket.Messages.newBuilder();
                             messageBuilder.setBatchId(message.getId());
                             if (message.getId() != -1) {
@@ -224,9 +217,10 @@ public class SessionHandler extends SimpleChannelHandler {
                                     }
                                 }
                             }
-                            byte[] body = packetBuilder.setBody(messageBuilder.build().toByteString())
-                                .build()
-                                .toByteArray();
+
+                            Packet.Builder packetBuilder = CanalPacket.Packet.newBuilder();
+                            packetBuilder.setType(PacketType.MESSAGES).setVersion(NettyUtils.VERSION);
+                            byte[] body = packetBuilder.setBody(messageBuilder.build().toByteString()).build().toByteArray();
                             NettyUtils.write(ctx.getChannel(), body, new ChannelFutureAggregator(get.getDestination(),
                                 get,
                                 packet.getType(),
@@ -235,8 +229,7 @@ public class SessionHandler extends SimpleChannelHandler {
                                 message.getId() == -1));// 输出数据
                         }
                     } else {
-                        byte[] errorBytes = NettyUtils.errorPacket(401,
-                            MessageFormatter.format("destination or clientId is null", get.toString()).getMessage());
+                        byte[] errorBytes = NettyUtils.errorPacket(401, "destination or clientId is null. Get: " + get);
                         NettyUtils.write(ctx.getChannel(),
                             errorBytes,
                             new ChannelFutureAggregator(get.getDestination(),
@@ -249,11 +242,10 @@ public class SessionHandler extends SimpleChannelHandler {
                     break;
                 case CLIENTACK:
                     ClientAck ack = CanalPacket.ClientAck.parseFrom(packet.getBody());
-                    MDC.put("destination", ack.getDestination());
                     if (StringUtils.isNotEmpty(ack.getDestination()) && StringUtils.isNotEmpty(ack.getClientId())) {
+                        MDC.put("destination", ack.getDestination());
                         if (ack.getBatchId() == 0L) {
-                            byte[] errorBytes = NettyUtils.errorPacket(402,
-                                MessageFormatter.format("batchId should assign value", ack.toString()).getMessage());
+                            byte[] errorBytes = NettyUtils.errorPacket(402, "batchId should assign value. Ack: "+ack);
                             NettyUtils.write(ctx.getChannel(),
                                 errorBytes,
                                 new ChannelFutureAggregator(ack.getDestination(),
@@ -265,7 +257,7 @@ public class SessionHandler extends SimpleChannelHandler {
                         } else if (ack.getBatchId() == -1L) { // -1代表上一次get没有数据，直接忽略之
                             // donothing
                         } else {
-                            clientIdentity = new ClientIdentity(ack.getDestination(), Short.valueOf(ack.getClientId()));
+                            ClientIdentity clientIdentity = new ClientIdentity(ack.getDestination(), Short.parseShort(ack.getClientId()));
                             embeddedServer.ack(clientIdentity, ack.getBatchId());
                             new ChannelFutureAggregator(ack.getDestination(),
                                 ack,
@@ -274,8 +266,7 @@ public class SessionHandler extends SimpleChannelHandler {
                                 System.nanoTime() - start).operationComplete(null);
                         }
                     } else {
-                        byte[] errorBytes = NettyUtils.errorPacket(401,
-                            MessageFormatter.format("destination or clientId is null", ack.toString()).getMessage());
+                        byte[] errorBytes = NettyUtils.errorPacket(401, "destination or clientId is null. Ack: " + ack);
                         NettyUtils.write(ctx.getChannel(),
                             errorBytes,
                             new ChannelFutureAggregator(ack.getDestination(),
@@ -288,25 +279,23 @@ public class SessionHandler extends SimpleChannelHandler {
                     break;
                 case CLIENTROLLBACK:
                     ClientRollback rollback = CanalPacket.ClientRollback.parseFrom(packet.getBody());
+                    if (StringUtils.isNotEmpty(rollback.getDestination()) && StringUtils.isNotEmpty(rollback.getClientId())) {
+                        ClientIdentity clientIdentity = new ClientIdentity(rollback.getDestination(), Short.parseShort(rollback.getClientId()));
                         MDC.put("destination", rollback.getDestination());
-                    if (StringUtils.isNotEmpty(rollback.getDestination())
-                        && StringUtils.isNotEmpty(rollback.getClientId())) {
-                        clientIdentity = new ClientIdentity(rollback.getDestination(),
-                            Short.valueOf(rollback.getClientId()));
+
                         if (rollback.getBatchId() == 0L) {
                             embeddedServer.rollback(clientIdentity);// 回滚所有批次
                         } else {
                             embeddedServer.rollback(clientIdentity, rollback.getBatchId()); // 只回滚单个批次
                         }
+
                         new ChannelFutureAggregator(rollback.getDestination(),
                             rollback,
                             packet.getType(),
                             0,
                             System.nanoTime() - start).operationComplete(null);
                     } else {
-                        byte[] errorBytes = NettyUtils.errorPacket(401,
-                            MessageFormatter.format("destination or clientId is null", rollback.toString())
-                                .getMessage());
+                        byte[] errorBytes = NettyUtils.errorPacket(401, "destination or clientId is null. Rollback: " + rollback);
                         NettyUtils.write(ctx.getChannel(),
                             errorBytes,
                             new ChannelFutureAggregator(rollback.getDestination(),
@@ -318,42 +307,38 @@ public class SessionHandler extends SimpleChannelHandler {
                     }
                     break;
                 default:
-                    byte[] errorBytes = NettyUtils.errorPacket(400,
-                        MessageFormatter.format("packet type={} is NOT supported!", packet.getType()).getMessage());
+                    byte[] errorBytes = NettyUtils.errorPacket(400, "packet type: " + packet.getType() + " is NOT supported!");
                     NettyUtils.write(ctx.getChannel(), errorBytes, new ChannelFutureAggregator(ctx.getChannel()
                         .getRemoteAddress()
                         .toString(), null, packet.getType(), errorBytes.length, System.nanoTime() - start, (short) 400));
                     break;
             }
         } catch (Throwable exception) {
-            byte[] errorBytes = NettyUtils.errorPacket(400,
-                MessageFormatter.format("something goes wrong with channel:{}, exception={}",
-                    ctx.getChannel(),
-                    ExceptionUtils.getStackTrace(exception)).getMessage());
+            String error = "something goes wrong with channel: " + ctx.getChannel() + ", exception: " + ExceptionUtils.getStackTrace(exception);
+            byte[] errorBytes = NettyUtils.errorPacket(400, error);
             NettyUtils.write(ctx.getChannel(), errorBytes, new ChannelFutureAggregator(ctx.getChannel()
-                .getRemoteAddress()
-                .toString(), null, packet.getType(), errorBytes.length, System.nanoTime() - start, (short) 400));
+                .getRemoteAddress().toString(), null, packet.getType(), errorBytes.length, System.nanoTime() - start, (short) 400));
         } finally {
             MDC.remove("destination");
         }
     }
 
-    public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
-        logger.error("something goes wrong with channel:{}, exception={}",
-            ctx.getChannel(),
-            ExceptionUtils.getStackTrace(e.getCause()));
-
-        ctx.getChannel().close();
-    }
-
-    public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
-        // logger.info("remove binding subscription value object if any...");
-        // ClientIdentity clientIdentity = (ClientIdentity) ctx.getAttachment();
-        // // 如果唯一的订阅者都取消了订阅，直接关闭服务，针对内部版本模式下可以减少资源浪费
-        // if (clientIdentity != null) {
-        // stopCanalInstanceIfNecessary(clientIdentity);
+//    public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
+//        logger.error("something goes wrong with channel:{}, exception={}",
+//            ctx.getChannel(),
+//            ExceptionUtils.getStackTrace(e.getCause()));
+//
+//        ctx.getChannel().close();
+//    }
+
+//    public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
+//        // logger.info("remove binding subscription value object if any...");
+//        // ClientIdentity clientIdentity = (ClientIdentity) ctx.getAttachment();
+//        // // 如果唯一的订阅者都取消了订阅，直接关闭服务，针对内部版本模式下可以减少资源浪费
+//        // if (clientIdentity != null) {
+//        // stopCanalInstanceIfNecessary(clientIdentity);
+//        // }
 //    }
-    }
 
     private void stopCanalInstanceIfNecessary(ClientIdentity clientIdentity) {
         List<ClientIdentity> clientIdentitys = embeddedServer.listAllSubscribe(clientIdentity.getDestination());
@@ -386,8 +371,8 @@ public class SessionHandler extends SimpleChannelHandler {
         }
     }
 
-    public void setEmbeddedServer(CanalServerWithEmbedded embeddedServer) {
-        this.embeddedServer = embeddedServer;
-    }
+//    public void setEmbeddedServer(CanalServerWithEmbedded embeddedServer) {
+//        this.embeddedServer = embeddedServer;
+//    }
 
 }