From 93be9d1909d231b7176292fb565e5963c157fd57 Mon Sep 17 00:00:00 2001
From: jason huang <jasonwshuang@gmail.com>
Date: Tue, 9 Jan 2018 18:21:14 +0800
Subject: [PATCH 1/2] dump connection disconnected

dump connection will be disconnected since SocketChannel.cache is overflow.
---
 .../parse/driver/mysql/socket/SocketChannelPool.java  | 11 ++++++-----
 1 file changed, 6 insertions(+), 5 deletions(-)

diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannelPool.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannelPool.java
index 79a2f691..1b7db29f 100644
--- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannelPool.java
+++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannelPool.java
@@ -7,7 +7,7 @@ import io.netty.channel.AdaptiveRecvByteBufAllocator;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.SimpleChannelInboundHandler;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.EventLoopGroup;
@@ -71,7 +71,7 @@ public abstract class SocketChannelPool {
         return socket;
     }
 
-    public static class BusinessHandler extends ChannelInboundHandlerAdapter {
+    public static class BusinessHandler extends SimpleChannelInboundHandler<ByteBuf> {
 
         private SocketChannel        socket = null;
         private final CountDownLatch latch  = new CountDownLatch(1);
@@ -92,18 +92,19 @@ public abstract class SocketChannelPool {
         }
 
         @Override
-        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+        protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
             if (socket != null) {
-                socket.writeCache((ByteBuf) msg);
+                socket.writeCache(msg);
             } else {
                 // TODO: need graceful error handler.
                 logger.error("no socket available.");
             }
-            ReferenceCountUtil.release(msg);// 添加防止内存泄漏的
         }
 
         @Override
         public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+            //need output error for troubeshooting.
+            logger.error("business error.", cause);
             ctx.close();
         }
     }

From a965d2cd04b49f07b1694e83223437a0d723acc3 Mon Sep 17 00:00:00 2001
From: jason huang <jasonwshuang@gmail.com>
Date: Tue, 9 Jan 2018 18:25:51 +0800
Subject: [PATCH 2/2] dump connection is disconnected

dump connection is disconnected since SocketChannel.cache is overflow.
---
 .../driver/mysql/socket/SocketChannel.java     | 18 +++++++++++++++---
 1 file changed, 15 insertions(+), 3 deletions(-)

diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannel.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannel.java
index 40ce6c32..9eee2ef7 100644
--- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannel.java
+++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannel.java
@@ -27,10 +27,22 @@ public class SocketChannel {
         this.channel = channel;
     }
 
-    public void writeCache(ByteBuf buf) {
+    public void writeCache(ByteBuf buf) throws InterruptedException {
         synchronized (lock) {
-            cache.discardReadBytes();// 回收内存
-            cache.writeBytes(buf);
+            while (true) {
+                cache.discardReadBytes();// 回收内存
+                //source buffer is empty.
+                if (!buf.isReadable()) {
+                    break;
+                }
+
+                if (cache.isWritable()) {
+                    cache.writeBytes(buf, Math.min(cache.writableBytes(), buf.readableBytes()));
+                } else {
+                    //dest buffer is full.
+                    lock.wait(100);
+                }
+            }
         }
     }