Merge pull request #2 from alibaba/master

merge master

README.md

@@ -80,3 +80,10 @@ See the wiki page for : <a href="https://github.com/alibaba/canal/wiki" >wiki文
 <li>新浪微博： agapple0002 </li>
 <li>报告issue：<a href="https://github.com/alibaba/canal/issues">issues</a></li>
 </ol>
+
+<pre>
+【招聘】阿里巴巴中间件团队招聘JAVA高级工程师
+岗位主要为技术型内容(非业务部门)，阿里中间件整个体系对于未来想在技术上有所沉淀的同学还是非常有帮助的
+工作地点：杭州、北京均可. ps. 阿里待遇向来都是不错的，有意者可以QQ、微博私聊. 
+具体招聘内容：https://job.alibaba.com/zhaopin/position_detail.htm?positionId=32666
+</pre>

client/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.0.24-SNAPSHOT</version>
+		<version>1.0.25-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>

client/src/main/java/com/alibaba/otter/canal/client/impl/ClusterCanalConnector.java

@@ -108,7 +108,7 @@ public class ClusterCanalConnector implements CanalConnector {
                 return;
             } catch (Throwable t) {
                 logger.warn("something goes wrong when subscribing from server:{}\n{}",
-                    currentConnector.getAddress(),
+                    currentConnector != null ? currentConnector.getAddress() : "null",
                     ExceptionUtils.getFullStackTrace(t));
                 times++;
                 restart();
@@ -127,7 +127,7 @@ public class ClusterCanalConnector implements CanalConnector {
                 return;
             } catch (Throwable t) {
                 logger.warn("something goes wrong when unsubscribing from server:{}\n{}",
-                    currentConnector.getAddress(),
+                    currentConnector != null ? currentConnector.getAddress() : "null",
                     ExceptionUtils.getFullStackTrace(t));
                 times++;
                 restart();
@@ -145,7 +145,7 @@ public class ClusterCanalConnector implements CanalConnector {
                 return msg;
             } catch (Throwable t) {
                 logger.warn("something goes wrong when getting data from server:{}\n{}",
-                    currentConnector.getAddress(),
+                    currentConnector != null ? currentConnector.getAddress() : "null",
                     ExceptionUtils.getFullStackTrace(t));
                 times++;
                 restart();
@@ -163,7 +163,7 @@ public class ClusterCanalConnector implements CanalConnector {
                 return msg;
             } catch (Throwable t) {
                 logger.warn("something goes wrong when getting data from server:{}\n{}",
-                    currentConnector.getAddress(),
+                    currentConnector != null ? currentConnector.getAddress() : "null",
                     ExceptionUtils.getFullStackTrace(t));
                 times++;
                 restart();

client/src/main/java/com/alibaba/otter/canal/client/impl/ClusterNodeAccessStrategy.java

@@ -114,5 +114,4 @@ public class ClusterNodeAccessStrategy implements CanalNodeAccessStrategy {
         return zkClient;
     }
 
-
 }

client/src/main/java/com/alibaba/otter/canal/client/impl/SimpleCanalConnector.java

@@ -5,7 +5,11 @@ import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
+import java.nio.channels.Channel;
+import java.nio.channels.Channels;
+import java.nio.channels.ReadableByteChannel;
 import java.nio.channels.SocketChannel;
+import java.nio.channels.WritableByteChannel;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
@@ -37,7 +41,6 @@ import com.alibaba.otter.canal.protocol.ClientIdentity;
 import com.alibaba.otter.canal.protocol.Message;
 import com.alibaba.otter.canal.protocol.exception.CanalClientException;
 import com.google.protobuf.ByteString;
-import com.google.protobuf.InvalidProtocolBufferException;
 
 /**
  * 基于{@linkplain CanalServerWithNetty}定义的网络协议接口，对于canal数据进行get/rollback/ack等操作
@@ -57,6 +60,8 @@ public class SimpleCanalConnector implements CanalConnector {
     private final ByteBuffer     readHeader            = ByteBuffer.allocate(4).order(ByteOrder.BIG_ENDIAN);
     private final ByteBuffer     writeHeader           = ByteBuffer.allocate(4).order(ByteOrder.BIG_ENDIAN);
     private SocketChannel        channel;
+    private ReadableByteChannel  readableChannel;
+    private WritableByteChannel  writableChannel;
     private List<Compression>    supportedCompressions = new ArrayList<Compression>();
     private ClientIdentity       clientIdentity;
     private ClientRunningMonitor runningMonitor;                                                             // 运行控制
@@ -130,7 +135,9 @@ public class SimpleCanalConnector implements CanalConnector {
                 address = getNextAddress();
             }
             channel.connect(address);
-            Packet p = Packet.parseFrom(readNextPacket(channel));
+            readableChannel = Channels.newChannel(channel.socket().getInputStream());
+            writableChannel = Channels.newChannel(channel.socket().getOutputStream());
+            Packet p = Packet.parseFrom(readNextPacket());
             if (p.getVersion() != 1) {
                 throw new CanalClientException("unsupported version at this client.");
             }
@@ -144,17 +151,17 @@ public class SimpleCanalConnector implements CanalConnector {
             //
             ClientAuth ca = ClientAuth.newBuilder()
                 .setUsername(username != null ? username : "")
+                .setPassword(ByteString.copyFromUtf8(password != null ? password : ""))
                 .setNetReadTimeout(soTimeout)
                 .setNetWriteTimeout(soTimeout)
                 .build();
-            writeWithHeader(channel,
+            writeWithHeader(Packet.newBuilder()
-                Packet.newBuilder()
                 .setType(PacketType.CLIENTAUTHENTICATION)
                 .setBody(ca.toByteString())
                 .build()
                 .toByteArray());
             //
-            Packet ack = Packet.parseFrom(readNextPacket(channel));
+            Packet ack = Packet.parseFrom(readNextPacket());
             if (ack.getType() != PacketType.ACK) {
                 throw new CanalClientException("unexpected packet type when ack is expected");
             }
@@ -173,14 +180,26 @@ public class SimpleCanalConnector implements CanalConnector {
     }
 
     private void doDisconnnect() throws CanalClientException {
+        if (readableChannel != null) {
+            quietlyClose(readableChannel);
+            readableChannel = null;
+        }
+        if (writableChannel != null) {
+            quietlyClose(writableChannel);
+            writableChannel = null;
+        }
         if (channel != null) {
+            quietlyClose(channel);
+            channel = null;
+        }
+    }
+
+    private void quietlyClose(Channel channel) {
         try {
             channel.close();
         } catch (IOException e) {
             logger.warn("exception on closing channel:{} \n {}", channel, e);
         }
-            channel = null;
-        }
     }
 
     public void subscribe() throws CanalClientException {
@@ -190,8 +209,7 @@ public class SimpleCanalConnector implements CanalConnector {
     public void subscribe(String filter) throws CanalClientException {
         waitClientRunning();
         try {
-            writeWithHeader(channel,
+            writeWithHeader(Packet.newBuilder()
-                Packet.newBuilder()
                 .setType(PacketType.SUBSCRIPTION)
                 .setBody(Sub.newBuilder()
                     .setDestination(clientIdentity.getDestination())
@@ -202,7 +220,7 @@ public class SimpleCanalConnector implements CanalConnector {
                 .build()
                 .toByteArray());
             //
-            Packet p = Packet.parseFrom(readNextPacket(channel));
+            Packet p = Packet.parseFrom(readNextPacket());
             Ack ack = Ack.parseFrom(p.getBody());
             if (ack.getErrorCode() > 0) {
                 throw new CanalClientException("failed to subscribe with reason: " + ack.getErrorMessage());
@@ -217,8 +235,7 @@ public class SimpleCanalConnector implements CanalConnector {
     public void unsubscribe() throws CanalClientException {
         waitClientRunning();
         try {
-            writeWithHeader(channel,
+            writeWithHeader(Packet.newBuilder()
-                Packet.newBuilder()
                 .setType(PacketType.UNSUBSCRIPTION)
                 .setBody(Unsub.newBuilder()
                     .setDestination(clientIdentity.getDestination())
@@ -228,7 +245,7 @@ public class SimpleCanalConnector implements CanalConnector {
                 .build()
                 .toByteArray());
             //
-            Packet p = Packet.parseFrom(readNextPacket(channel));
+            Packet p = Packet.parseFrom(readNextPacket());
             Ack ack = Ack.parseFrom(p.getBody());
             if (ack.getErrorCode() > 0) {
                 throw new CanalClientException("failed to unSubscribe with reason: " + ack.getErrorMessage());
@@ -261,8 +278,7 @@ public class SimpleCanalConnector implements CanalConnector {
                 unit = TimeUnit.MILLISECONDS;
             }
 
-            writeWithHeader(channel,
+            writeWithHeader(Packet.newBuilder()
-                Packet.newBuilder()
                 .setType(PacketType.GET)
                 .setBody(Get.newBuilder()
                     .setAutoAck(false)
@@ -275,15 +291,14 @@ public class SimpleCanalConnector implements CanalConnector {
                     .toByteString())
                 .build()
                 .toByteArray());
-
             return receiveMessages();
         } catch (IOException e) {
             throw new CanalClientException(e);
         }
     }
 
-    private Message receiveMessages() throws InvalidProtocolBufferException, IOException {
+    private Message receiveMessages() throws IOException {
-        Packet p = Packet.parseFrom(readNextPacket(channel));
+        Packet p = Packet.parseFrom(readNextPacket());
         switch (p.getType()) {
             case MESSAGES: {
                 if (!p.getCompression().equals(Compression.NONE)) {
@@ -315,7 +330,7 @@ public class SimpleCanalConnector implements CanalConnector {
             .setBatchId(batchId)
             .build();
         try {
-            writeWithHeader(channel, Packet.newBuilder()
+            writeWithHeader(Packet.newBuilder()
                 .setType(PacketType.CLIENTACK)
                 .setBody(ca.toByteString())
                 .build()
@@ -333,7 +348,7 @@ public class SimpleCanalConnector implements CanalConnector {
             .setBatchId(batchId)
             .build();
         try {
-            writeWithHeader(channel, Packet.newBuilder()
+            writeWithHeader(Packet.newBuilder()
                 .setType(PacketType.CLIENTROLLBACK)
                 .setBody(ca.toByteString())
                 .build()
@@ -350,7 +365,15 @@ public class SimpleCanalConnector implements CanalConnector {
 
     // ==================== helper method ====================
 
-    private void writeWithHeader(SocketChannel channel, byte[] body) throws IOException {
+    private void writeWithHeader(byte[] body) throws IOException {
+        writeWithHeader(writableChannel, body);
+    }
+
+    private byte[] readNextPacket() throws IOException {
+        return readNextPacket(readableChannel);
+    }
+
+    private void writeWithHeader(WritableByteChannel channel, byte[] body) throws IOException {
         synchronized (writeDataLock) {
             writeHeader.clear();
             writeHeader.putInt(body.length);
@@ -360,7 +383,7 @@ public class SimpleCanalConnector implements CanalConnector {
         }
     }
 
-    private byte[] readNextPacket(SocketChannel channel) throws IOException {
+    private byte[] readNextPacket(ReadableByteChannel channel) throws IOException {
         synchronized (readDataLock) {
             readHeader.clear();
             read(channel, readHeader);
@@ -371,7 +394,7 @@ public class SimpleCanalConnector implements CanalConnector {
         }
     }
 
-    private void read(SocketChannel channel, ByteBuffer buffer) throws IOException {
+    private void read(ReadableByteChannel channel, ByteBuffer buffer) throws IOException {
         while (buffer.hasRemaining()) {
             int r = channel.read(buffer);
             if (r == -1) {

common/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.0.24-SNAPSHOT</version>
+		<version>1.0.25-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<artifactId>canal.common</artifactId>
@@ -11,13 +11,18 @@
 	<name>canal common module for otter ${project.version}</name>
 	<url>http://b2b-doc.alibaba-inc.com/display/opentech/Otter</url>
 	<dependencies>
+		<!-- https://mvnrepository.com/artifact/io.netty/netty-all -->
+		<dependency>
+			<groupId>io.netty</groupId>
+			<artifactId>netty-all</artifactId>
+		</dependency>
 		<!-- zk -->
 		<dependency>
 			<groupId>org.apache.zookeeper</groupId>
 			<artifactId>zookeeper</artifactId>
 		</dependency>
 		<dependency>
-			<groupId>com.github.sgroschupf</groupId>
+			<groupId>com.101tec</groupId>
 			<artifactId>zkclient</artifactId>
 		</dependency>
 		<!-- external -->
@@ -28,7 +33,6 @@
 		<dependency>
 			<groupId>commons-lang</groupId>
 			<artifactId>commons-lang</artifactId>
-			<version>2.6</version>
 		</dependency>
 		<dependency>
 			<groupId>org.springframework</groupId>

common/src/main/java/com/alibaba/otter/canal/common/utils/JsonUtils.java

@@ -88,7 +88,8 @@ public class JsonUtils {
 
         public static InetAddressSerializer instance = new InetAddressSerializer();
 
-        public void write(JSONSerializer serializer, Object object, Object fieldName, Type fieldType)
+        @Override
+        public void write(JSONSerializer serializer, Object object, Object fieldName, Type fieldType, int features)
                                                                                                                    throws IOException {
             if (object == null) {
                 serializer.writeNull();

common/src/main/java/com/alibaba/otter/canal/common/zookeeper/ZooKeeperx.java

@@ -8,20 +8,15 @@ import java.util.List;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
-import org.I0Itec.zkclient.IZkConnection;
+import org.I0Itec.zkclient.ZkConnection;
 import org.I0Itec.zkclient.exception.ZkException;
 import org.apache.commons.lang.StringUtils;
 import org.apache.zookeeper.ClientCnxn;
-import org.apache.zookeeper.CreateMode;
-import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.Watcher;
-import org.apache.zookeeper.ZooDefs.Ids;
 import org.apache.zookeeper.ZooKeeper;
-import org.apache.zookeeper.ZooKeeper.States;
 import org.apache.zookeeper.client.ConnectStringParser;
 import org.apache.zookeeper.client.HostProvider;
 import org.apache.zookeeper.client.StaticHostProvider;
-import org.apache.zookeeper.data.Stat;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.util.ReflectionUtils;
@@ -32,7 +27,7 @@ import org.springframework.util.ReflectionUtils;
  * @author jianghang 2012-7-10 下午02:31:42
  * @version 1.0.0
  */
-public class ZooKeeperx implements IZkConnection {
+public class ZooKeeperx extends ZkConnection {
 
     private static final String SERVER_COMMA            = ";";
     private static final Logger logger                  = LoggerFactory.getLogger(ZooKeeperx.class);
@@ -40,12 +35,12 @@ public class ZooKeeperx implements IZkConnection {
     private static final Field  hostProviderField       = ReflectionUtils.findField(ClientCnxn.class, "hostProvider");
     private static final Field  serverAddressesField    = ReflectionUtils.findField(StaticHostProvider.class,
                                                             "serverAddresses");
+    private static final Field  zookeeperLockField      = ReflectionUtils.findField(ZkConnection.class,
+                                                            "_zookeeperLock");
+    private static final Field  zookeeperFiled          = ReflectionUtils.findField(ZkConnection.class, "_zk");
     private static final int    DEFAULT_SESSION_TIMEOUT = 90000;
 
-    private ZooKeeper           _zk                     = null;
+    private final List<String>  _serversList;
-    private Lock                _zookeeperLock          = new ReentrantLock();
-
-    private final List<String>  _servers;
     private final int           _sessionTimeOut;
 
     public ZooKeeperx(String zkServers){
@@ -53,109 +48,55 @@ public class ZooKeeperx implements IZkConnection {
     }
 
     public ZooKeeperx(String zkServers, int sessionTimeOut){
-        _servers = Arrays.asList(StringUtils.split(zkServers, SERVER_COMMA));
+        super(zkServers, sessionTimeOut);
+        _serversList = Arrays.asList(StringUtils.split(this.getServers(), SERVER_COMMA));
         _sessionTimeOut = sessionTimeOut;
     }
 
     @Override
     public void connect(Watcher watcher) {
+        ReflectionUtils.makeAccessible(zookeeperLockField);
+        ReflectionUtils.makeAccessible(zookeeperFiled);
+        Lock _zookeeperLock = (ReentrantLock) ReflectionUtils.getField(zookeeperLockField, this);
+        ZooKeeper _zk = (ZooKeeper) ReflectionUtils.getField(zookeeperFiled, this);
+
         _zookeeperLock.lock();
         try {
             if (_zk != null) {
                 throw new IllegalStateException("zk client has already been started");
             }
+            String zkServers = _serversList.get(0);
 
             try {
-                logger.debug("Creating new ZookKeeper instance to connect to " + _servers + ".");
+                logger.debug("Creating new ZookKeeper instance to connect to " + zkServers + ".");
-                _zk = new ZooKeeper(_servers.get(0), _sessionTimeOut, watcher);
+                _zk = new ZooKeeper(zkServers, _sessionTimeOut, watcher);
                 configMutliCluster(_zk);
+                ReflectionUtils.setField(zookeeperFiled, this, _zk);
             } catch (IOException e) {
-                throw new ZkException("Unable to connect to " + _servers, e);
+                throw new ZkException("Unable to connect to " + zkServers, e);
             }
         } finally {
             _zookeeperLock.unlock();
         }
     }
 
-    public void close() throws InterruptedException {
-        _zookeeperLock.lock();
-        try {
-            if (_zk != null) {
-                logger.debug("Closing ZooKeeper connected to " + _servers);
-                _zk.close();
-                _zk = null;
-            }
-        } finally {
-            _zookeeperLock.unlock();
-        }
-    }
-
-    public String create(String path, byte[] data, CreateMode mode) throws KeeperException, InterruptedException {
-        return _zk.create(path, data, Ids.OPEN_ACL_UNSAFE, mode);
-    }
-
-    public void delete(String path) throws InterruptedException, KeeperException {
-        _zk.delete(path, -1);
-    }
-
-    public boolean exists(String path, boolean watch) throws KeeperException, InterruptedException {
-        return _zk.exists(path, watch) != null;
-    }
-
-    public List<String> getChildren(final String path, final boolean watch) throws KeeperException,
-                                                                           InterruptedException {
-        return _zk.getChildren(path, watch);
-    }
-
-    public byte[] readData(String path, Stat stat, boolean watch) throws KeeperException, InterruptedException {
-        return _zk.getData(path, watch, stat);
-    }
-
-    public void writeData(String path, byte[] data) throws KeeperException, InterruptedException {
-        writeData(path, data, -1);
-    }
-
-    public void writeData(String path, byte[] data, int version) throws KeeperException, InterruptedException {
-        _zk.setData(path, data, version);
-    }
-
-    public States getZookeeperState() {
-        return _zk != null ? _zk.getState() : null;
-    }
-
-    public ZooKeeper getZookeeper() {
-        return _zk;
-    }
-
-    public long getCreateTime(String path) throws KeeperException, InterruptedException {
-        Stat stat = _zk.exists(path, false);
-        if (stat != null) {
-            return stat.getCtime();
-        }
-        return -1;
-    }
-
-    public String getServers() {
-        return StringUtils.join(_servers, SERVER_COMMA);
-    }
-
     // ===============================
 
     public void configMutliCluster(ZooKeeper zk) {
-        if (_servers.size() == 1) {
+        if (_serversList.size() == 1) {
             return;
         }
-        String cluster1 = _servers.get(0);
+        String cluster1 = _serversList.get(0);
         try {
-            if (_servers.size() > 1) {
+            if (_serversList.size() > 1) {
                 // 强制的声明accessible
                 ReflectionUtils.makeAccessible(clientCnxnField);
                 ReflectionUtils.makeAccessible(hostProviderField);
                 ReflectionUtils.makeAccessible(serverAddressesField);
 
                 // 添加第二组集群列表
-                for (int i = 1; i < _servers.size(); i++) {
+                for (int i = 1; i < _serversList.size(); i++) {
-                    String cluster = _servers.get(i);
+                    String cluster = _serversList.get(i);
                     // 强制获取zk中的地址信息
                     ClientCnxn cnxn = (ClientCnxn) ReflectionUtils.getField(clientCnxnField, zk);
                     HostProvider hostProvider = (HostProvider) ReflectionUtils.getField(hostProviderField, cnxn);

dbsync/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.0.24-SNAPSHOT</version>
+		<version>1.0.25-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>

dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonConversion.java

@@ -52,23 +52,24 @@ public class JsonConversion {
     public static final int  VALUE_ENTRY_SIZE_SMALL  = (1 + SMALL_OFFSET_SIZE);
     public static final int  VALUE_ENTRY_SIZE_LARGE  = (1 + LARGE_OFFSET_SIZE);
 
-    public static Json_Value parse_value(int type, LogBuffer buffer, long len) {
+    public static Json_Value parse_value(int type, LogBuffer buffer, long len, String charsetName) {
         buffer = buffer.duplicate(buffer.position(), (int) len);
         switch (type) {
             case JSONB_TYPE_SMALL_OBJECT:
-                return parse_array_or_object(Json_enum_type.OBJECT, buffer, len, false);
+                return parse_array_or_object(Json_enum_type.OBJECT, buffer, len, false, charsetName);
             case JSONB_TYPE_LARGE_OBJECT:
-                return parse_array_or_object(Json_enum_type.OBJECT, buffer, len, true);
+                return parse_array_or_object(Json_enum_type.OBJECT, buffer, len, true, charsetName);
             case JSONB_TYPE_SMALL_ARRAY:
-                return parse_array_or_object(Json_enum_type.ARRAY, buffer, len, false);
+                return parse_array_or_object(Json_enum_type.ARRAY, buffer, len, false, charsetName);
             case JSONB_TYPE_LARGE_ARRAY:
-                return parse_array_or_object(Json_enum_type.ARRAY, buffer, len, true);
+                return parse_array_or_object(Json_enum_type.ARRAY, buffer, len, true, charsetName);
             default:
-                return parse_scalar(type, buffer, len);
+                return parse_scalar(type, buffer, len, charsetName);
         }
     }
 
-    private static Json_Value parse_array_or_object(Json_enum_type type, LogBuffer buffer, long len, boolean large) {
+    private static Json_Value parse_array_or_object(Json_enum_type type, LogBuffer buffer, long len, boolean large,
+                                                    String charsetName) {
         long offset_size = large ? LARGE_OFFSET_SIZE : SMALL_OFFSET_SIZE;
         if (len < 2 * offset_size) {
             throw new IllegalArgumentException("illegal json data");
@@ -95,7 +96,7 @@ public class JsonConversion {
         return large ? buffer.getUint32() : buffer.getUint16();
     }
 
-    private static Json_Value parse_scalar(int type, LogBuffer buffer, long len) {
+    private static Json_Value parse_scalar(int type, LogBuffer buffer, long len, String charsetName) {
         switch (type) {
             case JSONB_TYPE_LITERAL:
                 /* purecov: inspected */
@@ -146,12 +147,12 @@ public class JsonConversion {
                     }
                 }
 
-                if (str_len == 0 || len < n + str_len) {
+                if (len < n + str_len) {
                     throw new IllegalArgumentException("illegal json data");
                 }
                 return new Json_Value(Json_enum_type.STRING, buffer.rewind()
                     .forward((int) n)
-                    .getFixString((int) str_len));
+                    .getFixString((int) str_len, charsetName));
             case JSONB_TYPE_OPAQUE:
                 /*
                  * There should always be at least one byte, which tells the
@@ -237,7 +238,7 @@ public class JsonConversion {
             this.m_large = large;
         }
 
-        public String key(int i) {
+        public String key(int i, String charsetName) {
             m_data.rewind();
             int offset_size = m_large ? LARGE_OFFSET_SIZE : SMALL_OFFSET_SIZE;
             int key_entry_size = m_large ? KEY_ENTRY_SIZE_LARGE : KEY_ENTRY_SIZE_SMALL;
@@ -250,10 +251,10 @@ public class JsonConversion {
             // entry, always two
             // bytes.
             long key_length = m_data.getUint16();
-            return m_data.rewind().forward((int) key_offset).getFixString((int) key_length);
+            return m_data.rewind().forward((int) key_offset).getFixString((int) key_length, charsetName);
         }
 
-        public Json_Value element(int i) {
+        public Json_Value element(int i, String charsetName) {
             m_data.rewind();
             int offset_size = m_large ? LARGE_OFFSET_SIZE : SMALL_OFFSET_SIZE;
             int key_entry_size = m_large ? KEY_ENTRY_SIZE_LARGE : KEY_ENTRY_SIZE_SMALL;
@@ -266,13 +267,13 @@ public class JsonConversion {
             int type = m_data.forward(entry_offset).getUint8();
             if (type == JSONB_TYPE_INT16 || type == JSONB_TYPE_UINT16 || type == JSONB_TYPE_LITERAL
                 || (m_large && (type == JSONB_TYPE_INT32 || type == JSONB_TYPE_UINT32))) {
-                return parse_scalar(type, m_data, value_entry_size - 1);
+                return parse_scalar(type, m_data, value_entry_size - 1, charsetName);
             }
             int value_offset = (int) read_offset_or_size(m_data, m_large);
-            return parse_value(type, m_data.rewind().forward(value_offset), (int) m_length - value_offset);
+            return parse_value(type, m_data.rewind().forward(value_offset), (int) m_length - value_offset, charsetName);
         }
 
-        public StringBuilder toJsonString(StringBuilder buf) {
+        public StringBuilder toJsonString(StringBuilder buf, String charsetName) {
             switch (m_type) {
                 case OBJECT:
                     buf.append("{");
@@ -280,9 +281,9 @@ public class JsonConversion {
                         if (i > 0) {
                             buf.append(", ");
                         }
-                        buf.append('"').append(key(i)).append('"');
+                        buf.append('"').append(key(i, charsetName)).append('"');
                         buf.append(": ");
-                        element(i).toJsonString(buf);
+                        element(i, charsetName).toJsonString(buf, charsetName);
                     }
                     buf.append("}");
                     break;
@@ -292,7 +293,7 @@ public class JsonConversion {
                         if (i > 0) {
                             buf.append(", ");
                         }
-                        element(i).toJsonString(buf);
+                        element(i, charsetName).toJsonString(buf, charsetName);
                     }
                     buf.append("]");
                     break;
@@ -361,7 +362,7 @@ public class JsonConversion {
                         }
                         buf.append('"').append(text).append('"');
                     } else {
-                        text = m_data.getFixString((int) m_length);
+                        text = m_data.getFixString((int) m_length, charsetName);
                         buf.append('"').append(text).append('"');
                     }
 

dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogBuffer.java

@@ -168,8 +168,8 @@ public class LogBuffer {
      * @param newLimit The new limit value; must be non-negative and no larger
      * than this buffer's capacity
      * @return This buffer
-     * @throws IllegalArgumentException If the preconditions on <tt>newLimit</tt>
+     * @throws IllegalArgumentException If the preconditions on
-     * do not hold
+     * <tt>newLimit</tt> do not hold
      */
     public final LogBuffer limit(int newLimit) {
         if (origin + newLimit > buffer.length || newLimit < 0) throw new IllegalArgumentException("capacity excceed: "

dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/FormatDescriptionLogEvent.java

@@ -141,7 +141,7 @@ public final class FormatDescriptionLogEvent extends StartLogEventV3 {
         }
     }
 
-    public FormatDescriptionLogEvent(final int binlogVersion,int binlogChecksum){
+    public FormatDescriptionLogEvent(final int binlogVersion, int binlogChecksum){
         this(binlogVersion);
         this.header.checksumAlg = binlogChecksum;
     }

dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/QueryLogEvent.java

@@ -330,15 +330,12 @@ public class QueryLogEvent extends LogEvent {
                                                           * MariaDb type,
                                                           * sec_part of NOW()
                                                           */
-                                                         + 1 + (MAX_DBS_IN_EVENT_MTS * (1 + NAME_LEN)) + 3 + 1 + 16 + 1 + 60/*
+                                                         + 1 + (MAX_DBS_IN_EVENT_MTS * (1 + NAME_LEN)) + 3 + 1 + 32 * 3
-                                                                                                                             * type
+                                                         + 1 + 60/*
-                                                                                                                             * ,
+                                                                  * type ,
-                                                                                                                             * user_len
+                                                                  * user_len ,
-                                                                                                                             * ,
+                                                                  * user ,
-                                                                                                                             * user
+                                                                  * host_len ,
-                                                                                                                             * ,
-                                                                                                                             * host_len
-                                                                                                                             * ,
                                                                   * host
                                                                   */);
     /**
@@ -548,6 +545,22 @@ public class QueryLogEvent extends LogEvent {
     public static final int Q_UPDATED_DB_NAMES                = 12;
 
     public static final int Q_MICROSECONDS                    = 13;
+    /**
+     * A old (unused now) code for Query_log_event status similar to
+     * G_COMMIT_TS.
+     */
+    public static final int Q_COMMIT_TS                       = 14;
+    /**
+     * A code for Query_log_event status, similar to G_COMMIT_TS2.
+     */
+    public static final int Q_COMMIT_TS2                      = 15;
+    /**
+     * The master connection @@session.explicit_defaults_for_timestamp which is
+     * recorded for queries, CREATE and ALTER table that is defined with a
+     * TIMESTAMP column, that are dependent on that feature. For pre-WL6292
+     * master's the associated with this code value is zero.
+     */
+    public static final int Q_EXPLICIT_DEFAULTS_FOR_TIMESTAMP = 16;
 
     /**
      * FROM MariaDB 5.5.34
@@ -632,6 +645,9 @@ public class QueryLogEvent extends LogEvent {
                             mtsAccessedDbNames[i] = buffer.getFixString(length < NAME_LEN ? length : NAME_LEN);
                         }
                         break;
+                    case Q_EXPLICIT_DEFAULTS_FOR_TIMESTAMP:
+                        buffer.forward(1);
+                        break;
                     case Q_HRNOW:
                         // int when_sec_part = buffer.getUint24();
                         buffer.forward(3);

dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogBuffer.java

@@ -952,9 +952,9 @@ public final class RowsLogBuffer {
                 len = buffer.getUint16();
                 buffer.forward(meta - 2);
                 int position = buffer.position();
-                Json_Value jsonValue = JsonConversion.parse_value(buffer.getUint8(), buffer, len - 1);
+                Json_Value jsonValue = JsonConversion.parse_value(buffer.getUint8(), buffer, len - 1, charsetName);
                 StringBuilder builder = new StringBuilder();
-                jsonValue.toJsonString(builder);
+                jsonValue.toJsonString(builder, charsetName);
                 value = builder.toString();
                 buffer.position(position + len);
                 // byte[] binary = new byte[len];

deployer/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.0.24-SNAPSHOT</version>
+		<version>1.0.25-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>

deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java

@@ -149,6 +149,11 @@ public class CanalController {
                                     public void handleNewSession() throws Exception {
                                         initCid(path);
                                     }
+
+                                    @Override
+                                    public void handleSessionEstablishmentError(Throwable error) throws Exception {
+                                        logger.error("failed to connect to zookeeper", error);
+                                    }
                                 });
                             }
                         } finally {
@@ -383,6 +388,11 @@ public class CanalController {
                 public void handleNewSession() throws Exception {
                     initCid(path);
                 }
+
+                @Override
+                public void handleSessionEstablishmentError(Throwable error) throws Exception {
+                    logger.error("failed to connect to zookeeper", error);
+                }
             });
         }
         // 优先启动embeded服务

deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/SpringInstanceConfigMonitor.java

@@ -49,6 +49,8 @@ public class SpringInstanceConfigMonitor extends AbstractCanalLifeCycle implemen
     private ScheduledExecutorService         executor             = Executors.newScheduledThreadPool(1,
                                                                       new NamedThreadFactory("canal-instance-scan"));
 
+    private volatile boolean                 isFirst              = true;
+
     public void start() {
         super.start();
         Assert.notNull(rootConf, "root conf dir is null!");
@@ -58,6 +60,9 @@ public class SpringInstanceConfigMonitor extends AbstractCanalLifeCycle implemen
             public void run() {
                 try {
                     scan();
+                    if (isFirst) {
+                        isFirst = false;
+                    }
                 } catch (Throwable e) {
                     logger.error("scan failed", e);
                 }
@@ -122,13 +127,18 @@ public class SpringInstanceConfigMonitor extends AbstractCanalLifeCycle implemen
 
             if (!actions.containsKey(destination) && instanceConfigs.length > 0) {
                 // 存在合法的instance.properties，并且第一次添加时，进行启动操作
-                notifyStart(instanceDir, destination);
+                notifyStart(instanceDir, destination, instanceConfigs);
             } else if (actions.containsKey(destination)) {
                 // 历史已经启动过
                 if (instanceConfigs.length == 0) { // 如果不存在合法的instance.properties
                     notifyStop(destination);
                 } else {
                     InstanceConfigFiles lastFile = lastFiles.get(destination);
+                    // 历史启动过 所以配置文件信息必然存在
+                    if (!isFirst && CollectionUtils.isEmpty(lastFile.getInstanceFiles())) {
+                        logger.error("[{}] is started, but not found instance file info.", destination);
+                    }
+
                     boolean hasChanged = judgeFileChanged(instanceConfigs, lastFile.getInstanceFiles());
                     // 通知变化
                     if (hasChanged) {
@@ -161,10 +171,19 @@ public class SpringInstanceConfigMonitor extends AbstractCanalLifeCycle implemen
         }
     }
 
-    private void notifyStart(File instanceDir, String destination) {
+    private void notifyStart(File instanceDir, String destination, File[] instanceConfigs) {
         try {
             defaultAction.start(destination);
             actions.put(destination, defaultAction);
+
+            // 启动成功后记录配置文件信息
+            InstanceConfigFiles lastFile = lastFiles.get(destination);
+            List<FileInfo> newFileInfo = new ArrayList<FileInfo>();
+            for (File instanceConfig : instanceConfigs) {
+                newFileInfo.add(new FileInfo(instanceConfig.getName(), instanceConfig.lastModified()));
+            }
+            lastFile.setInstanceFiles(newFileInfo);
+
             logger.info("auto notify start {} successful.", destination);
         } catch (Throwable e) {
             logger.error("scan add found[{}] but start failed", destination, ExceptionUtils.getFullStackTrace(e));

deployer/src/main/resources/logback.xml

@@ -75,9 +75,7 @@
     </logger>
     
 	<root level="WARN">
-		<!--  
+		<!--<appender-ref ref="STDOUT"/>-->
-		<appender-ref ref="STDOUT"/>
-		-->
 		<appender-ref ref="CANAL-ROOT" />
 	</root>
 </configuration>

deployer/src/main/resources/spring/default-instance.xml

@@ -128,14 +128,14 @@
 		<!-- 解析位点记录 -->
 		<property name="logPositionManager">
 			<bean class="com.alibaba.otter.canal.parse.index.FailbackLogPositionManager">
-				<property name="primary">
+				<constructor-arg>
 					<bean class="com.alibaba.otter.canal.parse.index.MemoryLogPositionManager" />
-				</property>
+				</constructor-arg>
-				<property name="failback">
+				<constructor-arg>
 					<bean class="com.alibaba.otter.canal.parse.index.MetaLogPositionManager">
-						<property name="metaManager" ref="metaManager" />
+						<constructor-arg ref="metaManager"/>
 					</bean>
-				</property>
+				</constructor-arg>
 			</bean>
 		</property>
 		

deployer/src/main/resources/spring/file-instance.xml

@@ -113,14 +113,14 @@
 		<!-- 解析位点记录 -->
 		<property name="logPositionManager">
 			<bean class="com.alibaba.otter.canal.parse.index.FailbackLogPositionManager">
-				<property name="primary">
+				<constructor-arg>
 					<bean class="com.alibaba.otter.canal.parse.index.MemoryLogPositionManager" />
-				</property>
+				</constructor-arg>
-				<property name="failback">
+				<constructor-arg>
 					<bean class="com.alibaba.otter.canal.parse.index.MetaLogPositionManager">
-						<property name="metaManager" ref="metaManager" />
+						<constructor-arg ref="metaManager"/>
 					</bean>
-				</property>
+				</constructor-arg>
 			</bean>
 		</property>
 		

deployer/src/main/resources/spring/local-instance.xml

@@ -123,14 +123,14 @@
 		<!-- 解析位点记录 -->
 		<property name="logPositionManager">
 			<bean class="com.alibaba.otter.canal.parse.index.FailbackLogPositionManager">
-				<property name="primary">
+				<constructor-arg>
 					<bean class="com.alibaba.otter.canal.parse.index.MemoryLogPositionManager" />
-				</property>
+				</constructor-arg>
-				<property name="failback">
+				<constructor-arg>
 					<bean class="com.alibaba.otter.canal.parse.index.MetaLogPositionManager">
-						<property name="metaManager" ref="metaManager" />
+						<constructor-arg ref="metaManager"/>
 					</bean>
-				</property>
+				</constructor-arg>
 			</bean>
 		</property>
 		

driver/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.0.24-SNAPSHOT</version>
+		<version>1.0.25-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>
@@ -33,10 +33,6 @@
 			<groupId>org.slf4j</groupId>
 			<artifactId>slf4j-api</artifactId>
 		</dependency>
-		<dependency>
-			<groupId>org.jboss.netty</groupId>
-			<artifactId>netty</artifactId>
-		</dependency>
 		<!-- test dependency -->
 		<dependency>
 			<groupId>junit</groupId>

driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnector.java

@@ -2,8 +2,6 @@ package com.alibaba.otter.canal.parse.driver.mysql;
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
-import java.nio.ByteBuffer;
-import java.nio.channels.SocketChannel;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.commons.lang.exception.ExceptionUtils;
@@ -16,6 +14,8 @@ import com.alibaba.otter.canal.parse.driver.mysql.packets.client.QuitCommandPack
 import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ErrorPacket;
 import com.alibaba.otter.canal.parse.driver.mysql.packets.server.HandshakeInitializationPacket;
 import com.alibaba.otter.canal.parse.driver.mysql.packets.server.Reply323Packet;
+import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannel;
+import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannelPool;
 import com.alibaba.otter.canal.parse.driver.mysql.utils.MySQLPasswordEncrypter;
 import com.alibaba.otter.canal.parse.driver.mysql.utils.PacketManager;
 
@@ -64,10 +64,8 @@ public class MysqlConnector {
     public void connect() throws IOException {
         if (connected.compareAndSet(false, true)) {
             try {
-                channel = SocketChannel.open();
+                channel = SocketChannelPool.open(address);
-                configChannel(channel);
                 logger.info("connect MysqlConnection to {}...", address);
-                channel.connect(address);
                 negotiate(channel);
             } catch (Exception e) {
                 disconnect();
@@ -103,8 +101,7 @@ public class MysqlConnector {
                     MysqlUpdateExecutor executor = new MysqlUpdateExecutor(connector);
                     executor.update("KILL CONNECTION " + connectionId);
                 } catch (Exception e) {
-                    // 忽略具体异常
+                    throw new IOException("KILL DUMP " + connectionId + " failure:" + ExceptionUtils.getStackTrace(e));
-                    logger.info("KILL DUMP " + connectionId + " failure:" + ExceptionUtils.getStackTrace(e));
                 } finally {
                     if (connector != null) {
                         connector.disconnect();
@@ -142,19 +139,7 @@ public class MysqlConnector {
         HeaderPacket quitHeader = new HeaderPacket();
         quitHeader.setPacketBodyLength(cmdBody.length);
         quitHeader.setPacketSequenceNumber((byte) 0x00);
-        PacketManager.write(channel,
+        PacketManager.writePkg(channel, quitHeader.toBytes(), cmdBody);
-            new ByteBuffer[] { ByteBuffer.wrap(quitHeader.toBytes()), ByteBuffer.wrap(cmdBody) });
-    }
-
-    // ====================== help method ====================
-
-    private void configChannel(SocketChannel channel) throws IOException {
-        channel.socket().setKeepAlive(true);
-        channel.socket().setReuseAddress(true);
-        channel.socket().setSoTimeout(soTimeout);
-        channel.socket().setTcpNoDelay(true);
-        channel.socket().setReceiveBufferSize(receiveBufferSize);
-        channel.socket().setSendBufferSize(sendBufferSize);
     }
 
     private void negotiate(SocketChannel channel) throws IOException {
@@ -191,8 +176,7 @@ public class MysqlConnector {
         h.setPacketBodyLength(clientAuthPkgBody.length);
         h.setPacketSequenceNumber((byte) (header.getPacketSequenceNumber() + 1));
 
-        PacketManager.write(channel,
+        PacketManager.writePkg(channel, h.toBytes(), clientAuthPkgBody);
-            new ByteBuffer[] { ByteBuffer.wrap(h.toBytes()), ByteBuffer.wrap(clientAuthPkgBody) });
         logger.info("client authentication packet is sent out.");
 
         // check auth result
@@ -228,7 +212,7 @@ public class MysqlConnector {
         h323.setPacketBodyLength(b323Body.length);
         h323.setPacketSequenceNumber((byte) (packetSequenceNumber + 1));
 
-        PacketManager.write(channel, new ByteBuffer[] { ByteBuffer.wrap(h323.toBytes()), ByteBuffer.wrap(b323Body) });
+        PacketManager.writePkg(channel, h323.toBytes(), b323Body);
         logger.info("client 323 authentication packet is sent out.");
         // check auth result
         HeaderPacket header = PacketManager.readHeader(channel, 4);

driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlQueryExecutor.java

@@ -1,7 +1,6 @@
 package com.alibaba.otter.canal.parse.driver.mysql;
 
 import java.io.IOException;
-import java.nio.channels.SocketChannel;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -12,6 +11,7 @@ import com.alibaba.otter.canal.parse.driver.mysql.packets.server.FieldPacket;
 import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ResultSetHeaderPacket;
 import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ResultSetPacket;
 import com.alibaba.otter.canal.parse.driver.mysql.packets.server.RowDataPacket;
+import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannel;
 import com.alibaba.otter.canal.parse.driver.mysql.utils.PacketManager;
 
 /**
@@ -24,9 +24,9 @@ public class MysqlQueryExecutor {
 
     private SocketChannel channel;
 
-    public MysqlQueryExecutor(MysqlConnector connector){
+    public MysqlQueryExecutor(MysqlConnector connector) throws IOException{
         if (!connector.isConnected()) {
-            throw new RuntimeException("should execute connector.connect() first");
+            throw new IOException("should execute connector.connect() first");
         }
 
         this.channel = connector.getChannel();
@@ -51,7 +51,7 @@ public class MysqlQueryExecutor {
         QueryCommandPacket cmd = new QueryCommandPacket();
         cmd.setQueryString(queryString);
         byte[] bodyBytes = cmd.toBytes();
-        PacketManager.write(channel, bodyBytes);
+        PacketManager.writeBody(channel, bodyBytes);
         byte[] body = readNextPacket();
 
         if (body[0] < 0) {
@@ -88,7 +88,7 @@ public class MysqlQueryExecutor {
         for (RowDataPacket r : rowData) {
             resultSet.getFieldValues().addAll(r.getColumns());
         }
-        resultSet.setSourceAddress(channel.socket().getRemoteSocketAddress());
+        resultSet.setSourceAddress(channel.getRemoteSocketAddress());
 
         return resultSet;
     }

driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlUpdateExecutor.java

@@ -1,7 +1,6 @@
 package com.alibaba.otter.canal.parse.driver.mysql;
 
 import java.io.IOException;
-import java.nio.channels.SocketChannel;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -21,28 +20,29 @@ public class MysqlUpdateExecutor {
 
     private static final Logger logger = LoggerFactory.getLogger(MysqlUpdateExecutor.class);
 
-    private SocketChannel       channel;
+    private MysqlConnector      connector;
 
-    public MysqlUpdateExecutor(MysqlConnector connector){
+    public MysqlUpdateExecutor(MysqlConnector connector) throws IOException{
         if (!connector.isConnected()) {
-            throw new RuntimeException("should execute connector.connect() first");
+            throw new IOException("should execute connector.connect() first");
         }
 
-        this.channel = connector.getChannel();
+        this.connector = connector;
     }
 
-    public MysqlUpdateExecutor(SocketChannel ch){
+    /*
-        this.channel = ch;
+     * public MysqlUpdateExecutor(SocketChannel ch){ this.channel = ch; }
-    }
+     */
 
     public OKPacket update(String updateString) throws IOException {
         QueryCommandPacket cmd = new QueryCommandPacket();
         cmd.setQueryString(updateString);
         byte[] bodyBytes = cmd.toBytes();
-        PacketManager.write(channel, bodyBytes);
+        PacketManager.writeBody(connector.getChannel(), bodyBytes);
 
         logger.debug("read update result...");
-        byte[] body = PacketManager.readBytes(channel, PacketManager.readHeader(channel, 4).getPacketBodyLength());
+        byte[] body = PacketManager.readBytes(connector.getChannel(),
+            PacketManager.readHeader(connector.getChannel(), 4).getPacketBodyLength());
         if (body[0] < 0) {
             ErrorPacket packet = new ErrorPacket();
             packet.fromBytes(body);

driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/server/FieldPacket.java

@@ -74,7 +74,7 @@ public class FieldPacket extends PacketWithHeaderPacket {
         this.decimals = data[index];
         index++;
         //
-        index += 2;//skip filter
+        index += 2;// skip filter
         //
         if (index < data.length) {
             reader.setIndex(index);

driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannel.java

@@ -0,0 +1,82 @@
+package com.alibaba.otter.canal.parse.driver.mysql.socket;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+
+import java.io.IOException;
+import java.net.SocketAddress;
+
+/**
+ * 封装netty的通信channel和数据接收缓存，实现读、写、连接校验的功能。 2016-12-28
+ * 
+ * @author luoyaogui
+ */
+public class SocketChannel {
+
+    private Channel channel = null;
+    private Object  lock    = new Object();
+    private ByteBuf cache   = PooledByteBufAllocator.DEFAULT.directBuffer(1024 * 1024); // 缓存大小
+
+    public Channel getChannel() {
+        return channel;
+    }
+
+    public void setChannel(Channel channel) {
+        this.channel = channel;
+    }
+
+    public void writeCache(ByteBuf buf) {
+        synchronized (lock) {
+            cache.discardReadBytes();// 回收内存
+            cache.writeBytes(buf);
+        }
+    }
+
+    public void writeChannel(byte[]... buf) throws IOException {
+        if (channel != null && channel.isWritable()) channel.writeAndFlush(Unpooled.copiedBuffer(buf));
+        else throw new IOException("write  failed  !  please checking !");
+    }
+
+    public byte[] read(int readSize) throws IOException {
+        do {
+            if (readSize > cache.readableBytes()) {
+                if (null == channel) {
+                    throw new IOException("socket has Interrupted !");
+                }
+                synchronized (this) {
+                    try {
+                        wait(100);
+                    } catch (InterruptedException e) {
+                        throw new IOException("socket has Interrupted !");
+                    }
+                }
+            } else {
+                byte[] back = new byte[readSize];
+                synchronized (lock) {
+                    cache.readBytes(back);
+                }
+                return back;
+            }
+        } while (true);
+    }
+
+    public boolean isConnected() {
+        return channel != null ? true : false;
+    }
+
+    public SocketAddress getRemoteSocketAddress() {
+        return channel != null ? channel.remoteAddress() : null;
+    }
+
+    public void close() {
+        if (channel != null) {
+            channel.close();
+        }
+        channel = null;
+        cache.discardReadBytes();// 回收已占用的内存
+        cache.release();// 释放整个内存
+        cache = null;
+    }
+}

driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannelPool.java

@@ -0,0 +1,100 @@
+package com.alibaba.otter.canal.parse.driver.mysql.socket;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.channel.AdaptiveRecvByteBufAllocator;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.util.ReferenceCountUtil;
+
+import java.io.IOException;
+import java.net.SocketAddress;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * @author luoyaogui 实现channel的管理（监听连接、读数据、回收） 2016-12-28
+ */
+@SuppressWarnings({ "rawtypes", "deprecation" })
+public abstract class SocketChannelPool {
+
+    private static EventLoopGroup              group     = new NioEventLoopGroup();                         // 非阻塞IO线程组
+    private static Bootstrap                   boot      = new Bootstrap();                                 // 主
+    private static Map<Channel, SocketChannel> chManager = new ConcurrentHashMap<Channel, SocketChannel>();
+
+    static {
+        boot.group(group)
+            .channel(NioSocketChannel.class)
+            .option(ChannelOption.SO_RCVBUF, 32 * 1024)
+            .option(ChannelOption.SO_SNDBUF, 32 * 1024)
+            .option(ChannelOption.TCP_NODELAY, true)
+            // 如果是延时敏感型应用，建议关闭Nagle算法
+            .option(ChannelOption.SO_KEEPALIVE, true)
+            .option(ChannelOption.RCVBUF_ALLOCATOR, AdaptiveRecvByteBufAllocator.DEFAULT)
+            .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
+            //
+            .handler(new ChannelInitializer() {
+
+                @Override
+                protected void initChannel(Channel arg0) throws Exception {
+                    arg0.pipeline().addLast(new BusinessHandler());// 命令过滤和handler添加管理
+                }
+            });
+    }
+
+    public static SocketChannel open(SocketAddress address) throws Exception {
+        final SocketChannel socket = new SocketChannel();
+        boot.connect(address).addListener(new ChannelFutureListener() {
+
+            @Override
+            public void operationComplete(ChannelFuture arg0) throws Exception {
+                if (arg0.isSuccess()) socket.setChannel(arg0.channel());
+                synchronized (socket) {
+                    socket.notify();
+                }
+            }
+        });
+        synchronized (socket) {
+            socket.wait();
+        }
+        if (null == socket.getChannel()) {
+            throw new IOException("can't create socket!");
+        }
+        chManager.put(socket.getChannel(), socket);
+        return socket;
+    }
+
+    public static class BusinessHandler extends ChannelInboundHandlerAdapter {
+
+        private SocketChannel socket = null;
+
+        @Override
+        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+            socket.setChannel(null);
+            chManager.remove(ctx.channel());// 移除
+        }
+
+        @Override
+        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+            if (null == socket) socket = chManager.get(ctx.channel());
+            if (socket != null) {
+                socket.writeCache((ByteBuf) msg);
+            }
+            ReferenceCountUtil.release(msg);// 添加防止内存泄漏的
+        }
+
+        @Override
+        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+            ctx.close();
+        }
+    }
+}

driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/PacketManager.java

@@ -1,69 +1,34 @@
 package com.alibaba.otter.canal.parse.driver.mysql.utils;
 
 import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.nio.channels.SocketChannel;
 
 import com.alibaba.otter.canal.parse.driver.mysql.packets.HeaderPacket;
+import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannel;
 
 public abstract class PacketManager {
 
     public static HeaderPacket readHeader(SocketChannel ch, int len) throws IOException {
         HeaderPacket header = new HeaderPacket();
-        header.fromBytes(readBytesAsBuffer(ch, len).array());
+        header.fromBytes(ch.read(len));
         return header;
     }
 
-    public static ByteBuffer readBytesAsBuffer(SocketChannel ch, int len) throws IOException {
-        ByteBuffer buffer = ByteBuffer.allocate(len);
-        while (buffer.hasRemaining()) {
-            int readNum = ch.read(buffer);
-            if (readNum == -1) {
-                throw new IOException("Unexpected End Stream");
-            }
-        }
-        return buffer;
-    }
-
     public static byte[] readBytes(SocketChannel ch, int len) throws IOException {
-        return readBytesAsBuffer(ch, len).array();
+        return ch.read(len);
     }
 
-    /**
+    public static void writePkg(SocketChannel ch, byte[]... srcs) throws IOException {
-     * Since We r using blocking IO, so we will just write once and assert the
+        ch.writeChannel(srcs);
-     * length to simplify the read operation.<br>
-     * If the block write doesn't work as we expected, we will change this
-     * implementation as per the result.
-     * 
-     * @param ch
-     * @param len
-     * @return
-     * @throws IOException
-     */
-    public static void write(SocketChannel ch, ByteBuffer[] srcs) throws IOException {
-        @SuppressWarnings("unused")
-        long total = 0;
-        for (ByteBuffer buffer : srcs) {
-            total += buffer.remaining();
     }
 
-        ch.write(srcs);
+    public static void writeBody(SocketChannel ch, byte[] body) throws IOException {
-        // https://github.com/alibaba/canal/issues/24
+        writeBody0(ch, body, (byte) 0);
-        // 部分windows用户会出现size != total的情况，jdk为java7/openjdk，估计和java版本有关，暂时不做检查
-        // long size = ch.write(srcs);
-        // if (size != total) {
-        // throw new IOException("unexpected blocking io behavior");
-        // }
     }
 
-    public static void write(SocketChannel ch, byte[] body) throws IOException {
+    public static void writeBody0(SocketChannel ch, byte[] body, byte packetSeqNumber) throws IOException {
-        write(ch, body, (byte) 0);
-    }
-
-    public static void write(SocketChannel ch, byte[] body, byte packetSeqNumber) throws IOException {
         HeaderPacket header = new HeaderPacket();
         header.setPacketBodyLength(body.length);
         header.setPacketSequenceNumber(packetSeqNumber);
-        write(ch, new ByteBuffer[] { ByteBuffer.wrap(header.toBytes()), ByteBuffer.wrap(body) });
+        ch.writeChannel(header.toBytes(), body);
     }
 }

driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnectorTest.java

@@ -4,7 +4,6 @@ import java.io.IOException;
 import java.net.InetSocketAddress;
 
 import org.junit.Assert;
-
 import org.junit.Test;
 
 import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ResultSetPacket;

example/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.0.24-SNAPSHOT</version>
+		<version>1.0.25-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>

example/src/main/java/com/alibaba/otter/canal/example/SimpleCanalClientTest.java

@@ -23,8 +23,11 @@ public class SimpleCanalClientTest extends AbstractCanalClientTest {
     public static void main(String args[]) {
         // 根据ip，直接创建链接，无HA的功能
         String destination = "example";
-        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(),
+        String ip = AddressUtils.getHostIp();
-            11111), destination, "", "");
+        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(ip, 11111),
+            destination,
+            "",
+            "");
 
         final SimpleCanalClientTest clientTest = new SimpleCanalClientTest(destination);
         clientTest.setConnector(connector);

filter/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.0.24-SNAPSHOT</version>
+		<version>1.0.25-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>

filter/src/test/java/com/alibaba/otter/canal/filter/AviaterFilterTest.java

@@ -1,7 +1,6 @@
 package com.alibaba.otter.canal.filter;
 
 import org.junit.Assert;
-
 import org.junit.Test;
 
 import com.alibaba.otter.canal.filter.aviater.AviaterELFilter;

filter/src/test/java/com/alibaba/otter/canal/filter/MutliAviaterFilterTest.java

@@ -5,9 +5,8 @@ import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import org.junit.Assert;
-
 import org.apache.commons.lang.RandomStringUtils;
+import org.junit.Assert;
 import org.junit.Test;
 
 import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter;

instance/core/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.0.24-SNAPSHOT</version>
+		<version>1.0.25-SNAPSHOT</version>
 		<relativePath>../../pom.xml</relativePath>
 	</parent>
 	<artifactId>canal.instance.core</artifactId>

instance/core/src/main/java/com/alibaba/otter/canal/instance/core/AbstractCanalInstance.java

@@ -1,5 +1,11 @@
 package com.alibaba.otter.canal.instance.core;
 
+import java.util.List;
+
+import org.apache.commons.lang.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
 import com.alibaba.otter.canal.common.alarm.CanalAlarmHandler;
 import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter;
@@ -16,17 +22,9 @@ import com.alibaba.otter.canal.protocol.ClientIdentity;
 import com.alibaba.otter.canal.sink.CanalEventSink;
 import com.alibaba.otter.canal.store.CanalEventStore;
 import com.alibaba.otter.canal.store.model.Event;
-import org.apache.commons.lang.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.List;
 
 /**
- * Created with Intellig IDEA.
+ * Created with Intellig IDEA. Author: yinxiu Date: 2016-01-07 Time: 22:26
- * Author: yinxiu
- * Date: 2016-01-07
- * Time: 22:26
  */
 public class AbstractCanalInstance extends AbstractCanalLifeCycle implements CanalInstance {
 
@@ -121,7 +119,6 @@ public class AbstractCanalInstance extends AbstractCanalLifeCycle implements Can
             alarmHandler.stop();
         }
 
-
         logger.info("stop successful....");
     }
 

instance/manager/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.0.24-SNAPSHOT</version>
+		<version>1.0.25-SNAPSHOT</version>
 		<relativePath>../../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>

instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java

@@ -346,29 +346,24 @@ public class CanalInstanceWithManager extends AbstractCanalInstance {
     protected CanalLogPositionManager initLogPositionManager() {
         logger.info("init logPositionPersistManager begin...");
         IndexMode indexMode = parameters.getIndexMode();
-        CanalLogPositionManager logPositionManager = null;
+        CanalLogPositionManager logPositionManager;
         if (indexMode.isMemory()) {
             logPositionManager = new MemoryLogPositionManager();
         } else if (indexMode.isZookeeper()) {
-            logPositionManager = new ZooKeeperLogPositionManager();
+            logPositionManager = new ZooKeeperLogPositionManager(getZkclientx());
-            ((ZooKeeperLogPositionManager) logPositionManager).setZkClientx(getZkclientx());
         } else if (indexMode.isMixed()) {
-            logPositionManager = new PeriodMixedLogPositionManager();
+            MemoryLogPositionManager memoryLogPositionManager = new MemoryLogPositionManager();
-
+            ZooKeeperLogPositionManager zooKeeperLogPositionManager = new ZooKeeperLogPositionManager(getZkclientx());
-            ZooKeeperLogPositionManager zooKeeperLogPositionManager = new ZooKeeperLogPositionManager();
+            logPositionManager = new PeriodMixedLogPositionManager(memoryLogPositionManager,
-            zooKeeperLogPositionManager.setZkClientx(getZkclientx());
+                zooKeeperLogPositionManager,
-            ((PeriodMixedLogPositionManager) logPositionManager).setZooKeeperLogPositionManager(zooKeeperLogPositionManager);
+                1000L);
         } else if (indexMode.isMeta()) {
-            logPositionManager = new MetaLogPositionManager();
+            logPositionManager = new MetaLogPositionManager(metaManager);
-            ((MetaLogPositionManager) logPositionManager).setMetaManager(metaManager);
         } else if (indexMode.isMemoryMetaFailback()) {
-            MemoryLogPositionManager primaryLogPositionManager = new MemoryLogPositionManager();
+            MemoryLogPositionManager primary = new MemoryLogPositionManager();
-            MetaLogPositionManager failbackLogPositionManager = new MetaLogPositionManager();
+            MetaLogPositionManager secondary = new MetaLogPositionManager(metaManager);
-            failbackLogPositionManager.setMetaManager(metaManager);
 
-            logPositionManager = new FailbackLogPositionManager();
+            logPositionManager = new FailbackLogPositionManager(primary, secondary);
-            ((FailbackLogPositionManager) logPositionManager).setPrimary(primaryLogPositionManager);
-            ((FailbackLogPositionManager) logPositionManager).setFailback(failbackLogPositionManager);
         } else {
             throw new CanalException("unsupport indexMode for " + indexMode);
         }

instance/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.0.24-SNAPSHOT</version>
+		<version>1.0.25-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>

instance/spring/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.0.24-SNAPSHOT</version>
+		<version>1.0.25-SNAPSHOT</version>
 		<relativePath>../../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>

instance/spring/src/main/java/com/alibaba/otter/canal/instance/spring/CanalInstanceWithSpring.java

@@ -2,11 +2,11 @@ package com.alibaba.otter.canal.instance.spring;
 
 import java.util.List;
 
-import com.alibaba.otter.canal.instance.core.AbstractCanalInstance;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import com.alibaba.otter.canal.common.alarm.CanalAlarmHandler;
+import com.alibaba.otter.canal.instance.core.AbstractCanalInstance;
 import com.alibaba.otter.canal.meta.CanalMetaManager;
 import com.alibaba.otter.canal.parse.CanalEventParser;
 import com.alibaba.otter.canal.protocol.CanalEntry;

instance/spring/src/test/resources/spring/default-instance.xml

@@ -120,14 +120,14 @@
 		<!-- 解析位点记录 -->
 		<property name="logPositionManager">
 			<bean class="com.alibaba.otter.canal.parse.index.FailbackLogPositionManager">
-				<property name="primary">
+				<constructor-arg>
 					<bean class="com.alibaba.otter.canal.parse.index.MemoryLogPositionManager" />
-				</property>
+				</constructor-arg>
-				<property name="failback">
+				<constructor-arg>
 					<bean class="com.alibaba.otter.canal.parse.index.MetaLogPositionManager">
-						<property name="metaManager" ref="metaManager" />
+						<constructor-arg ref="metaManager"/>
 					</bean>
-				</property>
+				</constructor-arg>
 			</bean>
 		</property>
 		

instance/spring/src/test/resources/spring/file-instance.xml

@@ -106,14 +106,16 @@
 		<!-- 解析位点记录 -->
 		<property name="logPositionManager">
 			<bean class="com.alibaba.otter.canal.parse.index.FailbackLogPositionManager">
-				<property name="primary">
+				<constructor-arg>
 					<bean class="com.alibaba.otter.canal.parse.index.MemoryLogPositionManager" />
-				</property>
+				</constructor-arg>
-				<property name="failback">
+				<constructor-arg>
 					<bean class="com.alibaba.otter.canal.parse.index.MetaLogPositionManager">
-						<property name="metaManager" ref="metaManager" />
+						<constructor-arg>
+							<ref bean="metaManager"/>
+						</constructor-arg>
 					</bean>
-				</property>
+				</constructor-arg>
 			</bean>
 		</property>
 		

meta/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.0.24-SNAPSHOT</version>
+		<version>1.0.25-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>

meta/src/test/java/com/alibaba/otter/canal/meta/FileMixedMetaManagerTest.java

@@ -5,9 +5,8 @@ import java.io.IOException;
 import java.util.List;
 import java.util.Map;
 
-import org.junit.Assert;
-
 import org.apache.commons.io.FileUtils;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 

meta/src/test/java/com/alibaba/otter/canal/meta/MemoryMetaManagerTest.java

@@ -3,7 +3,6 @@ package com.alibaba.otter.canal.meta;
 import java.util.Map;
 
 import org.junit.Assert;
-
 import org.junit.Test;
 
 import com.alibaba.otter.canal.protocol.position.PositionRange;

meta/src/test/java/com/alibaba/otter/canal/meta/MixedMetaManagerTest.java

@@ -3,9 +3,8 @@ package com.alibaba.otter.canal.meta;
 import java.util.List;
 import java.util.Map;
 
-import org.junit.Assert;
-
 import org.junit.After;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 

meta/src/test/java/com/alibaba/otter/canal/meta/PeriodMixedMetaManagerTest.java

@@ -3,9 +3,8 @@ package com.alibaba.otter.canal.meta;
 import java.util.List;
 import java.util.Map;
 
-import org.junit.Assert;
-
 import org.junit.After;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 

meta/src/test/java/com/alibaba/otter/canal/meta/ZooKeeperMetaManagerTest.java

@@ -2,9 +2,8 @@ package com.alibaba.otter.canal.meta;
 
 import java.util.Map;
 
-import org.junit.Assert;
-
 import org.junit.After;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 

parse/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.0.24-SNAPSHOT</version>
+		<version>1.0.25-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<artifactId>canal.parse</artifactId>

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java

@@ -191,6 +191,9 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
                                 } catch (TableIdNotFoundException e) {
                                     throw e;
                                 } catch (Throwable e) {
+                                    if (e.getCause() instanceof TableIdNotFoundException) {
+                                        throw (TableIdNotFoundException) e.getCause();
+                                    }
                                     // 记录一下，出错的位点信息
                                     processSinkError(e,
                                         this.lastPosition,

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java

@@ -84,20 +84,17 @@ public class LocalBinLogConnection implements ErosaConnection {
             while (running) {
                 boolean needContinue = true;
                 LogEvent event = null;
-                L:while (fetcher.fetch()) {
+                L: while (fetcher.fetch()) {
-                    /*event = decoder.decode(fetcher, context);
+                    /*
-                    if (event == null) {
+                     * event = decoder.decode(fetcher, context); if (event ==
-                        throw new CanalParseException("parse failed");
+                     * null) { throw new CanalParseException("parse failed"); }
-                    }
+                     * if (!func.sink(event)) { needContinue = false; break; }
-
+                     */
-                    if (!func.sink(event)) {
-                        needContinue = false;
-                        break;
-                    }*/
 
                     do {
                         if (event == null) {
-							event = new RotateLogEvent(context.getLogPosition().getFileName(), context.getLogPosition().getPosition());
+                            event = new RotateLogEvent(context.getLogPosition().getFileName(), context.getLogPosition()
+                                .getPosition());
                         } else {
                             event = decoder.decode(fetcher, context);
                         }

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java

@@ -2,7 +2,6 @@ package com.alibaba.otter.canal.parse.inbound.mysql;
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
-import java.nio.ByteBuffer;
 import java.nio.charset.Charset;
 import java.util.List;
 
@@ -25,8 +24,6 @@ import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.DirectLogFetcher;
 import com.taobao.tddl.dbsync.binlog.LogContext;
 import com.taobao.tddl.dbsync.binlog.LogDecoder;
 import com.taobao.tddl.dbsync.binlog.LogEvent;
-import com.taobao.tddl.dbsync.binlog.LogPosition;
-import com.taobao.tddl.dbsync.binlog.event.FormatDescriptionLogEvent;
 
 public class MysqlConnection implements ErosaConnection {
 
@@ -37,13 +34,11 @@ public class MysqlConnection implements ErosaConnection {
     private Charset             charset = Charset.forName("UTF-8");
     private BinlogFormat        binlogFormat;
     private BinlogImage         binlogImage;
-    private int                 binlogChecksum;
 
     public MysqlConnection(){
     }
 
     public MysqlConnection(InetSocketAddress address, String username, String password){
-
         connector = new MysqlConnector(address, username, password);
     }
 
@@ -93,7 +88,6 @@ public class MysqlConnection implements ErosaConnection {
         decoder.handle(LogEvent.QUERY_EVENT);
         decoder.handle(LogEvent.XID_EVENT);
         LogContext context = new LogContext();
-        context.setLogPosition(new LogPosition(binlogfilename));
         while (fetcher.fetch()) {
             LogEvent event = null;
             event = decoder.decode(fetcher, context);
@@ -110,14 +104,11 @@ public class MysqlConnection implements ErosaConnection {
 
     public void dump(String binlogfilename, Long binlogPosition, SinkFunction func) throws IOException {
         updateSettings();
-        loadBinlogChecksum();
         sendBinlogDump(binlogfilename, binlogPosition);
         DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize());
         fetcher.start(connector.getChannel());
         LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT);
         LogContext context = new LogContext();
-        context.setLogPosition(new LogPosition(binlogfilename));
-        context.setFormatDescription(new FormatDescriptionLogEvent(4, binlogChecksum));
         while (fetcher.fetch()) {
             LogEvent event = null;
             event = decoder.decode(fetcher, context);
@@ -147,9 +138,7 @@ public class MysqlConnection implements ErosaConnection {
         HeaderPacket binlogDumpHeader = new HeaderPacket();
         binlogDumpHeader.setPacketBodyLength(cmdBody.length);
         binlogDumpHeader.setPacketSequenceNumber((byte) 0x00);
-        PacketManager.write(connector.getChannel(), new ByteBuffer[] { ByteBuffer.wrap(binlogDumpHeader.toBytes()),
+        PacketManager.writePkg(connector.getChannel(), binlogDumpHeader.toBytes(), cmdBody);
-                ByteBuffer.wrap(cmdBody) });
-
         connector.setDumping(true);
     }
 
@@ -204,8 +193,15 @@ public class MysqlConnection implements ErosaConnection {
             // 如果不设置会出现错误： Slave can not handle replication events with the
             // checksum that master is configured to log
             // 但也不能乱设置，需要和mysql server的checksum配置一致，不然RotateLogEvent会出现乱码
-            // '@@global.binlog_checksum'需要去掉单引号,在mysql 5.6.29下导致master退出
+            update("set @master_binlog_checksum= '@@global.binlog_checksum'");
-            update("set @master_binlog_checksum= @@global.binlog_checksum");
+        } catch (Exception e) {
+            logger.warn(ExceptionUtils.getFullStackTrace(e));
+        }
+
+        try {
+            // 参考:https://github.com/alibaba/canal/issues/284
+            // mysql5.6需要设置slave_uuid避免被server kill链接
+            update("set @slave_uuid=uuid()");
         } catch (Exception e) {
             if (!StringUtils.contains(e.getMessage(), "Unknown system variable")) {
                 logger.warn(ExceptionUtils.getFullStackTrace(e));
@@ -267,27 +263,6 @@ public class MysqlConnection implements ErosaConnection {
         }
     }
 
-    /**
-     * 获取主库checksum信息
-     * https://dev.mysql.com/doc/refman/5.6/en/replication-options
-     * -binary-log.html#option_mysqld_binlog-checksum
-     */
-    private void loadBinlogChecksum() {
-        ResultSetPacket rs = null;
-        try {
-            rs = query("select @master_binlog_checksum");
-        } catch (IOException e) {
-            throw new CanalParseException(e);
-        }
-
-        List<String> columnValues = rs.getFieldValues();
-        if (columnValues != null && columnValues.size() >= 1 && columnValues.get(0).toUpperCase().equals("CRC32")) {
-            binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_CRC32;
-        } else {
-            binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_OFF;
-        }
-    }
-
     public static enum BinlogFormat {
 
         STATEMENT("STATEMENT"), ROW("ROW"), MIXED("MIXED");

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java

@@ -182,7 +182,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
     protected void stopHeartBeat() {
         TimerTask heartBeatTimerTask = this.heartBeatTimerTask;
         super.stopHeartBeat();
-        if (heartBeatTimerTask != null) {
+        if (heartBeatTimerTask != null && heartBeatTimerTask instanceof MysqlDetectingTimeTask) {
             MysqlConnection mysqlConnection = ((MysqlDetectingTimeTask) heartBeatTimerTask).getMysqlConnection();
             try {
                 mysqlConnection.disconnect();

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DirectLogFetcher.java

@@ -3,13 +3,12 @@ package com.alibaba.otter.canal.parse.inbound.mysql.dbsync;
 import java.io.IOException;
 import java.io.InterruptedIOException;
 import java.net.SocketTimeoutException;
-import java.nio.ByteBuffer;
 import java.nio.channels.ClosedByInterruptException;
-import java.nio.channels.SocketChannel;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannel;
 import com.taobao.tddl.dbsync.binlog.LogFetcher;
 
 /**
@@ -149,20 +148,8 @@ public class DirectLogFetcher extends LogFetcher {
     private final boolean fetch0(final int off, final int len) throws IOException {
         ensureCapacity(off + len);
 
-        ByteBuffer buffer = ByteBuffer.wrap(this.buffer, off, len);
+        byte[] read = channel.read(len);
-        while (buffer.hasRemaining()) {
+        System.arraycopy(read, 0, this.buffer, off, len);
-            int readNum = channel.read(buffer);
-            if (readNum == -1) {
-                throw new IOException("Unexpected End Stream");
-            }
-        }
-
-        // for (int count, n = 0; n < len; n += count) {
-        // if (0 > (count = input.read(buffer, off + n, len - n))) {
-        // // Reached end of input stream
-        // return false;
-        // }
-        // }
 
         if (limit < off + len) limit = off + len;
         return true;

parse/src/main/java/com/alibaba/otter/canal/parse/index/AbstractLogPositionManager.java

@@ -0,0 +1,9 @@
+package com.alibaba.otter.canal.parse.index;
+
+import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
+
+/**
+ * Created by yinxiu on 17/3/17. Email: marklin.hz@gmail.com
+ */
+public abstract class AbstractLogPositionManager extends AbstractCanalLifeCycle implements CanalLogPositionManager {
+}

parse/src/main/java/com/alibaba/otter/canal/parse/index/CanalLogPositionManager.java

@@ -5,14 +5,12 @@ import com.alibaba.otter.canal.parse.exception.CanalParseException;
 import com.alibaba.otter.canal.protocol.position.LogPosition;
 
 /**
- * 接口组合
+ * Created by yinxiu on 17/3/17. Email: marklin.hz@gmail.com
- * 
- * @author jianghang 2012-7-7 上午10:02:02
- * @version 1.0.0
  */
 public interface CanalLogPositionManager extends CanalLifeCycle {
 
     LogPosition getLatestIndexBy(String destination);
 
     void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException;
+
 }

parse/src/main/java/com/alibaba/otter/canal/parse/index/FailbackLogPositionManager.java

@@ -1,74 +1,83 @@
 package com.alibaba.otter.canal.parse.index;
 
-import org.springframework.util.Assert;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
-import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
 import com.alibaba.otter.canal.parse.exception.CanalParseException;
 import com.alibaba.otter.canal.protocol.position.LogPosition;
 
 /**
+ * Created by yinxiu on 17/3/18. Email: marklin.hz@gmail.com
  * 实现基于failover查找的机制完成meta的操作
  *
  * <pre>
  * 应用场景：比如针对内存buffer，出现HA切换，先尝试从内存buffer区中找到lastest position，如果不存在才尝试找一下meta里消费的信息
  * </pre>
- * 
- * @author jianghang 2012-7-20 下午02:33:20
  */
-public class FailbackLogPositionManager extends AbstractCanalLifeCycle implements CanalLogPositionManager {
+public class FailbackLogPositionManager extends AbstractLogPositionManager {
 
-    private CanalLogPositionManager primary;
+    private final static Logger           logger = LoggerFactory.getLogger(FailbackLogPositionManager.class);
-    private CanalLogPositionManager failback;
 
+    private final CanalLogPositionManager primary;
+    private final CanalLogPositionManager secondary;
+
+    public FailbackLogPositionManager(CanalLogPositionManager primary, CanalLogPositionManager secondary){
+        if (primary == null) {
+            throw new NullPointerException("nul primary LogPositionManager");
+        }
+        if (secondary == null) {
+            throw new NullPointerException("nul secondary LogPositionManager");
+        }
+
+        this.primary = primary;
+        this.secondary = secondary;
+    }
+
+    @Override
     public void start() {
         super.start();
-        Assert.notNull(primary);
-        Assert.notNull(failback);
 
         if (!primary.isStart()) {
             primary.start();
         }
 
-        if (!failback.isStart()) {
+        if (!secondary.isStart()) {
-            failback.start();
+            secondary.start();
         }
     }
 
+    @Override
     public void stop() {
         super.stop();
 
+        if (secondary.isStart()) {
+            secondary.stop();
+        }
+
         if (primary.isStart()) {
             primary.stop();
         }
-
-        if (failback.isStart()) {
-            failback.stop();
-        }
     }
 
+    @Override
     public LogPosition getLatestIndexBy(String destination) {
         LogPosition logPosition = primary.getLatestIndexBy(destination);
-        if (logPosition == null) {
+        if (logPosition != null) {
-            return failback.getLatestIndexBy(destination);
-        } else {
             return logPosition;
         }
+        return secondary.getLatestIndexBy(destination);
     }
 
+    @Override
     public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException {
         try {
             primary.persistLogPosition(destination, logPosition);
         } catch (CanalParseException e) {
-            failback.persistLogPosition(destination, logPosition);
+            logger.warn("persistLogPosition use primary log position manager exception. destination: {}, logPosition: {}",
+                destination,
+                logPosition,
+                e);
+            secondary.persistLogPosition(destination, logPosition);
         }
     }
-
-    public void setPrimary(CanalLogPositionManager primary) {
-        this.primary = primary;
-    }
-
-    public void setFailback(CanalLogPositionManager failback) {
-        this.failback = failback;
-    }
-
 }

parse/src/main/java/com/alibaba/otter/canal/parse/index/FileMixedLogPositionManager.java

@@ -16,45 +16,72 @@ import java.util.concurrent.TimeUnit;
 import org.apache.commons.io.FileUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.util.Assert;
 
 import com.alibaba.otter.canal.common.utils.JsonUtils;
 import com.alibaba.otter.canal.meta.exception.CanalMetaManagerException;
+import com.alibaba.otter.canal.parse.exception.CanalParseException;
 import com.alibaba.otter.canal.protocol.position.LogPosition;
 import com.google.common.base.Function;
 import com.google.common.collect.MigrateMap;
 
 /**
- * 基于文件刷新的log position实现
+ * Created by yinxiu on 17/3/18. Email: marklin.hz@gmail.com 基于文件刷新的log
+ * position实现
  *
  * <pre>
  * 策略：
  * 1. 先写内存，然后定时刷新数据到File
  * 2. 数据采取overwrite模式(只保留最后一次)
  * </pre>
- * 
- * @author jianghang 2013-4-15 下午09:40:48
- * @version 1.0.4
  */
-public class FileMixedLogPositionManager extends MemoryLogPositionManager {
+public class FileMixedLogPositionManager extends AbstractLogPositionManager {
+
+    private final static Logger      logger       = LoggerFactory.getLogger(FileMixedLogPositionManager.class);
+    private final static Charset     charset      = Charset.forName("UTF-8");
 
-    private static final Logger      logger       = LoggerFactory.getLogger(FileMixedLogPositionManager.class);
-    private static final Charset     charset      = Charset.forName("UTF-8");
     private File                     dataDir;
-    private String                   dataFileName = "parse.dat";
+
     private Map<String, File>        dataFileCaches;
-    private ScheduledExecutorService executor;
+
-    @SuppressWarnings("serial")
+    private ScheduledExecutorService executorService;
+
     private final LogPosition        nullPosition = new LogPosition() {
                                                   };
 
-    private long                     period       = 1000;                                                      // 单位ms
+    private MemoryLogPositionManager memoryLogPositionManager;
+
+    private long                     period;
     private Set<String>              persistTasks;
 
+    public FileMixedLogPositionManager(File dataDir, long period, MemoryLogPositionManager memoryLogPositionManager){
+        if (dataDir == null) {
+            throw new NullPointerException("null dataDir");
+        }
+        if (period <= 0) {
+            throw new IllegalArgumentException("period must be positive, given: " + period);
+        }
+        if (memoryLogPositionManager == null) {
+            throw new NullPointerException("null memoryLogPositionManager");
+        }
+        this.dataDir = dataDir;
+        this.period = period;
+        this.memoryLogPositionManager = memoryLogPositionManager;
+
+        this.dataFileCaches = MigrateMap.makeComputingMap(new Function<String, File>() {
+
+            public File apply(String destination) {
+                return getDataFile(destination);
+            }
+        });
+
+        this.executorService = Executors.newScheduledThreadPool(1);
+        this.persistTasks = Collections.synchronizedSet(new HashSet<String>());
+    }
+
+    @Override
     public void start() {
         super.start();
 
-        Assert.notNull(dataDir);
         if (!dataDir.exists()) {
             try {
                 FileUtils.forceMkdir(dataDir);
@@ -67,30 +94,12 @@ public class FileMixedLogPositionManager extends MemoryLogPositionManager {
             throw new CanalMetaManagerException("dir[" + dataDir.getPath() + "] can not read/write");
         }
 
-        dataFileCaches = MigrateMap.makeComputingMap(new Function<String, File>() {
+        if (!memoryLogPositionManager.isStart()) {
-
+            memoryLogPositionManager.start();
-            public File apply(String destination) {
-                return getDataFile(destination);
         }
-        });
-
-        executor = Executors.newScheduledThreadPool(1);
-        positions = MigrateMap.makeComputingMap(new Function<String, LogPosition>() {
-
-            public LogPosition apply(String destination) {
-                LogPosition logPosition = loadDataFromFile(dataFileCaches.get(destination));
-                if (logPosition == null) {
-                    return nullPosition;
-                } else {
-                    return logPosition;
-                }
-            }
-        });
-
-        persistTasks = Collections.synchronizedSet(new HashSet<String>());
 
         // 启动定时工作任务
-        executor.scheduleAtFixedRate(new Runnable() {
+        executorService.scheduleAtFixedRate(new Runnable() {
 
             public void run() {
                 List<String> tasks = new ArrayList<String>(persistTasks);
@@ -106,28 +115,35 @@ public class FileMixedLogPositionManager extends MemoryLogPositionManager {
                 }
             }
         }, period, period, TimeUnit.MILLISECONDS);
+
     }
 
+    @Override
     public void stop() {
         super.stop();
 
         flushDataToFile();
-        executor.shutdownNow();
+        executorService.shutdown();
-        positions.clear();
+        memoryLogPositionManager.stop();
-    }
-
-    public void persistLogPosition(String destination, LogPosition logPosition) {
-        persistTasks.add(destination);// 添加到任务队列中进行触发
-        super.persistLogPosition(destination, logPosition);
     }
 
+    @Override
     public LogPosition getLatestIndexBy(String destination) {
-        LogPosition logPostion = super.getLatestIndexBy(destination);
+        LogPosition logPosition = memoryLogPositionManager.getLatestIndexBy(destination);
-        if (logPostion == nullPosition) {
+        if (logPosition != null) {
-            return null;
+            return logPosition;
-        } else {
-            return logPostion;
         }
+        logPosition = loadDataFromFile(dataFileCaches.get(destination));
+        if (logPosition == null) {
+            return nullPosition;
+        }
+        return logPosition;
+    }
+
+    @Override
+    public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException {
+        persistTasks.add(destination);
+        memoryLogPositionManager.persistLogPosition(destination, logPosition);
     }
 
     // ============================ helper method ======================
@@ -142,11 +158,12 @@ public class FileMixedLogPositionManager extends MemoryLogPositionManager {
             }
         }
 
+        String dataFileName = "parse.dat";
         return new File(destinationMetaDir, dataFileName);
     }
 
     private void flushDataToFile() {
-        for (String destination : positions.keySet()) {
+        for (String destination : memoryLogPositionManager.destinations()) {
             flushDataToFile(destination);
         }
     }
@@ -156,7 +173,7 @@ public class FileMixedLogPositionManager extends MemoryLogPositionManager {
     }
 
     private void flushDataToFile(String destination, File dataFile) {
-        LogPosition position = positions.get(destination);
+        LogPosition position = memoryLogPositionManager.getLatestIndexBy(destination);
         if (position != null && position != nullPosition) {
             String json = JsonUtils.marshalToString(position);
             try {
@@ -179,16 +196,4 @@ public class FileMixedLogPositionManager extends MemoryLogPositionManager {
             throw new CanalMetaManagerException(e);
         }
     }
-
-    public void setDataDir(String dataDir) {
-        this.dataDir = new File(dataDir);
-    }
-
-    public void setDataDir(File dataDir) {
-        this.dataDir = dataDir;
-    }
-
-    public void setPeriod(long period) {
-        this.period = period;
-    }
 }

parse/src/main/java/com/alibaba/otter/canal/parse/index/MemoryLogPositionManager.java

@@ -1,39 +1,43 @@
 package com.alibaba.otter.canal.parse.index;
 
 import java.util.Map;
+import java.util.Set;
 
-import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
+import com.alibaba.otter.canal.parse.exception.CanalParseException;
 import com.alibaba.otter.canal.protocol.position.LogPosition;
 import com.google.common.collect.MapMaker;
 
 /**
- * 基于内存的实现
+ * Created by yinxiu on 17/3/17. Email: marklin.hz@gmail.com
- * 
- * @author jianghang 2012-7-7 上午10:17:23
- * @version 1.0.0
  */
-public class MemoryLogPositionManager extends AbstractCanalLifeCycle implements CanalLogPositionManager {
+public class MemoryLogPositionManager extends AbstractLogPositionManager {
 
-    protected Map<String, LogPosition> positions;
+    private Map<String, LogPosition> positions;
 
+    @Override
     public void start() {
         super.start();
-
         positions = new MapMaker().makeMap();
     }
 
+    @Override
     public void stop() {
         super.stop();
-
         positions.clear();
     }
 
+    @Override
     public LogPosition getLatestIndexBy(String destination) {
         return positions.get(destination);
     }
 
-    public void persistLogPosition(String destination, LogPosition logPosition) {
+    @Override
+    public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException {
         positions.put(destination, logPosition);
     }
 
+    public Set<String> destinations() {
+        return positions.keySet();
+    }
+
 }

parse/src/main/java/com/alibaba/otter/canal/parse/index/MetaLogPositionManager.java

@@ -4,52 +4,56 @@ import java.util.List;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 
-import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
 import com.alibaba.otter.canal.meta.CanalMetaManager;
+import com.alibaba.otter.canal.parse.exception.CanalParseException;
 import com.alibaba.otter.canal.protocol.ClientIdentity;
 import com.alibaba.otter.canal.protocol.position.LogPosition;
 import com.alibaba.otter.canal.store.helper.CanalEventUtils;
 
 /**
- * 基于meta信息的实现
+ * Created by yinxiu on 17/3/18. Email: marklin.hz@gmail.com
- * 
- * @author jianghang 2012-7-10 下午05:02:33
- * @version 1.0.0
  */
-public class MetaLogPositionManager extends AbstractCanalLifeCycle implements CanalLogPositionManager {
+public class MetaLogPositionManager extends AbstractLogPositionManager {
 
-    private static final Logger logger = LoggerFactory.getLogger(MetaLogPositionManager.class);
+    private final static Logger    logger = LoggerFactory.getLogger(MetaLogPositionManager.class);
-    private CanalMetaManager    metaManager;
 
-    public void start() {
+    private final CanalMetaManager metaManager;
-        super.start();
+
-        Assert.notNull(metaManager);
+    public MetaLogPositionManager(CanalMetaManager metaManager){
-        if (!metaManager.isStart()) {
+        if (metaManager == null) {
-            metaManager.start();
+            throw new NullPointerException("null metaManager");
-        }
         }
 
+        this.metaManager = metaManager;
+    }
+
+    @Override
     public void stop() {
         super.stop();
+
         if (metaManager.isStart()) {
             metaManager.stop();
         }
     }
 
-    public void persistLogPosition(String destination, LogPosition logPosition) {
+    @Override
-        // do nothing
+    public void start() {
-        logger.info("persist LogPosition:{}", destination, logPosition);
+        super.start();
+
+        if (!metaManager.isStart()) {
+            metaManager.start();
+        }
     }
 
+    @Override
     public LogPosition getLatestIndexBy(String destination) {
-        List<ClientIdentity> clientIdentitys = metaManager.listAllSubscribeInfo(destination);
+        List<ClientIdentity> clientIdentities = metaManager.listAllSubscribeInfo(destination);
         LogPosition result = null;
-        if (!CollectionUtils.isEmpty(clientIdentitys)) {
+        if (!CollectionUtils.isEmpty(clientIdentities)) {
             // 尝试找到一个最小的logPosition
-            for (ClientIdentity clientIdentity : clientIdentitys) {
+            for (ClientIdentity clientIdentity : clientIdentities) {
                 LogPosition position = (LogPosition) metaManager.getCursor(clientIdentity);
                 if (position == null) {
                     continue;
@@ -66,8 +70,9 @@ public class MetaLogPositionManager extends AbstractCanalLifeCycle implements Ca
         return result;
     }
 
-    public void setMetaManager(CanalMetaManager metaManager) {
+    @Override
-        this.metaManager = metaManager;
+    public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException {
+        // do nothing
+        logger.info("persist LogPosition:{}", destination, logPosition);
     }
-
 }

parse/src/main/java/com/alibaba/otter/canal/parse/index/MixedLogPositionManager.java

@@ -5,86 +5,83 @@ import java.util.concurrent.Executors;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.util.Assert;
 
+import com.alibaba.otter.canal.common.zookeeper.ZkClientx;
+import com.alibaba.otter.canal.parse.exception.CanalParseException;
 import com.alibaba.otter.canal.protocol.position.LogPosition;
-import com.google.common.base.Function;
-import com.google.common.collect.MigrateMap;
 
 /**
- * 混合memory + zookeeper的存储模式
+ * Created by yinxiu on 17/3/17. Email: marklin.hz@gmail.com Memory first.
- * 
+ * Asynchronous commit position info to ZK.
- * @author jianghang 2012-7-7 上午10:33:19
- * @version 1.0.0
  */
-public class MixedLogPositionManager extends MemoryLogPositionManager implements CanalLogPositionManager {
+public class MixedLogPositionManager extends AbstractLogPositionManager {
 
-    private static final Logger         logger       = LoggerFactory.getLogger(MixedLogPositionManager.class);
+    private final Logger                      logger = LoggerFactory.getLogger(MixedLogPositionManager.class);
-    private ZooKeeperLogPositionManager zooKeeperLogPositionManager;
-    private ExecutorService             executor;
-    @SuppressWarnings("serial")
-    private final LogPosition           nullPosition = new LogPosition() {
-                                                     };
 
+    private final MemoryLogPositionManager    memoryLogPositionManager;
+    private final ZooKeeperLogPositionManager zooKeeperLogPositionManager;
+
+    private final ExecutorService             executor;
+
+    public MixedLogPositionManager(ZkClientx zkClient){
+        if (zkClient == null) {
+            throw new NullPointerException("null zkClient");
+        }
+
+        this.memoryLogPositionManager = new MemoryLogPositionManager();
+        this.zooKeeperLogPositionManager = new ZooKeeperLogPositionManager(zkClient);
+
+        this.executor = Executors.newFixedThreadPool(1);
+    }
+
+    @Override
     public void start() {
         super.start();
 
-        Assert.notNull(zooKeeperLogPositionManager);
+        if (!memoryLogPositionManager.isStart()) {
+            memoryLogPositionManager.start();
+        }
+
         if (!zooKeeperLogPositionManager.isStart()) {
             zooKeeperLogPositionManager.start();
         }
-        executor = Executors.newFixedThreadPool(1);
-        positions = MigrateMap.makeComputingMap(new Function<String, LogPosition>() {
-
-            public LogPosition apply(String destination) {
-                LogPosition logPosition = zooKeeperLogPositionManager.getLatestIndexBy(destination);
-                if (logPosition == null) {
-                    return nullPosition;
-                } else {
-                    return logPosition;
-                }
-            }
-        });
     }
 
+    @Override
     public void stop() {
         super.stop();
 
-        if (zooKeeperLogPositionManager.isStart()) {
+        executor.shutdown();
         zooKeeperLogPositionManager.stop();
-        }
+        memoryLogPositionManager.stop();
-        executor.shutdownNow();
-        positions.clear();
     }
 
-    public void persistLogPosition(final String destination, final LogPosition logPosition) {
+    @Override
-        super.persistLogPosition(destination, logPosition);
+    public LogPosition getLatestIndexBy(String destination) {
+        LogPosition logPosition = memoryLogPositionManager.getLatestIndexBy(destination);
+        if (logPosition != null) {
+            return logPosition;
+        }
+        logPosition = zooKeeperLogPositionManager.getLatestIndexBy(destination);
+        // 这里保持和重构前的逻辑一致,重新添加到Memory中
+        if (logPosition != null) {
+            memoryLogPositionManager.persistLogPosition(destination, logPosition);
+        }
+        return logPosition;
+    }
+
+    @Override
+    public void persistLogPosition(final String destination, final LogPosition logPosition) throws CanalParseException {
+        memoryLogPositionManager.persistLogPosition(destination, logPosition);
         executor.submit(new Runnable() {
 
             public void run() {
                 try {
                     zooKeeperLogPositionManager.persistLogPosition(destination, logPosition);
                 } catch (Exception e) {
-                    logger.error("ERROR # persist to zookeepr has an error", e);
+                    logger.error("ERROR # persist to zookeeper has an error", e);
                 }
             }
         });
-
     }
-
-    public LogPosition getLatestIndexBy(String destination) {
-        LogPosition logPosition = super.getLatestIndexBy(destination);
-        if (logPosition == nullPosition) {
-            return null;
-        } else {
-            return logPosition;
-        }
-    }
-
-    // ======================== setter / getter ======================
-
-    public void setZooKeeperLogPositionManager(ZooKeeperLogPositionManager zooKeeperLogPositionManager) {
-        this.zooKeeperLogPositionManager = zooKeeperLogPositionManager;
-    }
-
 }

parse/src/main/java/com/alibaba/otter/canal/parse/index/PeriodMixedLogPositionManager.java

@@ -11,54 +11,77 @@ import java.util.concurrent.TimeUnit;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.util.Assert;
 
+import com.alibaba.otter.canal.parse.exception.CanalParseException;
 import com.alibaba.otter.canal.protocol.position.LogPosition;
-import com.google.common.base.Function;
-import com.google.common.collect.MigrateMap;
 
 /**
- * 基于定时刷新的策略的mixed实现
+ * Created by yinxiu on 17/3/18. Email: marklin.hz@gmail.com
- * 
- * @author jianghang 2012-9-12 上午11:18:14
- * @version 1.0.0
  */
-public class PeriodMixedLogPositionManager extends MemoryLogPositionManager implements CanalLogPositionManager {
+public class PeriodMixedLogPositionManager extends AbstractLogPositionManager {
 
     private static final Logger         logger       = LoggerFactory.getLogger(PeriodMixedLogPositionManager.class);
+
+    private MemoryLogPositionManager    memoryLogPositionManager;
     private ZooKeeperLogPositionManager zooKeeperLogPositionManager;
-    private ScheduledExecutorService    executor;
+    private ScheduledExecutorService    executorService;
-    @SuppressWarnings("serial")
+
+    private long                        period;
+    private Set<String>                 persistTasks;
+
     private final LogPosition           nullPosition = new LogPosition() {
                                                      };
 
-    private long                        period       = 1000;                                                        // 单位ms
+    public PeriodMixedLogPositionManager(MemoryLogPositionManager memoryLogPositionManager,
-    private Set<String>                 persistTasks;
+                                         ZooKeeperLogPositionManager zooKeeperLogPositionManager, long period){
+        if (memoryLogPositionManager == null) {
+            throw new NullPointerException("null memoryLogPositionManager");
+        }
 
+        if (zooKeeperLogPositionManager == null) {
+            throw new NullPointerException("null zooKeeperLogPositionManager");
+        }
+
+        if (period <= 0) {
+            throw new IllegalArgumentException("period must be positive, given: " + period);
+        }
+
+        this.memoryLogPositionManager = memoryLogPositionManager;
+        this.zooKeeperLogPositionManager = zooKeeperLogPositionManager;
+        this.period = period;
+        this.persistTasks = Collections.synchronizedSet(new HashSet<String>());
+        this.executorService = Executors.newScheduledThreadPool(1);
+    }
+
+    @Override
+    public void stop() {
+        super.stop();
+
+        if (zooKeeperLogPositionManager.isStart()) {
+            zooKeeperLogPositionManager.stop();
+        }
+
+        if (memoryLogPositionManager.isStart()) {
+            memoryLogPositionManager.stop();
+        }
+
+        executorService.shutdown();
+    }
+
+    @Override
     public void start() {
         super.start();
 
-        Assert.notNull(zooKeeperLogPositionManager);
+        if (!memoryLogPositionManager.isStart()) {
+            memoryLogPositionManager.start();
+        }
+
         if (!zooKeeperLogPositionManager.isStart()) {
             zooKeeperLogPositionManager.start();
         }
-        executor = Executors.newScheduledThreadPool(1);
-        positions = MigrateMap.makeComputingMap(new Function<String, LogPosition>() {
-
-            public LogPosition apply(String destination) {
-                LogPosition logPosition = zooKeeperLogPositionManager.getLatestIndexBy(destination);
-                if (logPosition == null) {
-                    return nullPosition;
-                } else {
-                    return logPosition;
-                }
-            }
-        });
-
-        persistTasks = Collections.synchronizedSet(new HashSet<String>());
 
         // 启动定时工作任务
-        executor.scheduleAtFixedRate(new Runnable() {
+        executorService.scheduleAtFixedRate(new Runnable() {
 
             public void run() {
                 List<String> tasks = new ArrayList<String>(persistTasks);
@@ -76,36 +99,19 @@ public class PeriodMixedLogPositionManager extends MemoryLogPositionManager impl
         }, period, period, TimeUnit.MILLISECONDS);
     }
 
-    public void stop() {
+    @Override
-        super.stop();
-
-        if (zooKeeperLogPositionManager.isStart()) {
-            zooKeeperLogPositionManager.stop();
-        }
-        executor.shutdownNow();
-        positions.clear();
-    }
-
-    public void persistLogPosition(String destination, LogPosition logPosition) {
-        persistTasks.add(destination);// 添加到任务队列中进行触发
-        super.persistLogPosition(destination, logPosition);
-    }
-
     public LogPosition getLatestIndexBy(String destination) {
-        LogPosition logPostion = super.getLatestIndexBy(destination);
+        LogPosition logPosition = memoryLogPositionManager.getLatestIndexBy(destination);
-        if (logPostion == nullPosition) {
+        if (logPosition == nullPosition) {
             return null;
         } else {
-            return logPostion;
+            return logPosition;
         }
     }
 
-    public void setZooKeeperLogPositionManager(ZooKeeperLogPositionManager zooKeeperLogPositionManager) {
+    @Override
-        this.zooKeeperLogPositionManager = zooKeeperLogPositionManager;
+    public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException {
+        persistTasks.add(destination);
+        memoryLogPositionManager.persistLogPosition(destination, logPosition);
     }
-
-    public void setPeriod(long period) {
-        this.period = period;
-    }
-
 }

parse/src/main/java/com/alibaba/otter/canal/parse/index/ZooKeeperLogPositionManager.java

@@ -1,33 +1,29 @@
 package com.alibaba.otter.canal.parse.index;
 
 import org.I0Itec.zkclient.exception.ZkNoNodeException;
-import org.springframework.util.Assert;
 
-import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
 import com.alibaba.otter.canal.common.utils.JsonUtils;
 import com.alibaba.otter.canal.common.zookeeper.ZkClientx;
 import com.alibaba.otter.canal.common.zookeeper.ZookeeperPathUtils;
+import com.alibaba.otter.canal.parse.exception.CanalParseException;
 import com.alibaba.otter.canal.protocol.position.LogPosition;
 
 /**
- * 基于zk的实现
+ * Created by yinxiu on 17/3/17. Email: marklin.hz@gmail.com
- * 
- * @author jianghang 2012-7-7 上午10:08:27
- * @version 1.0.0
  */
-public class ZooKeeperLogPositionManager extends AbstractCanalLifeCycle implements CanalLogPositionManager {
+public class ZooKeeperLogPositionManager extends AbstractLogPositionManager {
 
-    private ZkClientx zkClientx;
+    private final ZkClientx zkClientx;
 
-    public void start() {
+    public ZooKeeperLogPositionManager(ZkClientx zkClient){
-        super.start();
+        if (zkClient == null) {
-        Assert.notNull(zkClientx);
+            throw new NullPointerException("null zkClient");
         }
 
-    public void stop() {
+        this.zkClientx = zkClient;
-        super.stop();
     }
 
+    @Override
     public LogPosition getLatestIndexBy(String destination) {
         String path = ZookeeperPathUtils.getParsePath(destination);
         byte[] data = zkClientx.readData(path, true);
@@ -38,7 +34,8 @@ public class ZooKeeperLogPositionManager extends AbstractCanalLifeCycle implemen
         return JsonUtils.unmarshalFromByte(data, LogPosition.class);
     }
 
-    public void persistLogPosition(String destination, LogPosition logPosition) {
+    @Override
+    public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException {
         String path = ZookeeperPathUtils.getParsePath(destination);
         byte[] data = JsonUtils.marshalToByte(logPosition);
         try {
@@ -48,10 +45,4 @@ public class ZooKeeperLogPositionManager extends AbstractCanalLifeCycle implemen
         }
     }
 
-    // ================== setter / getter =================
-
-    public void setZkClientx(ZkClientx zkClientx) {
-        this.zkClientx = zkClientx;
-    }
-
 }

parse/src/test/java/com/alibaba/otter/canal/parse/DirectLogFetcherTest.java

@@ -2,7 +2,6 @@ package com.alibaba.otter.canal.parse;
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
-import java.nio.ByteBuffer;
 
 import org.junit.Assert;
 import org.junit.Test;
@@ -94,7 +93,6 @@ public class DirectLogFetcherTest {
         HeaderPacket binlogDumpHeader = new HeaderPacket();
         binlogDumpHeader.setPacketBodyLength(cmdBody.length);
         binlogDumpHeader.setPacketSequenceNumber((byte) 0x00);
-        PacketManager.write(connector.getChannel(), new ByteBuffer[] { ByteBuffer.wrap(binlogDumpHeader.toBytes()),
+        PacketManager.writePkg(connector.getChannel(), binlogDumpHeader.toBytes(), cmdBody);
-                ByteBuffer.wrap(cmdBody) });
     }
 }

parse/src/test/java/com/alibaba/otter/canal/parse/inbound/EventTransactionBufferTest.java

@@ -6,7 +6,6 @@ import java.util.Date;
 import java.util.List;
 
 import org.junit.Assert;
-
 import org.junit.Test;
 
 import com.alibaba.otter.canal.parse.inbound.EventTransactionBuffer.TransactionFlushCallback;

parse/src/test/java/com/alibaba/otter/canal/parse/inbound/group/GroupEventPaserTest.java

@@ -8,7 +8,7 @@ import com.alibaba.otter.canal.parse.exception.CanalParseException;
 import com.alibaba.otter.canal.parse.inbound.AbstractBinlogParser;
 import com.alibaba.otter.canal.parse.inbound.BinlogParser;
 import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser;
-import com.alibaba.otter.canal.parse.stub.AbstractCanalLogPositionManager;
+import com.alibaba.otter.canal.parse.index.AbstractLogPositionManager;
 import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
 import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
 import com.alibaba.otter.canal.protocol.position.EntryPosition;
@@ -70,15 +70,17 @@ public class GroupEventPaserTest {
         mysqlEventPaser.setMasterPosition(defaultPosition);
         mysqlEventPaser.setBinlogParser(buildParser(buildAuthentication()));
         mysqlEventPaser.setEventSink(new EntryEventSink());
-        mysqlEventPaser.setLogPositionManager(new AbstractCanalLogPositionManager() {
+        mysqlEventPaser.setLogPositionManager(new AbstractLogPositionManager() {
-
-            public void persistLogPosition(String destination, LogPosition logPosition) {
-                // System.out.println(logPosition);
-            }
 
+            @Override
             public LogPosition getLatestIndexBy(String destination) {
                 return null;
             }
+
+            @Override
+            public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException {
+                System.out.println(logPosition);
+            }
         });
         return mysqlEventPaser;
     }

parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogDumpTest.java

@@ -5,11 +5,11 @@ import java.nio.charset.Charset;
 import java.util.List;
 
 import org.junit.Assert;
-
 import org.junit.Test;
 
+import com.alibaba.otter.canal.parse.exception.CanalParseException;
+import com.alibaba.otter.canal.parse.index.AbstractLogPositionManager;
 import com.alibaba.otter.canal.parse.stub.AbstractCanalEventSinkTest;
-import com.alibaba.otter.canal.parse.stub.AbstractCanalLogPositionManager;
 import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
 import com.alibaba.otter.canal.protocol.CanalEntry.Column;
 import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
@@ -81,16 +81,17 @@ public class LocalBinlogDumpTest {
             }
 
         });
-        controller.setLogPositionManager(new AbstractCanalLogPositionManager() {
+        controller.setLogPositionManager(new AbstractLogPositionManager() {
-
-            public void persistLogPosition(String destination, LogPosition logPosition) {
-                System.out.println(logPosition);
-            }
 
             @Override
             public LogPosition getLatestIndexBy(String destination) {
                 return null;
             }
+
+            @Override
+            public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException {
+                System.out.println(logPosition);
+            }
         });
 
         controller.start();

parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParserTest.java

@@ -12,8 +12,8 @@ import org.junit.Before;
 import org.junit.Test;
 
 import com.alibaba.otter.canal.parse.helper.TimeoutChecker;
+import com.alibaba.otter.canal.parse.index.AbstractLogPositionManager;
 import com.alibaba.otter.canal.parse.stub.AbstractCanalEventSinkTest;
-import com.alibaba.otter.canal.parse.stub.AbstractCanalLogPositionManager;
 import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
 import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
 import com.alibaba.otter.canal.protocol.position.EntryPosition;
@@ -70,7 +70,7 @@ public class LocalBinlogEventParserTest {
 
         });
 
-        controller.setLogPositionManager(new AbstractCanalLogPositionManager() {
+        controller.setLogPositionManager(new AbstractLogPositionManager() {
 
             public void persistLogPosition(String destination, LogPosition logPosition) {
                 System.out.println(logPosition);
@@ -133,7 +133,7 @@ public class LocalBinlogEventParserTest {
             }
         });
 
-        controller.setLogPositionManager(new AbstractCanalLogPositionManager() {
+        controller.setLogPositionManager(new AbstractLogPositionManager() {
 
             public void persistLogPosition(String destination, LogPosition logPosition) {
                 System.out.println(logPosition);
@@ -198,7 +198,7 @@ public class LocalBinlogEventParserTest {
             }
         });
 
-        controller.setLogPositionManager(new AbstractCanalLogPositionManager() {
+        controller.setLogPositionManager(new AbstractLogPositionManager() {
 
             public void persistLogPosition(String destination, LogPosition logPosition) {
                 System.out.println(logPosition);

parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlDumpTest.java

@@ -5,11 +5,11 @@ import java.nio.charset.Charset;
 import java.util.List;
 
 import org.junit.Assert;
-
 import org.junit.Test;
 
+import com.alibaba.otter.canal.parse.exception.CanalParseException;
+import com.alibaba.otter.canal.parse.index.AbstractLogPositionManager;
 import com.alibaba.otter.canal.parse.stub.AbstractCanalEventSinkTest;
-import com.alibaba.otter.canal.parse.stub.AbstractCanalLogPositionManager;
 import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
 import com.alibaba.otter.canal.protocol.CanalEntry.Column;
 import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
@@ -84,16 +84,17 @@ public class MysqlDumpTest {
             }
 
         });
-        controller.setLogPositionManager(new AbstractCanalLogPositionManager() {
+        controller.setLogPositionManager(new AbstractLogPositionManager() {
-
-            public void persistLogPosition(String destination, LogPosition logPosition) {
-                System.out.println(logPosition);
-            }
 
             @Override
             public LogPosition getLatestIndexBy(String destination) {
                 return null;
             }
+
+            @Override
+            public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException {
+                System.out.println(logPosition);
+            }
         });
 
         controller.start();

parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParserTest.java

@@ -8,9 +8,10 @@ import java.util.concurrent.atomic.AtomicLong;
 import org.junit.Assert;
 import org.junit.Test;
 
+import com.alibaba.otter.canal.parse.exception.CanalParseException;
 import com.alibaba.otter.canal.parse.helper.TimeoutChecker;
+import com.alibaba.otter.canal.parse.index.AbstractLogPositionManager;
 import com.alibaba.otter.canal.parse.stub.AbstractCanalEventSinkTest;
-import com.alibaba.otter.canal.parse.stub.AbstractCanalLogPositionManager;
 import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
 import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
 import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
@@ -67,16 +68,17 @@ public class MysqlEventParserTest {
             }
         });
 
-        controller.setLogPositionManager(new AbstractCanalLogPositionManager() {
+        controller.setLogPositionManager(new AbstractLogPositionManager() {
-
-            public void persistLogPosition(String destination, LogPosition logPosition) {
-                System.out.println(logPosition);
-            }
 
             @Override
             public LogPosition getLatestIndexBy(String destination) {
                 return null;
             }
+
+            @Override
+            public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException {
+                System.out.println(logPosition);
+            }
         });
 
         controller.start();
@@ -136,7 +138,7 @@ public class MysqlEventParserTest {
             }
         });
 
-        controller.setLogPositionManager(new AbstractCanalLogPositionManager() {
+        controller.setLogPositionManager(new AbstractLogPositionManager() {
 
             public void persistLogPosition(String destination, LogPosition logPosition) {
                 System.out.println(logPosition);
@@ -205,7 +207,7 @@ public class MysqlEventParserTest {
             }
         });
 
-        controller.setLogPositionManager(new AbstractCanalLogPositionManager() {
+        controller.setLogPositionManager(new AbstractLogPositionManager() {
 
             public void persistLogPosition(String destination, LogPosition logPosition) {
                 System.out.println(logPosition);
@@ -281,7 +283,7 @@ public class MysqlEventParserTest {
             }
         });
 
-        controller.setLogPositionManager(new AbstractCanalLogPositionManager() {
+        controller.setLogPositionManager(new AbstractLogPositionManager() {
 
             public void persistLogPosition(String destination, LogPosition logPosition) {
                 System.out.println(logPosition);

parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/SimpleDdlParserTest.java

@@ -164,7 +164,7 @@ public class SimpleDdlParserTest {
         Assert.assertEquals("retl_mark1", result.getOriTableName());
         Assert.assertEquals("retl_mark3", result.getTableName());
 
-        //正则匹配test case
+        // 正则匹配test case
 
         queryString = "rename table totl_mark to totl_mark2";
         result = SimpleDdlParser.parse(queryString, "retl");

parse/src/test/java/com/alibaba/otter/canal/parse/index/FileMixedLogPositionManagerTest.java

@@ -3,9 +3,8 @@ package com.alibaba.otter.canal.parse.index;
 import java.io.File;
 import java.io.IOException;
 
-import org.junit.Assert;
-
 import org.apache.commons.io.FileUtils;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -27,17 +26,19 @@ public class FileMixedLogPositionManagerTest extends AbstractLogPositionManagerT
 
     @Test
     public void testAll() {
-        FileMixedLogPositionManager logPositionManager = new FileMixedLogPositionManager();
+        MemoryLogPositionManager memoryLogPositionManager = new MemoryLogPositionManager();
-        logPositionManager.setDataDir(dataDir);
+
-        logPositionManager.setPeriod(100);
+        FileMixedLogPositionManager logPositionManager = new FileMixedLogPositionManager(dataDir,
+            1000,
+            memoryLogPositionManager);
         logPositionManager.start();
 
         LogPosition position2 = doTest(logPositionManager);
         sleep(1500);
 
-        FileMixedLogPositionManager logPositionManager2 = new FileMixedLogPositionManager();
+        FileMixedLogPositionManager logPositionManager2 = new FileMixedLogPositionManager(dataDir,
-        logPositionManager2.setDataDir(dataDir);
+            1000,
-        logPositionManager2.setPeriod(100);
+            memoryLogPositionManager);
         logPositionManager2.start();
 
         LogPosition getPosition2 = logPositionManager2.getLatestIndexBy(destination);

parse/src/test/java/com/alibaba/otter/canal/parse/index/MetaLogPositionManagerTest.java

@@ -3,9 +3,8 @@ package com.alibaba.otter.canal.parse.index;
 import java.net.InetSocketAddress;
 import java.util.Date;
 
-import org.junit.Assert;
-
 import org.junit.After;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -46,8 +45,7 @@ public class MetaLogPositionManagerTest extends AbstractLogPositionManagerTest {
         metaManager.setZooKeeperMetaManager(zooKeeperMetaManager);
         metaManager.start();
 
-        MetaLogPositionManager logPositionManager = new MetaLogPositionManager();
+        MetaLogPositionManager logPositionManager = new MetaLogPositionManager(metaManager);
-        logPositionManager.setMetaManager(metaManager);
         logPositionManager.start();
         // 构建meta信息
         ClientIdentity client1 = new ClientIdentity(destination, (short) 1);

parse/src/test/java/com/alibaba/otter/canal/parse/index/MixedLogPositionManagerTest.java

@@ -1,8 +1,7 @@
 package com.alibaba.otter.canal.parse.index;
 
-import org.junit.Assert;
-
 import org.junit.After;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -28,19 +27,16 @@ public class MixedLogPositionManagerTest extends AbstractLogPositionManagerTest
 
     @Test
     public void testAll() {
-        MixedLogPositionManager logPositionManager = new MixedLogPositionManager();
+        MemoryLogPositionManager memoryLogPositionManager = new MemoryLogPositionManager();
+        ZooKeeperLogPositionManager zookeeperLogPositionManager = new ZooKeeperLogPositionManager(zkclientx);
 
-        ZooKeeperLogPositionManager zookeeperLogPositionManager = new ZooKeeperLogPositionManager();
+        MixedLogPositionManager logPositionManager = new MixedLogPositionManager(zkclientx);
-        zookeeperLogPositionManager.setZkClientx(zkclientx);
-
-        logPositionManager.setZooKeeperLogPositionManager(zookeeperLogPositionManager);
         logPositionManager.start();
 
         LogPosition position2 = doTest(logPositionManager);
         sleep(1000);
 
-        MixedLogPositionManager logPositionManager2 = new MixedLogPositionManager();
+        MixedLogPositionManager logPositionManager2 = new MixedLogPositionManager(zkclientx);
-        logPositionManager2.setZooKeeperLogPositionManager(zookeeperLogPositionManager);
         logPositionManager2.start();
 
         LogPosition getPosition2 = logPositionManager2.getLatestIndexBy(destination);

parse/src/test/java/com/alibaba/otter/canal/parse/index/PeriodMixedLogPositionManagerTest.java

@@ -1,8 +1,7 @@
 package com.alibaba.otter.canal.parse.index;
 
-import org.junit.Assert;
-
 import org.junit.After;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -28,19 +27,21 @@ public class PeriodMixedLogPositionManagerTest extends AbstractLogPositionManage
 
     @Test
     public void testAll() {
-        PeriodMixedLogPositionManager logPositionManager = new PeriodMixedLogPositionManager();
+        MemoryLogPositionManager memoryLogPositionManager = new MemoryLogPositionManager();
+        ZooKeeperLogPositionManager zookeeperLogPositionManager = new ZooKeeperLogPositionManager(zkclientx);
 
-        ZooKeeperLogPositionManager zookeeperLogPositionManager = new ZooKeeperLogPositionManager();
+        PeriodMixedLogPositionManager logPositionManager = new PeriodMixedLogPositionManager(memoryLogPositionManager,
-        zookeeperLogPositionManager.setZkClientx(zkclientx);
+            zookeeperLogPositionManager,
+            1000L);
 
-        logPositionManager.setZooKeeperLogPositionManager(zookeeperLogPositionManager);
         logPositionManager.start();
 
         LogPosition position2 = doTest(logPositionManager);
         sleep(1500);
 
-        PeriodMixedLogPositionManager logPositionManager2 = new PeriodMixedLogPositionManager();
+        PeriodMixedLogPositionManager logPositionManager2 = new PeriodMixedLogPositionManager(memoryLogPositionManager,
-        logPositionManager2.setZooKeeperLogPositionManager(zookeeperLogPositionManager);
+            zookeeperLogPositionManager,
+            1000L);
         logPositionManager2.start();
 
         LogPosition getPosition2 = logPositionManager2.getLatestIndexBy(destination);

parse/src/test/java/com/alibaba/otter/canal/parse/index/ZooKeeperLogPositionManagerTest.java

@@ -25,8 +25,7 @@ public class ZooKeeperLogPositionManagerTest extends AbstractLogPositionManagerT
 
     @Test
     public void testAll() {
-        ZooKeeperLogPositionManager logPositionManager = new ZooKeeperLogPositionManager();
+        ZooKeeperLogPositionManager logPositionManager = new ZooKeeperLogPositionManager(zkclientx);
-        logPositionManager.setZkClientx(zkclientx);
         logPositionManager.start();
 
         doTest(logPositionManager);

parse/src/test/java/com/alibaba/otter/canal/parse/stub/AbstractCanalLogPositionManager.java

@@ -1,8 +0,0 @@
-package com.alibaba.otter.canal.parse.stub;
-
-import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
-import com.alibaba.otter.canal.parse.index.CanalLogPositionManager;
-
-public abstract class AbstractCanalLogPositionManager extends AbstractCanalLifeCycle implements CanalLogPositionManager {
-
-}

pom.xml

@@ -4,7 +4,7 @@
     <artifactId>canal</artifactId>
     <packaging>pom</packaging>
     <name>canal module for otter ${project.version}</name>
-    <version>1.0.24-SNAPSHOT</version>
+    <version>1.0.25-SNAPSHOT</version>
     <url>https://github.com/alibaba/canal</url>
     <parent>
         <groupId>org.sonatype.oss</groupId>
@@ -170,14 +170,14 @@
                 </exclusions>
             </dependency>
             <dependency>
-                <groupId>com.github.sgroschupf</groupId>
+                <groupId>com.101tec</groupId>
                 <artifactId>zkclient</artifactId>
-                <version>0.1</version>
+                <version>0.10</version>
             </dependency>
             <dependency>
                 <groupId>com.alibaba</groupId>
                 <artifactId>fastjson</artifactId>
-                <version>1.1.35</version>
+                <version>1.2.28</version>
             </dependency>
             <dependency>
                 <groupId>com.google.guava</groupId>
@@ -195,9 +195,9 @@
                 <version>2.0.8</version>
             </dependency>
             <dependency>
-                <groupId>org.jboss.netty</groupId>
+                <groupId>io.netty</groupId>
-                <artifactId>netty</artifactId>
+                <artifactId>netty-all</artifactId>
-                <version>3.2.5.Final</version>
+                <version>4.1.6.Final</version>
             </dependency>
             <dependency>
                 <groupId>com.google.protobuf</groupId>

protocol/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.0.24-SNAPSHOT</version>
+		<version>1.0.25-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>

server/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.0.24-SNAPSHOT</version>
+		<version>1.0.25-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<artifactId>canal.server</artifactId>

server/src/main/java/com/alibaba/otter/canal/server/netty/CanalServerWithNetty.java

@@ -9,6 +9,8 @@ import org.jboss.netty.channel.Channel;
 import org.jboss.netty.channel.ChannelPipeline;
 import org.jboss.netty.channel.ChannelPipelineFactory;
 import org.jboss.netty.channel.Channels;
+import org.jboss.netty.channel.group.ChannelGroup;
+import org.jboss.netty.channel.group.DefaultChannelGroup;
 import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
 
 import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
@@ -32,6 +34,10 @@ public class CanalServerWithNetty extends AbstractCanalLifeCycle implements Cana
     private int                     port;
     private Channel                 serverChannel = null;
     private ServerBootstrap         bootstrap     = null;
+    private ChannelGroup            childGroups   = null; // socket channel
+                                                          // container, used to
+                                                          // close sockets
+                                                          // explicitly.
 
     private static class SingletonHolder {
 
@@ -40,6 +46,7 @@ public class CanalServerWithNetty extends AbstractCanalLifeCycle implements Cana
 
     private CanalServerWithNetty(){
         this.embeddedServer = CanalServerWithEmbedded.instance();
+        this.childGroups = new DefaultChannelGroup();
     }
 
     public static CanalServerWithNetty instance() {
@@ -55,6 +62,17 @@ public class CanalServerWithNetty extends AbstractCanalLifeCycle implements Cana
 
         this.bootstrap = new ServerBootstrap(new NioServerSocketChannelFactory(Executors.newCachedThreadPool(),
             Executors.newCachedThreadPool()));
+        /*
+         * enable keep-alive mechanism, handle abnormal network connection
+         * scenarios on OS level. the threshold parameters are depended on OS.
+         * e.g. On Linux: net.ipv4.tcp_keepalive_time = 300
+         * net.ipv4.tcp_keepalive_probes = 2 net.ipv4.tcp_keepalive_intvl = 30
+         */
+        bootstrap.setOption("child.keepAlive", true);
+        /*
+         * optional parameter.
+         */
+        bootstrap.setOption("child.tcpNoDelay", true);
 
         // 构造对应的pipeline
         bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
@@ -62,7 +80,9 @@ public class CanalServerWithNetty extends AbstractCanalLifeCycle implements Cana
             public ChannelPipeline getPipeline() throws Exception {
                 ChannelPipeline pipelines = Channels.pipeline();
                 pipelines.addLast(FixedHeaderFrameDecoder.class.getName(), new FixedHeaderFrameDecoder());
-                pipelines.addLast(HandshakeInitializationHandler.class.getName(), new HandshakeInitializationHandler());
+                // support to maintain child socket channel.
+                pipelines.addLast(HandshakeInitializationHandler.class.getName(),
+                    new HandshakeInitializationHandler(childGroups));
                 pipelines.addLast(ClientAuthenticationHandler.class.getName(),
                     new ClientAuthenticationHandler(embeddedServer));
 
@@ -87,6 +107,12 @@ public class CanalServerWithNetty extends AbstractCanalLifeCycle implements Cana
             this.serverChannel.close().awaitUninterruptibly(1000);
         }
 
+        // close sockets explicitly to reduce socket channel hung in complicated
+        // network environment.
+        if (this.childGroups != null) {
+            this.childGroups.close().awaitUninterruptibly(5000);
+        }
+
         if (this.bootstrap != null) {
             this.bootstrap.releaseExternalResources();
         }

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

@@ -1,5 +1,7 @@
 package com.alibaba.otter.canal.server.netty.handler;
 
+import java.util.concurrent.TimeUnit;
+
 import org.apache.commons.lang.StringUtils;
 import org.jboss.netty.buffer.ChannelBuffer;
 import org.jboss.netty.channel.ChannelFuture;
@@ -87,10 +89,13 @@ public class ClientAuthenticationHandler extends SimpleChannelHandler {
                         if (clientAuth.getNetWriteTimeout() > 0) {
                             writeTimeout = clientAuth.getNetWriteTimeout();
                         }
+                        // fix bug: soTimeout parameter's unit from connector is
+                        // millseconds.
                         IdleStateHandler idleStateHandler = new IdleStateHandler(NettyUtils.hashedWheelTimer,
                             readTimeout,
                             writeTimeout,
-                            0);
+                            0,
+                            TimeUnit.MILLISECONDS);
                         ctx.getPipeline().addBefore(SessionHandler.class.getName(),
                             IdleStateHandler.class.getName(),
                             idleStateHandler);

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

@@ -3,6 +3,7 @@ package com.alibaba.otter.canal.server.netty.handler;
 import org.jboss.netty.channel.ChannelHandlerContext;
 import org.jboss.netty.channel.ChannelStateEvent;
 import org.jboss.netty.channel.SimpleChannelHandler;
+import org.jboss.netty.channel.group.ChannelGroup;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -19,9 +20,21 @@ import com.alibaba.otter.canal.server.netty.NettyUtils;
  */
 public class HandshakeInitializationHandler extends SimpleChannelHandler {
 
+    // support to maintain socket channel.
+    private ChannelGroup childGroups;
+
+    public HandshakeInitializationHandler(ChannelGroup childGroups){
+        this.childGroups = childGroups;
+    }
+
     private static final Logger logger = LoggerFactory.getLogger(HandshakeInitializationHandler.class);
 
     public void channelOpen(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
+        // add new socket channel in channel container, used to manage sockets.
+        if (childGroups != null) {
+            childGroups.add(ctx.getChannel());
+        }
+
         byte[] body = Packet.newBuilder()
             .setType(CanalPacket.PacketType.HANDSHAKE)
             .setBody(Handshake.newBuilder().build().toByteString())

sink/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.0.24-SNAPSHOT</version>
+		<version>1.0.25-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>

store/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.0.24-SNAPSHOT</version>
+		<version>1.0.25-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>

store/src/test/java/com/alibaba/otter/cancel/store/memory/buffer/MemoryEventStoreMultiThreadTest.java

@@ -7,10 +7,9 @@ import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 
-import org.junit.Assert;
-
 import org.apache.commons.lang.math.RandomUtils;
 import org.junit.After;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 import org.springframework.util.CollectionUtils;