[maven-release-plugin] prepare for next development iteration

issues#244

.gitignore

@@ -13,3 +13,4 @@ otter.properties
 jtester.properties
 .idea/
 *.iml
+.DS_Store

README.md

@@ -9,7 +9,7 @@
 
 <h1>背景</h1>
 <p style="font-size: 14px;">   早期，阿里巴巴B2B公司因为存在杭州和美国双机房部署，存在跨机房同步的业务需求。不过早期的数据库同步业务，主要是基于trigger的方式获取增量变更，不过从2010年开始，阿里系公司开始逐步的尝试基于数据库的日志解析，获取增量变更进行同步，由此衍生出了增量订阅&amp;消费的业务，从此开启了一段新纪元。</p>
-<p style="font-size: 14px;">   ps. 目前内部版本已经支持mysql和oracle部分版本的日志解析，当前的canal开源版本支持5.6及以下的版本(阿里内部mysql 5.6.10, mysql 5.5.18和5.1.40/48)</p>
+<p style="font-size: 14px;">   ps. 目前内部版本已经支持mysql和oracle部分版本的日志解析，当前的canal开源版本支持5.7及以下的版本(阿里内部mysql 5.7.13, 5.6.10, mysql 5.5.18和5.1.40/48)</p>
 <p style="font-size: 14px;"> </p>
 <p style="font-size: 14px;">基于日志增量订阅&amp;消费支持的业务：</p>
 <ol style="font-size: 14px;">
@@ -62,11 +62,16 @@ See the wiki page for : <a href="https://github.com/alibaba/canal/wiki" >wiki文
 <li><a href="https://github.com/alibaba/canal/releases">Download</a></li>
 </ul>
 
-<h1>相关资料: </h1>
+<h1>相关资料</h1>
 
 * ADC阿里技术嘉年华分享ppt (放在google docs上，可能需要翻墙): <a href="https://docs.google.com/presentation/d/1MkszUPYRDkfVPz9IqOT1LLT5d9tuwde_WC8GZvjaDRg/edit?usp=sharing">ppt下载</href>  
 * [与阿里巴巴的RocketMQ配合使用](https://github.com/alibaba/RocketMQ)
 
+<h1>相关开源</h1>
+<ol>
+<li>阿里巴巴分布式数据库同步系统(解决中美异地机房)：<a href="http://github.com/alibaba/otter">http://github.com/alibaba/otter</a></li>
+<li>阿里巴巴去Oracle数据迁移同步工具(目标支持MySQL/DRDS)：<a href="http://github.com/alibaba/yugong">http://github.com/alibaba/yugong</a></li>
+</ol>
 
 <h1>问题反馈</h1>
 <ol>

client/pom.xml

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

client/src/main/java/com/alibaba/otter/canal/client/CanalNodeAccessStrategy.java

@@ -10,5 +10,7 @@ import java.net.SocketAddress;
  */
 public interface CanalNodeAccessStrategy {
 
+    SocketAddress currentNode();
+
     SocketAddress nextNode();
 }

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

@@ -41,11 +41,17 @@ public class ClusterCanalConnector implements CanalConnector {
 
     public void connect() throws CanalClientException {
         while (currentConnector == null) {
-            SocketAddress nextAddress = this.accessStrategy.nextNode();
             int times = 0;
             while (true) {
                 try {
-                    currentConnector = new SimpleCanalConnector(nextAddress, username, password, destination);
+                    currentConnector = new SimpleCanalConnector(null, username, password, destination) {
+
+                        @Override
+                        public SocketAddress getNextAddress() {
+                            return accessStrategy.nextNode();
+                        }
+
+                    };
                     currentConnector.setSoTimeout(soTimeout);
                     if (filter != null) {
                         currentConnector.setFilter(filter);
@@ -57,7 +63,7 @@ public class ClusterCanalConnector implements CanalConnector {
                     currentConnector.connect();
                     break;
                 } catch (Exception e) {
-                    logger.warn("failed to connect to:{} after retry {} times", nextAddress, times);
+                    logger.warn("failed to connect to:{} after retry {} times", accessStrategy.currentNode(), times);
                     currentConnector.disconnect();
                     currentConnector = null;
                     // retry for #retryTimes for each node when trying to

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

@@ -62,6 +62,11 @@ public class ClusterNodeAccessStrategy implements CanalNodeAccessStrategy {
         initRunning(this.zkClient.readData(runningPath, true));
     }
 
+    @Override
+    public SocketAddress currentNode() {
+        return nextNode();
+    }
+
     public SocketAddress nextNode() {
         if (runningAddress != null) {// 如果服务已经启动，直接选择当前正在工作的节点
             return runningAddress;
@@ -109,4 +114,5 @@ public class ClusterNodeAccessStrategy implements CanalNodeAccessStrategy {
         return zkClient;
     }
 
+
 }

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

@@ -125,6 +125,10 @@ public class SimpleCanalConnector implements CanalConnector {
         try {
             channel = SocketChannel.open();
             channel.socket().setSoTimeout(soTimeout);
+            SocketAddress address = getAddress();
+            if (address == null) {
+                address = getNextAddress();
+            }
             channel.connect(address);
             Packet p = Packet.parseFrom(readNextPacket(channel));
             if (p.getVersion() != 1) {
@@ -434,6 +438,10 @@ public class SimpleCanalConnector implements CanalConnector {
         }
     }
 
+    public SocketAddress getNextAddress() {
+        return null;
+    }
+
     public SocketAddress getAddress() {
         return address;
     }

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

@@ -32,4 +32,9 @@ public class SimpleNodeAccessStrategy implements CanalNodeAccessStrategy {
         }
     }
 
+    @Override
+    public SocketAddress currentNode() {
+        return nodes.get(index);
+    }
+
 }

client/src/main/java/com/alibaba/otter/canal/client/impl/running/ClientRunningMonitor.java

@@ -1,6 +1,7 @@
 package com.alibaba.otter.canal.client.impl.running;
 
 import java.net.InetSocketAddress;
+import java.text.MessageFormat;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -20,6 +21,7 @@ import com.alibaba.otter.canal.common.utils.BooleanMutex;
 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.protocol.exception.CanalClientException;
 
 /**
  * clinet running控制
@@ -62,6 +64,8 @@ public class ClientRunningMonitor extends AbstractCanalLifeCycle {
             public void handleDataDeleted(String dataPath) throws Exception {
                 MDC.put("destination", destination);
                 mutex.set(false);
+                // 触发一下退出,可能是人为干预的释放操作或者网络闪断引起的session expired timeout
+                processActiveExit();
                 if (!release && activeData != null && isMine(activeData.getAddress())) {
                     // 如果上一次active的状态就是本机，则即时触发一下active抢占
                     initRunning();
@@ -96,7 +100,11 @@ public class ClientRunningMonitor extends AbstractCanalLifeCycle {
         releaseRunning(); // 尝试一下release
     }
 
-    public void initRunning() {
+    // 改动记录：
+    // 1,在方法上加synchronized关键字,保证同步顺序执行;
+    // 2,判断Zk上已经存在的activeData是否是本机，是的话把mutex重置为true，否则会导致死锁
+    // 3,增加异常处理，保证出现异常时，running节点能被删除,否则会导致死锁
+    public synchronized void initRunning() {
         if (!isStart()) {
             return;
         }
@@ -116,11 +124,22 @@ public class ClientRunningMonitor extends AbstractCanalLifeCycle {
                 initRunning();
             } else {
                 activeData = JsonUtils.unmarshalFromByte(bytes, ClientRunningData.class);
+                // 如果发现已经存在,判断一下是否自己,避免活锁
+                if (activeData.getAddress().contains(":") && isMine(activeData.getAddress())) {
+                    mutex.set(true);
+                }
             }
         } catch (ZkNoNodeException e) {
             zkClient.createPersistent(ZookeeperPathUtils.getClientIdNodePath(this.destination, clientData.getClientId()),
                 true); // 尝试创建父节点
             initRunning();
+        } catch (Throwable t) {
+            logger.error(MessageFormat.format("There is an error when execute initRunning method, with destination [{0}].",
+                destination),
+                t);
+            // 出现任何异常尝试release
+            releaseRunning();
+            throw new CanalClientException("something goes wrong in initRunning method. ", t);
         }
     }
 

common/pom.xml

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

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

@@ -1,12 +1,19 @@
 package com.alibaba.otter.canal.common.utils;
 
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.Inet4Address;
+import java.net.Inet6Address;
+import java.net.InetAddress;
 import java.util.Arrays;
 import java.util.List;
 
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.TypeReference;
 import com.alibaba.fastjson.serializer.JSONSerializer;
+import com.alibaba.fastjson.serializer.ObjectSerializer;
 import com.alibaba.fastjson.serializer.PropertyFilter;
+import com.alibaba.fastjson.serializer.SerializeConfig;
 import com.alibaba.fastjson.serializer.SerializeWriter;
 import com.alibaba.fastjson.serializer.SerializerFeature;
 
@@ -17,6 +24,12 @@ import com.alibaba.fastjson.serializer.SerializerFeature;
  */
 public class JsonUtils {
 
+    static {
+        SerializeConfig.getGlobalInstance().put(InetAddress.class, InetAddressSerializer.instance);
+        SerializeConfig.getGlobalInstance().put(Inet4Address.class, InetAddressSerializer.instance);
+        SerializeConfig.getGlobalInstance().put(Inet6Address.class, InetAddressSerializer.instance);
+    }
+
     public static <T> T unmarshalFromByte(byte[] bytes, Class<T> targetClass) {
         return (T) JSON.parseObject(bytes, targetClass);// 默认为UTF-8
     }
@@ -70,4 +83,21 @@ public class JsonUtils {
             out.close();
         }
     }
+
+    public static class InetAddressSerializer implements ObjectSerializer {
+
+        public static InetAddressSerializer instance = new InetAddressSerializer();
+
+        public void write(JSONSerializer serializer, Object object, Object fieldName, Type fieldType)
+                                                                                                     throws IOException {
+            if (object == null) {
+                serializer.writeNull();
+                return;
+            }
+
+            InetAddress address = (InetAddress) object;
+            // 优先使用name
+            serializer.write(address.getHostName());
+        }
+    }
 }

common/src/main/java/com/alibaba/otter/canal/common/zookeeper/running/ServerRunningMonitor.java

@@ -87,6 +87,10 @@ public class ServerRunningMonitor extends AbstractCanalLifeCycle {
 
     }
 
+    public void init() {
+        processStart();
+    }
+
     public void start() {
         super.start();
         processStart();

dbsync/pom.xml

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

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

@@ -0,0 +1,402 @@
+package com.taobao.tddl.dbsync.binlog;
+
+/**
+ * 处理下MySQL json二进制转化为可读的字符串
+ * 
+ * @author agapple 2016年6月30日 上午11:26:17
+ * @since 1.0.22
+ */
+public class JsonConversion {
+
+    // JSON TYPE
+    public static final int  JSONB_TYPE_SMALL_OBJECT = 0x0;
+    public static final int  JSONB_TYPE_LARGE_OBJECT = 0x1;
+    public static final int  JSONB_TYPE_SMALL_ARRAY  = 0x2;
+    public static final int  JSONB_TYPE_LARGE_ARRAY  = 0x3;
+    public static final int  JSONB_TYPE_LITERAL      = 0x4;
+    public static final int  JSONB_TYPE_INT16        = 0x5;
+    public static final int  JSONB_TYPE_UINT16       = 0x6;
+    public static final int  JSONB_TYPE_INT32        = 0x7;
+    public static final int  JSONB_TYPE_UINT32       = 0x8;
+    public static final int  JSONB_TYPE_INT64        = 0x9;
+    public static final int  JSONB_TYPE_UINT64       = 0xA;
+    public static final int  JSONB_TYPE_DOUBLE       = 0xB;
+    public static final int  JSONB_TYPE_STRING       = 0xC;
+    public static final int  JSONB_TYPE_OPAQUE       = 0xF;
+    public static final char JSONB_NULL_LITERAL      = '\0';
+    public static final char JSONB_TRUE_LITERAL      = '\1';
+    public static final char JSONB_FALSE_LITERAL     = '\2';
+
+    /*
+     * The size of offset or size fields in the small and the large storage
+     * format for JSON objects and JSON arrays.
+     */
+    public static final int  SMALL_OFFSET_SIZE       = 2;
+    public static final int  LARGE_OFFSET_SIZE       = 4;
+
+    /*
+     * The size of key entries for objects when using the small storage format
+     * or the large storage format. In the small format it is 4 bytes (2 bytes
+     * for key length and 2 bytes for key offset). In the large format it is 6
+     * (2 bytes for length, 4 bytes for offset).
+     */
+    public static final int  KEY_ENTRY_SIZE_SMALL    = (2 + SMALL_OFFSET_SIZE);
+    public static final int  KEY_ENTRY_SIZE_LARGE    = (2 + LARGE_OFFSET_SIZE);
+
+    /*
+     * The size of value entries for objects or arrays. When using the small
+     * storage format, the entry size is 3 (1 byte for type, 2 bytes for
+     * offset). When using the large storage format, it is 5 (1 byte for type, 4
+     * bytes for offset).
+     */
+    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) {
+        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);
+            case JSONB_TYPE_LARGE_OBJECT:
+                return parse_array_or_object(Json_enum_type.OBJECT, buffer, len, true);
+            case JSONB_TYPE_SMALL_ARRAY:
+                return parse_array_or_object(Json_enum_type.ARRAY, buffer, len, false);
+            case JSONB_TYPE_LARGE_ARRAY:
+                return parse_array_or_object(Json_enum_type.ARRAY, buffer, len, true);
+            default:
+                return parse_scalar(type, buffer, len);
+        }
+    }
+
+    private static Json_Value parse_array_or_object(Json_enum_type type, LogBuffer buffer, long len, boolean large) {
+        long offset_size = large ? LARGE_OFFSET_SIZE : SMALL_OFFSET_SIZE;
+        if (len < 2 * offset_size) {
+            throw new IllegalArgumentException("illegal json data");
+        }
+        long element_count = read_offset_or_size(buffer, large);
+        long bytes = read_offset_or_size(buffer, large);
+
+        if (bytes > len) {
+            throw new IllegalArgumentException("illegal json data");
+        }
+        long header_size = 2 * offset_size;
+        if (type == Json_enum_type.OBJECT) {
+            header_size += element_count * (large ? KEY_ENTRY_SIZE_LARGE : KEY_ENTRY_SIZE_SMALL);
+        }
+
+        header_size += element_count * (large ? VALUE_ENTRY_SIZE_LARGE : VALUE_ENTRY_SIZE_SMALL);
+        if (header_size > bytes) {
+            throw new IllegalArgumentException("illegal json data");
+        }
+        return new Json_Value(type, buffer.rewind(), element_count, bytes, large);
+    }
+
+    private static long read_offset_or_size(LogBuffer buffer, boolean large) {
+        return large ? buffer.getUint32() : buffer.getUint16();
+    }
+
+    private static Json_Value parse_scalar(int type, LogBuffer buffer, long len) {
+        switch (type) {
+            case JSONB_TYPE_LITERAL:
+                /* purecov: inspected */
+                int data = buffer.getUint8();
+                switch (data) {
+                    case JSONB_NULL_LITERAL:
+                        return new Json_Value(Json_enum_type.LITERAL_NULL);
+                    case JSONB_TRUE_LITERAL:
+                        return new Json_Value(Json_enum_type.LITERAL_TRUE);
+                    case JSONB_FALSE_LITERAL:
+                        return new Json_Value(Json_enum_type.LITERAL_FALSE);
+                    default:
+                        throw new IllegalArgumentException("illegal json data");
+                }
+            case JSONB_TYPE_INT16:
+                return new Json_Value(Json_enum_type.INT, buffer.getInt16());
+            case JSONB_TYPE_INT32:
+                return new Json_Value(Json_enum_type.INT, buffer.getInt32());
+            case JSONB_TYPE_INT64:
+                return new Json_Value(Json_enum_type.INT, buffer.getLong64());
+            case JSONB_TYPE_UINT16:
+                return new Json_Value(Json_enum_type.UINT, buffer.getUint16());
+            case JSONB_TYPE_UINT32:
+                return new Json_Value(Json_enum_type.UINT, buffer.getUint32());
+            case JSONB_TYPE_UINT64:
+                return new Json_Value(Json_enum_type.UINT, buffer.getUlong64());
+            case JSONB_TYPE_DOUBLE:
+                return new Json_Value(Json_enum_type.DOUBLE, Double.valueOf(buffer.getDouble64()));
+            case JSONB_TYPE_STRING:
+                int max_bytes = (int) Math.min(len, 5);
+                long tlen = 0;
+                long str_len = 0;
+                long n = 0;
+                byte[] datas = buffer.getData(max_bytes);
+                for (int i = 0; i < max_bytes; i++) {
+                    // Get the next 7 bits of the length.
+                    tlen |= (datas[i] & 0x7f) << (7 * i);
+                    if ((datas[i] & 0x80) == 0) {
+                        // The length shouldn't exceed 32 bits.
+                        if (tlen > 4294967296L) {
+                            throw new IllegalArgumentException("illegal json data");
+                        }
+
+                        // This was the last byte. Return successfully.
+                        n = i + 1;
+                        str_len = tlen;
+                        break;
+                    }
+                }
+
+                if (str_len == 0 || 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));
+            case JSONB_TYPE_OPAQUE:
+                /*
+                 * There should always be at least one byte, which tells the
+                 * field type of the opaque value.
+                 */
+                // The type is encoded as a uint8 that maps to an
+                // enum_field_types.
+                int type_byte = buffer.getUint8();
+                int position = buffer.position();
+                // Then there's the length of the value.
+                int q_max_bytes = (int) Math.min(len, 5);
+                long q_tlen = 0;
+                long q_str_len = 0;
+                long q_n = 0;
+                byte[] q_datas = buffer.getData(q_max_bytes);
+                for (int i = 0; i < q_max_bytes; i++) {
+                    // Get the next 7 bits of the length.
+                    q_tlen |= (q_datas[i] & 0x7f) << (7 * i);
+                    if ((q_datas[i] & 0x80) == 0) {
+                        // The length shouldn't exceed 32 bits.
+                        if (q_tlen > 4294967296L) {
+                            throw new IllegalArgumentException("illegal json data");
+                        }
+
+                        // This was the last byte. Return successfully.
+                        q_n = i + 1;
+                        q_str_len = q_tlen;
+                        break;
+                    }
+                }
+
+                if (q_str_len == 0 || len < q_n + q_str_len) {
+                    throw new IllegalArgumentException("illegal json data");
+                }
+                return new Json_Value(type_byte, buffer.position(position).forward((int) q_n), q_str_len);
+            default:
+                throw new IllegalArgumentException("illegal json data");
+        }
+    }
+
+    public static class Json_Value {
+
+        Json_enum_type m_type;
+        int            m_field_type;
+        LogBuffer      m_data;
+        long           m_element_count;
+        long           m_length;
+        String         m_string_value;
+        Number         m_int_value;
+        double         m_double_value;
+        boolean        m_large;
+
+        public Json_Value(Json_enum_type t){
+            this.m_type = t;
+        }
+
+        public Json_Value(Json_enum_type t, Number val){
+            this.m_type = t;
+            if (t == Json_enum_type.DOUBLE) {
+                this.m_double_value = val.doubleValue();
+            } else {
+                this.m_int_value = val;
+            }
+        }
+
+        public Json_Value(Json_enum_type t, String value){
+            this.m_type = t;
+            this.m_string_value = value;
+        }
+
+        public Json_Value(int field_type, LogBuffer data, long bytes){
+            this.m_type = Json_enum_type.OPAQUE; // 不确定类型
+            this.m_field_type = field_type;
+            this.m_data = data;
+            this.m_length = bytes;
+        }
+
+        public Json_Value(Json_enum_type t, LogBuffer data, long element_count, long bytes, boolean large){
+            this.m_type = t;
+            this.m_data = data;
+            this.m_element_count = element_count;
+            this.m_length = bytes;
+            this.m_large = large;
+        }
+
+        public String key(int i) {
+            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;
+            int entry_offset = 2 * offset_size + key_entry_size * i;
+            // The offset of the key is the first part of the key
+            // entry.
+            m_data.forward(entry_offset);
+            long key_offset = read_offset_or_size(m_data, m_large);
+            // The length of the key is the second part of the
+            // entry, always two
+            // bytes.
+            long key_length = m_data.getUint16();
+            return m_data.rewind().forward((int) key_offset).getFixString((int) key_length);
+        }
+
+        public Json_Value element(int i) {
+            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;
+            int value_entry_size = m_large ? VALUE_ENTRY_SIZE_LARGE : VALUE_ENTRY_SIZE_SMALL;
+            int first_entry_offset = 2 * offset_size;
+            if (m_type == Json_enum_type.OBJECT) {
+                first_entry_offset += m_element_count * key_entry_size;
+            }
+            int entry_offset = first_entry_offset + value_entry_size * i;
+            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);
+            }
+            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);
+        }
+
+        public StringBuilder toJsonString(StringBuilder buf) {
+            switch (m_type) {
+                case OBJECT:
+                    buf.append("{");
+                    for (int i = 0; i < m_element_count; ++i) {
+                        if (i > 0) {
+                            buf.append(", ");
+                        }
+                        buf.append('"').append(key(i)).append('"');
+                        buf.append(": ");
+                        element(i).toJsonString(buf);
+                    }
+                    buf.append("}");
+                    break;
+                case ARRAY:
+                    buf.append("[");
+                    for (int i = 0; i < m_element_count; ++i) {
+                        if (i > 0) {
+                            buf.append(", ");
+                        }
+                        element(i).toJsonString(buf);
+                    }
+                    buf.append("]");
+                    break;
+                case DOUBLE:
+                    buf.append(Double.valueOf(m_double_value).toString());
+                    break;
+                case INT:
+                    buf.append(m_int_value.toString());
+                    break;
+                case UINT:
+                    buf.append(m_int_value.toString());
+                    break;
+                case LITERAL_FALSE:
+                    buf.append("false");
+                    break;
+                case LITERAL_TRUE:
+                    buf.append("true");
+                    break;
+                case LITERAL_NULL:
+                    buf.append("NULL");
+                    break;
+                case OPAQUE:
+                    String text = null;
+                    if (m_field_type == LogEvent.MYSQL_TYPE_NEWDECIMAL) {
+                        int precision = m_data.getInt8();
+                        int scale = m_data.getInt8();
+                        text = m_data.getDecimal(precision, scale).toPlainString();
+                        buf.append(text);
+                    } else if (m_field_type == LogEvent.MYSQL_TYPE_TIME) {
+                        long packed_value = m_data.getLong64();
+                        if (packed_value == 0) {
+                            text = "00:00:00";
+                        } else {
+                            long ultime = Math.abs(packed_value);
+                            long intpart = ultime >> 24;
+                            int frac = (int) (ultime % (1L << 24));
+                            text = String.format("%s%02d:%02d:%02d",
+                                packed_value >= 0 ? "" : "-",
+                                (int) ((intpart >> 12) % (1 << 10)),
+                                (int) ((intpart >> 6) % (1 << 6)),
+                                (int) (intpart % (1 << 6)));
+                            text = text + "." + usecondsToStr(frac, 6);
+                        }
+                        buf.append('"').append(text).append('"');
+                    } else if (m_field_type == LogEvent.MYSQL_TYPE_DATE || m_field_type == LogEvent.MYSQL_TYPE_DATETIME
+                               || m_field_type == LogEvent.MYSQL_TYPE_TIMESTAMP) {
+                        long packed_value = m_data.getLong64();
+                        if (packed_value == 0) {
+                            text = "0000-00-00 00:00:00";
+                        } else {
+                            // 构造TimeStamp只处理到秒
+                            long ultime = Math.abs(packed_value);
+                            long intpart = ultime >> 24;
+                            int frac = (int) (ultime % (1L << 24));
+                            long ymd = intpart >> 17;
+                            long ym = ymd >> 5;
+                            long hms = intpart % (1 << 17);
+                            text = String.format("%04d-%02d-%02d %02d:%02d:%02d",
+                                (int) (ym / 13),
+                                (int) (ym % 13),
+                                (int) (ymd % (1 << 5)),
+                                (int) (hms >> 12),
+                                (int) ((hms >> 6) % (1 << 6)),
+                                (int) (hms % (1 << 6)));
+                            text = text + "." + usecondsToStr(frac, 6);
+                        }
+                        buf.append('"').append(text).append('"');
+                    } else {
+                        text = m_data.getFixString((int) m_length);
+                        buf.append('"').append(text).append('"');
+                    }
+
+                    break;
+                case STRING:
+                    buf.append('"').append(m_string_value).append('"');
+                    break;
+                case ERROR:
+                    throw new IllegalArgumentException("illegal json data");
+            }
+
+            return buf;
+        }
+    }
+
+    public static enum Json_enum_type {
+        OBJECT, ARRAY, STRING, INT, UINT, DOUBLE, LITERAL_NULL, LITERAL_TRUE, LITERAL_FALSE, OPAQUE, ERROR
+    }
+
+    private static String usecondsToStr(int frac, int meta) {
+        String sec = String.valueOf(frac);
+        if (meta > 6) {
+            throw new IllegalArgumentException("unknow useconds meta : " + meta);
+        }
+
+        if (sec.length() < 6) {
+            StringBuilder result = new StringBuilder(6);
+            int len = 6 - sec.length();
+            for (; len > 0; len--) {
+                result.append('0');
+            }
+            result.append(sec);
+            sec = result.toString();
+        }
+
+        return sec.substring(0, meta);
+    }
+}

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

@@ -267,6 +267,7 @@ public abstract class LogEvent {
     public static final int    MYSQL_TYPE_TIMESTAMP2                    = 17;
     public static final int    MYSQL_TYPE_DATETIME2                     = 18;
     public static final int    MYSQL_TYPE_TIME2                         = 19;
+    public static final int    MYSQL_TYPE_JSON                          = 245;
     public static final int    MYSQL_TYPE_NEWDECIMAL                    = 246;
     public static final int    MYSQL_TYPE_ENUM                          = 247;
     public static final int    MYSQL_TYPE_SET                           = 248;

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

@@ -140,10 +140,15 @@ public final class FormatDescriptionLogEvent extends StartLogEventV3 {
                 throw new IOException("Unknown binlog version: " + binlogVersion);
         }
     }
-
+    
+    public FormatDescriptionLogEvent(final int binlogVersion,int binlogChecksum){
+    	this(binlogVersion);
+    	this.header.checksumAlg = binlogChecksum;
+    }
+    
     public FormatDescriptionLogEvent(final int binlogVersion){
         this.binlogVersion = binlogVersion;
-
+        
         postHeaderLen = new short[ENUM_END_EVENT];
         /* identify binlog format */
         switch (binlogVersion) {

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

@@ -90,6 +90,7 @@ public final class RotateLogEvent extends LogEvent {
         int filenameLen = buffer.limit() - filenameOffset;
         if (filenameLen > FN_REFLEN - 1) filenameLen = FN_REFLEN - 1;
         buffer.position(filenameOffset);
+        
         filename = buffer.getFixString(filenameLen);
     }
 

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

@@ -8,6 +8,8 @@ import java.util.BitSet;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
+import com.taobao.tddl.dbsync.binlog.JsonConversion;
+import com.taobao.tddl.dbsync.binlog.JsonConversion.Json_Value;
 import com.taobao.tddl.dbsync.binlog.LogBuffer;
 import com.taobao.tddl.dbsync.binlog.LogEvent;
 
@@ -25,6 +27,7 @@ public final class RowsLogBuffer {
     public static final long   DATETIMEF_INT_OFS = 0x8000000000L;
     public static final long   TIMEF_INT_OFS     = 0x800000L;
     public static final long   TIMEF_OFS         = 0x800000000000L;
+
     private final LogBuffer    buffer;
     private final int          columnLen;
     private final String       charsetName;
@@ -945,6 +948,22 @@ public final class RowsLogBuffer {
                 length = len;
                 break;
             }
+            case LogEvent.MYSQL_TYPE_JSON: {
+                len = buffer.getUint16();
+                buffer.forward(meta - 2);
+                int position = buffer.position();
+                Json_Value jsonValue = JsonConversion.parse_value(buffer.getUint8(), buffer, len - 1);
+                StringBuilder builder = new StringBuilder();
+                jsonValue.toJsonString(builder);
+                value = builder.toString();
+                buffer.position(position + len);
+                // byte[] binary = new byte[len];
+                // buffer.fillBytes(binary, 0, len);
+                // value = binary;
+                javaType = Types.VARCHAR;
+                length = len;
+                break;
+            }
             case LogEvent.MYSQL_TYPE_GEOMETRY: {
                 /*
                  * MYSQL_TYPE_GEOMETRY: copy from BLOB or TEXT
@@ -1026,4 +1045,5 @@ public final class RowsLogBuffer {
 
         return sec.substring(0, meta);
     }
+
 }

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

@@ -402,6 +402,7 @@ public final class TableMapLogEvent extends LogEvent {
                 case MYSQL_TYPE_DOUBLE:
                 case MYSQL_TYPE_FLOAT:
                 case MYSQL_TYPE_GEOMETRY:
+                case MYSQL_TYPE_JSON:
                     /*
                      * These types store a single byte.
                      */

deployer/pom.xml

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

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

@@ -5,6 +5,7 @@ import java.util.Properties;
 
 import org.I0Itec.zkclient.IZkStateListener;
 import org.I0Itec.zkclient.exception.ZkNoNodeException;
+import org.I0Itec.zkclient.exception.ZkNodeExistsException;
 import org.apache.commons.lang.BooleanUtils;
 import org.apache.commons.lang.StringUtils;
 import org.apache.zookeeper.Watcher.Event.KeeperState;
@@ -88,9 +89,9 @@ public class CanalController {
         cid = Long.valueOf(getProperty(properties, CanalConstants.CANAL_ID));
         ip = getProperty(properties, CanalConstants.CANAL_IP);
         port = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_PORT));
-        embededCanalServer = new CanalServerWithEmbedded();
+        embededCanalServer = CanalServerWithEmbedded.instance();
         embededCanalServer.setCanalInstanceGenerator(instanceGenerator);// 设置自定义的instanceGenerator
-        canalServer = new CanalServerWithNetty(embededCanalServer);
+        canalServer = CanalServerWithNetty.instance();
         canalServer.setIp(ip);
         canalServer.setPort(port);
 
@@ -172,6 +173,8 @@ public class CanalController {
                 if (zkclientx != null) {
                     runningMonitor.setZkClient(zkclientx);
                 }
+                // 触发创建一下cid节点
+                runningMonitor.init();
                 return runningMonitor;
             }
         }));
@@ -189,10 +192,10 @@ public class CanalController {
                         instanceConfigs.put(destination, config);
                     }
 
-                    if (!config.getLazy() && !embededCanalServer.isStart(destination)) {
+                    if (!embededCanalServer.isStart(destination)) {
                         // HA机制启动
                         ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination);
-                        if (!runningMonitor.isStart()) {
+                        if (!config.getLazy() && !runningMonitor.isStart()) {
                             runningMonitor.start();
                         }
                     }
@@ -231,7 +234,13 @@ public class CanalController {
                         if (StringUtils.isEmpty(rootDir)) {
                             rootDir = "../conf";
                         }
-                        monitor.setRootConf(rootDir);
+
+                        if (StringUtils.equals("otter-canal", System.getProperty("appName"))) {
+                            monitor.setRootConf(rootDir);
+                        } else {
+                            // eclipse debug模式
+                            monitor.setRootConf("src/main/resources/");
+                        }
                         return monitor;
                     } else if (mode.isManager()) {
                         return new ManagerInstanceConfigMonitor();
@@ -383,10 +392,10 @@ public class CanalController {
             final String destination = entry.getKey();
             InstanceConfig config = entry.getValue();
             // 创建destination的工作节点
-            if (!config.getLazy() && !embededCanalServer.isStart(destination)) {
+            if (!embededCanalServer.isStart(destination)) {
                 // HA机制启动
                 ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination);
-                if (!runningMonitor.isStart()) {
+                if (!config.getLazy() && !runningMonitor.isStart()) {
                     runningMonitor.start();
                 }
             }
@@ -442,6 +451,9 @@ public class CanalController {
                 String parentDir = path.substring(0, path.lastIndexOf('/'));
                 zkclientx.createPersistent(parentDir, true);
                 zkclientx.createEphemeral(path);
+            } catch (ZkNodeExistsException e) {
+                // ignore
+                // 因为第一次启动时创建了cid,但在stop/start的时可能会关闭和新建,允许出现NodeExists问题s
             }
 
         }

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

@@ -175,6 +175,7 @@ public class SpringInstanceConfigMonitor extends AbstractCanalLifeCycle implemen
         InstanceAction action = actions.remove(destination);
         try {
             action.stop(destination);
+            lastFiles.remove(destination);
             logger.info("auto notify stop {} successful.", destination);
         } catch (Throwable e) {
             logger.error("scan delete found[{}] but stop failed", destination, ExceptionUtils.getFullStackTrace(e));

deployer/src/main/resources/canal.properties

@@ -40,6 +40,7 @@ canal.instance.filter.query.dcl = false
 canal.instance.filter.query.dml = false
 canal.instance.filter.query.ddl = false
 canal.instance.filter.table.error = false
+canal.instance.filter.rows = false
 
 # binlog format/image check
 canal.instance.binlog.format = ROW,STATEMENT,MIXED 
@@ -63,4 +64,4 @@ canal.instance.global.lazy = false
 #canal.instance.global.manager.address = 127.0.0.1:1099
 #canal.instance.global.spring.xml = classpath:spring/memory-instance.xml
 canal.instance.global.spring.xml = classpath:spring/file-instance.xml
-#canal.instance.global.spring.xml = classpath:spring/default-instance.xml
+#canal.instance.global.spring.xml = classpath:spring/default-instance.xml

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

@@ -178,8 +178,9 @@
 		<property name="filterQueryDml" value="${canal.instance.filter.query.dml:false}" />
 		<property name="filterQueryDcl" value="${canal.instance.filter.query.dcl:false}" />
 		<property name="filterQueryDdl" value="${canal.instance.filter.query.ddl:false}" />
+		<property name="filterRows" value="${canal.instance.filter.rows:false}" />
 		<property name="filterTableError" value="${canal.instance.filter.table.error:false}" />
 		<property name="supportBinlogFormats" value="${canal.instance.binlog.format}" />
 		<property name="supportBinlogImages" value="${canal.instance.binlog.image}" />
 	</bean>
-</beans>
+</beans>

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

@@ -163,8 +163,9 @@
 		<property name="filterQueryDml" value="${canal.instance.filter.query.dml:false}" />
 		<property name="filterQueryDcl" value="${canal.instance.filter.query.dcl:false}" />
 		<property name="filterQueryDdl" value="${canal.instance.filter.query.ddl:false}" />
+		<property name="filterRows" value="${canal.instance.filter.rows:false}" />
 		<property name="filterTableError" value="${canal.instance.filter.table.error:false}" />
 		<property name="supportBinlogFormats" value="${canal.instance.binlog.format}" />
 		<property name="supportBinlogImages" value="${canal.instance.binlog.image}" />
 	</bean>
-</beans>
+</beans>

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

@@ -250,8 +250,9 @@
 		<property name="filterQueryDml" value="${canal.instance.filter.query.dml:false}" />
 		<property name="filterQueryDcl" value="${canal.instance.filter.query.dcl:false}" />
 		<property name="filterQueryDdl" value="${canal.instance.filter.query.ddl:false}" />
+		<property name="filterRows" value="${canal.instance.filter.rows:false}" />
 		<property name="filterTableError" value="${canal.instance.filter.table.error:false}" />
 		<property name="supportBinlogFormats" value="${canal.instance.binlog.format}" />
 		<property name="supportBinlogImages" value="${canal.instance.binlog.image}" />
 	</bean>
-</beans>
+</beans>

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

@@ -0,0 +1,163 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:tx="http://www.springframework.org/schema/tx"
+	xmlns:aop="http://www.springframework.org/schema/aop" xmlns:lang="http://www.springframework.org/schema/lang"
+	xmlns:context="http://www.springframework.org/schema/context"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
+           http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-2.0.xsd
+           http://www.springframework.org/schema/lang http://www.springframework.org/schema/lang/spring-lang-2.0.xsd
+           http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.0.xsd
+           http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd"
+	default-autowire="byName">
+
+	<!-- properties -->
+	<bean class="com.alibaba.otter.canal.instance.spring.support.PropertyPlaceholderConfigurer" lazy-init="false">
+		<property name="ignoreResourceNotFound" value="true" />
+		<property name="systemPropertiesModeName" value="SYSTEM_PROPERTIES_MODE_OVERRIDE"/><!-- 允许system覆盖 -->
+		<property name="locationNames">
+			<list>
+				<value>classpath:canal.properties</value>
+				<value>classpath:${canal.instance.destination:}/instance.properties</value>
+			</list>
+		</property>
+	</bean>
+	
+	<bean id="socketAddressEditor" class="com.alibaba.otter.canal.instance.spring.support.SocketAddressEditor" />
+	<bean class="org.springframework.beans.factory.config.CustomEditorConfigurer"> 
+		<property name="propertyEditorRegistrars">
+			<list>
+				<ref bean="socketAddressEditor" />
+			</list>
+		</property>
+	</bean>
+	
+	<bean id="instance" class="com.alibaba.otter.canal.instance.spring.CanalInstanceWithSpring">
+		<property name="destination" value="${canal.instance.destination}" />
+		<property name="eventParser">
+			<ref local="eventParser" />
+		</property>
+		<property name="eventSink">
+			<ref local="eventSink" />
+		</property>
+		<property name="eventStore">
+			<ref local="eventStore" />
+		</property>
+		<property name="metaManager">
+			<ref local="metaManager" />
+		</property>
+		<property name="alarmHandler">
+			<ref local="alarmHandler" />
+		</property>
+		<property name="eventFetcher">
+			<ref local="eventFetcher"/>
+		</property>
+	</bean>
+	
+	<bean id="eventFetcher" class="com.alibaba.otter.canal.mq.fetch.CanalEventFetcherWithMq">
+		<property name="enable" value="${canal.instance.mq.enable}"/>
+		<property name="sendMode" value="${canal.instance.mq.sendMode}"/>
+	</bean>
+	
+	<!-- 报警处理类 -->
+	<bean id="alarmHandler" class="com.alibaba.otter.canal.common.alarm.MsgAlarmHandler">
+		<property name="enabled" value="${canal.alarm.enable}" />
+		<property name="environment" value="${canal.alarm.environment}" />
+		<property name="intercenterApi" value="${canal.alarm.intercenter.api}" />
+		<property name="administrators" value="${canal.alarm.administrators}" />
+	</bean>
+	
+	<bean id="zkClientx" class="org.springframework.beans.factory.config.MethodInvokingFactoryBean" >
+		<property name="targetClass" value="com.alibaba.otter.canal.common.zookeeper.ZkClientx" />
+		<property name="targetMethod" value="getZkClient" />
+		<property name="arguments">
+			<list>
+				<value>${canal.zkServers:127.0.0.1:2181}</value>
+			</list>
+		</property>
+	</bean>
+	
+	<bean id="metaManager" class="com.alibaba.otter.canal.meta.PeriodMixedMetaManager">
+		<property name="zooKeeperMetaManager">
+			<bean class="com.alibaba.otter.canal.meta.ZooKeeperMetaManager">
+				<property name="zkClientx" ref="zkClientx" />
+			</bean>
+		</property>
+		<property name="period" value="${canal.zookeeper.flush.period:1000}" />
+	</bean>
+	
+	<bean id="eventStore" class="com.alibaba.otter.canal.store.memory.MemoryEventStoreWithBuffer">
+		<property name="bufferSize" value="${canal.instance.memory.buffer.size:16384}" />
+		<property name="bufferMemUnit" value="${canal.instance.memory.buffer.memunit:1024}" />
+		<property name="batchMode" value="${canal.instance.memory.batch.mode:MEMSIZE}" />
+		<property name="ddlIsolation" value="${canal.instance.get.ddl.isolation:false}" />
+	</bean>
+	
+	<bean id="eventSink" class="com.alibaba.otter.canal.sink.entry.EntryEventSink">
+		<property name="eventStore" ref="eventStore" />
+	</bean>
+
+	<bean id="eventParser" class="com.alibaba.otter.canal.parse.inbound.mysql.LocalBinlogEventParser">
+		<property name="destination" value="${canal.instance.destination}" />			
+		<property name="alarmHandler" ref="alarmHandler" />
+		
+		<!-- 解析过滤处理 -->
+		<property name="eventFilter">
+			<bean class="com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter" >
+				<constructor-arg index="0" value="${canal.instance.filter.regex:.*\..*}" />
+			</bean>
+		</property>
+		
+		<property name="eventBlackFilter">
+			<bean class="com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter" >
+				<constructor-arg index="0" value="${canal.instance.filter.black.regex:}" />
+				<constructor-arg index="1" value="false" />
+			</bean>
+		</property>
+		
+		<!-- 最大事务解析大小，超过该大小后事务将被切分为多个事务投递 -->
+		<property name="transactionSize" value="${canal.instance.transaction.size:1024}" />
+			
+		<!-- 解析编码 -->
+		<property name="connectionCharset" value="${canal.instance.connectionCharset:UTF-8}" />
+	
+		<!-- 解析位点记录 -->
+		<property name="logPositionManager">
+			<bean class="com.alibaba.otter.canal.parse.index.FailbackLogPositionManager">
+				<property name="primary">
+					<bean class="com.alibaba.otter.canal.parse.index.MemoryLogPositionManager" />
+				</property>
+				<property name="failback">
+					<bean class="com.alibaba.otter.canal.parse.index.MetaLogPositionManager">
+						<property name="metaManager" ref="metaManager" />
+					</bean>
+				</property>
+			</bean>
+		</property>
+		
+		<!-- 解析数据库信息 -->
+		<property name="masterInfo">
+			<bean class="com.alibaba.otter.canal.parse.support.AuthenticationInfo">
+				<property name="address" value="${canal.instance.master.address}" />
+				<property name="username" value="${canal.instance.dbUsername:retl}" />
+				<property name="password" value="${canal.instance.dbPassword:retl}" />
+				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:retl}" />
+			</bean>
+		</property>
+		
+		<!-- 解析起始位点 -->
+		<property name="masterPosition">
+			<bean class="com.alibaba.otter.canal.protocol.position.EntryPosition">
+				<property name="journalName" value="${canal.instance.master.journal.name}" />
+				<property name="position" value="${canal.instance.master.position}" />
+				<property name="timestamp" value="${canal.instance.master.timestamp}" />
+			</bean>
+		</property>
+		<property name="filterQueryDml" value="${canal.instance.filter.query.dml:false}" />
+		<property name="filterQueryDcl" value="${canal.instance.filter.query.dcl:false}" />
+		<property name="filterQueryDdl" value="${canal.instance.filter.query.ddl:false}" />
+		<property name="filterRows" value="${canal.instance.filter.rows:false}" />
+		<property name="filterTableError" value="${canal.instance.filter.table.error:false}" />
+		<property name="needWait" value="${canal.instance.parser.needWait:true}"/>
+		<property name="directory" value="${canal.instance.parser.directory}"/>
+	</bean>
+</beans>

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

@@ -151,8 +151,9 @@
 		<property name="filterQueryDml" value="${canal.instance.filter.query.dml:false}" />
 		<property name="filterQueryDcl" value="${canal.instance.filter.query.dcl:false}" />
 		<property name="filterQueryDdl" value="${canal.instance.filter.query.ddl:false}" />
+		<property name="filterRows" value="${canal.instance.filter.rows:false}" />
 		<property name="filterTableError" value="${canal.instance.filter.table.error:false}" />
 		<property name="supportBinlogFormats" value="${canal.instance.binlog.format}" />
 		<property name="supportBinlogImages" value="${canal.instance.binlog.image}" />
 	</bean>
-</beans>
+</beans>

driver/pom.xml

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

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

@@ -12,6 +12,7 @@ import org.slf4j.LoggerFactory;
 
 import com.alibaba.otter.canal.parse.driver.mysql.packets.HeaderPacket;
 import com.alibaba.otter.canal.parse.driver.mysql.packets.client.ClientAuthenticationPacket;
+import com.alibaba.otter.canal.parse.driver.mysql.packets.client.QuitCommandPacket;
 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;
@@ -26,25 +27,27 @@ import com.alibaba.otter.canal.parse.driver.mysql.utils.PacketManager;
  */
 public class MysqlConnector {
 
-    private static final Logger logger            = LoggerFactory.getLogger(MysqlConnector.class);
-    private InetSocketAddress   address;
-    private String              username;
-    private String              password;
+    private static final Logger logger = LoggerFactory.getLogger(MysqlConnector.class);
+    private InetSocketAddress address;
+    private String username;
+    private String password;
 
-    private byte                charsetNumber     = 33;
-    private String              defaultSchema     = "retl";
-    private int                 soTimeout         = 30 * 1000;
-    private int                 receiveBufferSize = 16 * 1024;
-    private int                 sendBufferSize    = 16 * 1024;
+    private byte charsetNumber = 33;
+    private String defaultSchema = "retl";
+    private int soTimeout = 30 * 1000;
+    private int receiveBufferSize = 16 * 1024;
+    private int sendBufferSize = 16 * 1024;
 
-    private SocketChannel       channel;
-    private AtomicBoolean       connected         = new AtomicBoolean(false);
+    private SocketChannel channel;
+    private volatile boolean dumping = false;
+    // mysql connectinnId
+    private long connectionId = -1;
+    private AtomicBoolean connected = new AtomicBoolean(false);
 
     public MysqlConnector(){
     }
 
     public MysqlConnector(InetSocketAddress address, String username, String password){
-
         this.address = address;
         this.username = username;
         this.password = password;
@@ -86,11 +89,30 @@ public class MysqlConnector {
                 if (channel != null) {
                     channel.close();
                 }
-
                 logger.info("disConnect MysqlConnection to {}...", address);
             } catch (Exception e) {
                 throw new IOException("disconnect " + this.address + " failure:" + ExceptionUtils.getStackTrace(e));
             }
+
+            // 执行一次quit
+            if (dumping && connectionId >= 0) {
+                MysqlConnector connector = null;
+                try {
+                    connector = this.fork();
+                    connector.connect();
+                    MysqlUpdateExecutor executor = new MysqlUpdateExecutor(connector);
+                    executor.update("KILL CONNECTION " + connectionId);
+                } catch (Exception e) {
+                    // 忽略具体异常
+                    logger.info("KILL DUMP " + connectionId + " failure:" + ExceptionUtils.getStackTrace(e));
+                } finally {
+                    if (connector != null) {
+                        connector.disconnect();
+                    }
+                }
+
+                dumping = false;
+            }
         } else {
             logger.info("the channel {} is not connected", this.address);
         }
@@ -113,6 +135,17 @@ public class MysqlConnector {
         return connector;
     }
 
+    public void quit() throws IOException {
+        QuitCommandPacket quit = new QuitCommandPacket();
+        byte[] cmdBody = quit.toBytes();
+
+        HeaderPacket quitHeader = new HeaderPacket();
+        quitHeader.setPacketBodyLength(cmdBody.length);
+        quitHeader.setPacketSequenceNumber((byte) 0x00);
+        PacketManager.write(channel,
+            new ByteBuffer[] { ByteBuffer.wrap(quitHeader.toBytes()), ByteBuffer.wrap(cmdBody) });
+    }
+
     // ====================== help method ====================
 
     private void configChannel(SocketChannel channel) throws IOException {
@@ -140,6 +173,7 @@ public class MysqlConnector {
         }
         HandshakeInitializationPacket handshakePacket = new HandshakeInitializationPacket();
         handshakePacket.fromBytes(body);
+        connectionId = handshakePacket.threadId; // 记录一下connection
 
         logger.info("handshake initialization packet received, prepare the client authentication packet to send");
 
@@ -291,4 +325,20 @@ public class MysqlConnector {
         this.password = password;
     }
 
+    public long getConnectionId() {
+        return connectionId;
+    }
+
+    public void setConnectionId(long connectionId) {
+        this.connectionId = connectionId;
+    }
+
+    public boolean isDumping() {
+        return dumping;
+    }
+
+    public void setDumping(boolean dumping) {
+        this.dumping = dumping;
+    }
+
 }

driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/client/QuitCommandPacket.java

@@ -0,0 +1,35 @@
+package com.alibaba.otter.canal.parse.driver.mysql.packets.client;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+
+import com.alibaba.otter.canal.parse.driver.mysql.packets.CommandPacket;
+
+/**
+ * quit cmd
+ * 
+ * @author agapple 2016年3月1日 下午8:33:02
+ * @since 1.0.22
+ */
+public class QuitCommandPacket extends CommandPacket {
+
+    public static final byte[] QUIT = new byte[] { 1, 0, 0, 0, 1 };
+
+    public QuitCommandPacket(){
+        setCommand((byte) 0x01);
+    }
+
+    @Override
+    public void fromBytes(byte[] data) throws IOException {
+
+    }
+
+    @Override
+    public byte[] toBytes() throws IOException {
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        out.write(getCommand());
+        out.write(QUIT);
+        return out.toByteArray();
+    }
+
+}

example/pom.xml

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

filter/pom.xml

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

instance/core/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.0.21</version>
+		<version>1.0.24-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

@@ -0,0 +1,247 @@
+package com.alibaba.otter.canal.instance.core;
+
+import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
+import com.alibaba.otter.canal.common.alarm.CanalAlarmHandler;
+import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter;
+import com.alibaba.otter.canal.meta.CanalMetaManager;
+import com.alibaba.otter.canal.parse.CanalEventParser;
+import com.alibaba.otter.canal.parse.ha.CanalHAController;
+import com.alibaba.otter.canal.parse.ha.HeartBeatHAController;
+import com.alibaba.otter.canal.parse.inbound.AbstractEventParser;
+import com.alibaba.otter.canal.parse.inbound.group.GroupEventParser;
+import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser;
+import com.alibaba.otter.canal.parse.index.CanalLogPositionManager;
+import com.alibaba.otter.canal.protocol.CanalEntry;
+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.
+ * Author: yinxiu
+ * Date: 2016-01-07
+ * Time: 22:26
+ */
+public class AbstractCanalInstance extends AbstractCanalLifeCycle implements CanalInstance {
+
+    private static final Logger logger = LoggerFactory.getLogger(AbstractCanalInstance.class);
+
+    protected Long                                   canalId;                               // 和manager交互唯一标示
+    protected String                                 destination;                           // 队列名字
+    protected CanalEventStore<Event>                 eventStore;                            // 有序队列
+
+    protected CanalEventParser                       eventParser;                           // 解析对应的数据信息
+    protected CanalEventSink<List<CanalEntry.Entry>> eventSink;                             // 链接parse和store的桥接器
+    protected CanalMetaManager                       metaManager;                           // 消费信息管理器
+    protected CanalAlarmHandler                      alarmHandler;                          // alarm报警机制
+
+    @Override
+    public boolean subscribeChange(ClientIdentity identity) {
+        if (StringUtils.isNotEmpty(identity.getFilter())) {
+            logger.info("subscribe filter change to " + identity.getFilter());
+            AviaterRegexFilter aviaterFilter = new AviaterRegexFilter(identity.getFilter());
+
+            boolean isGroup = (eventParser instanceof GroupEventParser);
+            if (isGroup) {
+                // 处理group的模式
+                List<CanalEventParser> eventParsers = ((GroupEventParser) eventParser).getEventParsers();
+                for (CanalEventParser singleEventParser : eventParsers) {// 需要遍历启动
+                    ((AbstractEventParser) singleEventParser).setEventFilter(aviaterFilter);
+                }
+            } else {
+                ((AbstractEventParser) eventParser).setEventFilter(aviaterFilter);
+            }
+
+        }
+
+        // filter的处理规则
+        // a. parser处理数据过滤处理
+        // b. sink处理数据的路由&分发,一份parse数据经过sink后可以分发为多份，每份的数据可以根据自己的过滤规则不同而有不同的数据
+        // 后续内存版的一对多分发，可以考虑
+        return true;
+    }
+
+    @Override
+    public void start() {
+        super.start();
+        if (!metaManager.isStart()) {
+            metaManager.start();
+        }
+
+        if (!alarmHandler.isStart()) {
+            alarmHandler.start();
+        }
+
+        if (!eventStore.isStart()) {
+            eventStore.start();
+        }
+
+        if (!eventSink.isStart()) {
+            eventSink.start();
+        }
+
+        if (!eventParser.isStart()) {
+            beforeStartEventParser(eventParser);
+            eventParser.start();
+            afterStartEventParser(eventParser);
+        }
+        logger.info("start successful....");
+    }
+
+    @Override
+    public void stop() {
+        super.stop();
+        logger.info("stop CannalInstance for {}-{} ", new Object[] { canalId, destination });
+
+        if (eventParser.isStart()) {
+            beforeStopEventParser(eventParser);
+            eventParser.stop();
+            afterStopEventParser(eventParser);
+        }
+
+        if (eventSink.isStart()) {
+            eventSink.stop();
+        }
+
+        if (eventStore.isStart()) {
+            eventStore.stop();
+        }
+
+        if (metaManager.isStart()) {
+            metaManager.stop();
+        }
+
+        if (alarmHandler.isStart()) {
+            alarmHandler.stop();
+        }
+
+
+        logger.info("stop successful....");
+    }
+
+    protected void beforeStartEventParser(CanalEventParser eventParser) {
+
+        boolean isGroup = (eventParser instanceof GroupEventParser);
+        if (isGroup) {
+            // 处理group的模式
+            List<CanalEventParser> eventParsers = ((GroupEventParser) eventParser).getEventParsers();
+            for (CanalEventParser singleEventParser : eventParsers) {// 需要遍历启动
+                startEventParserInternal(singleEventParser, true);
+            }
+        } else {
+            startEventParserInternal(eventParser, false);
+        }
+    }
+
+    // around event parser, default impl
+    protected void afterStartEventParser(CanalEventParser eventParser) {
+        // 读取一下历史订阅的filter信息
+        List<ClientIdentity> clientIdentitys = metaManager.listAllSubscribeInfo(destination);
+        for (ClientIdentity clientIdentity : clientIdentitys) {
+            subscribeChange(clientIdentity);
+        }
+    }
+
+    // around event parser
+    protected void beforeStopEventParser(CanalEventParser eventParser) {
+        // noop
+    }
+
+    protected void afterStopEventParser(CanalEventParser eventParser) {
+
+        boolean isGroup = (eventParser instanceof GroupEventParser);
+        if (isGroup) {
+            // 处理group的模式
+            List<CanalEventParser> eventParsers = ((GroupEventParser) eventParser).getEventParsers();
+            for (CanalEventParser singleEventParser : eventParsers) {// 需要遍历启动
+                stopEventParserInternal(singleEventParser);
+            }
+        } else {
+            stopEventParserInternal(eventParser);
+        }
+    }
+
+    /**
+     * 初始化单个eventParser，不需要考虑group
+     */
+    protected void startEventParserInternal(CanalEventParser eventParser, boolean isGroup) {
+        if (eventParser instanceof AbstractEventParser) {
+            AbstractEventParser abstractEventParser = (AbstractEventParser) eventParser;
+            // 首先启动log position管理器
+            CanalLogPositionManager logPositionManager = abstractEventParser.getLogPositionManager();
+            if (!logPositionManager.isStart()) {
+                logPositionManager.start();
+            }
+        }
+
+        if (eventParser instanceof MysqlEventParser) {
+            MysqlEventParser mysqlEventParser = (MysqlEventParser) eventParser;
+            CanalHAController haController = mysqlEventParser.getHaController();
+
+            if (haController instanceof HeartBeatHAController) {
+                ((HeartBeatHAController) haController).setCanalHASwitchable(mysqlEventParser);
+            }
+
+            if (!haController.isStart()) {
+                haController.start();
+            }
+
+        }
+    }
+
+    protected void stopEventParserInternal(CanalEventParser eventParser) {
+        if (eventParser instanceof AbstractEventParser) {
+            AbstractEventParser abstractEventParser = (AbstractEventParser) eventParser;
+            // 首先启动log position管理器
+            CanalLogPositionManager logPositionManager = abstractEventParser.getLogPositionManager();
+            if (logPositionManager.isStart()) {
+                logPositionManager.stop();
+            }
+        }
+
+        if (eventParser instanceof MysqlEventParser) {
+            MysqlEventParser mysqlEventParser = (MysqlEventParser) eventParser;
+            CanalHAController haController = mysqlEventParser.getHaController();
+            if (haController.isStart()) {
+                haController.stop();
+            }
+        }
+    }
+
+    // ==================getter==================================
+    @Override
+    public String getDestination() {
+        return destination;
+    }
+
+    @Override
+    public CanalEventParser getEventParser() {
+        return eventParser;
+    }
+
+    @Override
+    public CanalEventSink getEventSink() {
+        return eventSink;
+    }
+
+    @Override
+    public CanalEventStore getEventStore() {
+        return eventStore;
+    }
+
+    @Override
+    public CanalMetaManager getMetaManager() {
+        return metaManager;
+    }
+
+    @Override
+    public CanalAlarmHandler getAlarmHandler() {
+        return alarmHandler;
+    }
+}

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

@@ -1,105 +0,0 @@
-package com.alibaba.otter.canal.instance.core;
-
-import java.util.List;
-
-import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
-import com.alibaba.otter.canal.parse.CanalEventParser;
-import com.alibaba.otter.canal.parse.ha.CanalHAController;
-import com.alibaba.otter.canal.parse.ha.HeartBeatHAController;
-import com.alibaba.otter.canal.parse.inbound.AbstractEventParser;
-import com.alibaba.otter.canal.parse.inbound.group.GroupEventParser;
-import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser;
-import com.alibaba.otter.canal.parse.index.CanalLogPositionManager;
-
-/**
- * @author zebin.xuzb 2012-10-17 下午3:12:34
- * @version 1.0.0
- */
-public abstract class CanalInstanceSupport extends AbstractCanalLifeCycle {
-
-    protected void beforeStartEventParser(CanalEventParser eventParser) {
-
-        boolean isGroup = (eventParser instanceof GroupEventParser);
-        if (isGroup) {
-            // 处理group的模式
-            List<CanalEventParser> eventParsers = ((GroupEventParser) eventParser).getEventParsers();
-            for (CanalEventParser singleEventParser : eventParsers) {// 需要遍历启动
-                startEventParserInternal(singleEventParser, true);
-            }
-        } else {
-            startEventParserInternal(eventParser, false);
-        }
-    }
-
-    // around event parser
-    protected void afterStartEventParser(CanalEventParser eventParser) {
-        // noop
-    }
-
-    // around event parser
-    protected void beforeStopEventParser(CanalEventParser eventParser) {
-        // noop
-    }
-
-    protected void afterStopEventParser(CanalEventParser eventParser) {
-
-        boolean isGroup = (eventParser instanceof GroupEventParser);
-        if (isGroup) {
-            // 处理group的模式
-            List<CanalEventParser> eventParsers = ((GroupEventParser) eventParser).getEventParsers();
-            for (CanalEventParser singleEventParser : eventParsers) {// 需要遍历启动
-                stopEventParserInternal(singleEventParser);
-            }
-        } else {
-            stopEventParserInternal(eventParser);
-        }
-    }
-
-    /**
-     * 初始化单个eventParser，不需要考虑group
-     */
-    protected void startEventParserInternal(CanalEventParser eventParser, boolean isGroup) {
-        if (eventParser instanceof AbstractEventParser) {
-            AbstractEventParser abstractEventParser = (AbstractEventParser) eventParser;
-            // 首先启动log position管理器
-            CanalLogPositionManager logPositionManager = abstractEventParser.getLogPositionManager();
-            if (!logPositionManager.isStart()) {
-                logPositionManager.start();
-            }
-        }
-
-        if (eventParser instanceof MysqlEventParser) {
-            MysqlEventParser mysqlEventParser = (MysqlEventParser) eventParser;
-            CanalHAController haController = mysqlEventParser.getHaController();
-
-            if (haController instanceof HeartBeatHAController) {
-                ((HeartBeatHAController) haController).setCanalHASwitchable(mysqlEventParser);
-            }
-
-            if (!haController.isStart()) {
-                haController.start();
-            }
-
-        }
-    }
-
-    protected void stopEventParserInternal(CanalEventParser eventParser) {
-        if (eventParser instanceof AbstractEventParser) {
-            AbstractEventParser abstractEventParser = (AbstractEventParser) eventParser;
-            // 首先启动log position管理器
-            CanalLogPositionManager logPositionManager = abstractEventParser.getLogPositionManager();
-            if (logPositionManager.isStart()) {
-                logPositionManager.stop();
-            }
-        }
-
-        if (eventParser instanceof MysqlEventParser) {
-            MysqlEventParser mysqlEventParser = (MysqlEventParser) eventParser;
-            CanalHAController haController = mysqlEventParser.getHaController();
-            if (haController.isStart()) {
-                haController.stop();
-            }
-        }
-    }
-
-}

instance/manager/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.0.21</version>
+		<version>1.0.24-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

@@ -17,8 +17,7 @@ import com.alibaba.otter.canal.common.alarm.LogAlarmHandler;
 import com.alibaba.otter.canal.common.utils.JsonUtils;
 import com.alibaba.otter.canal.common.zookeeper.ZkClientx;
 import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter;
-import com.alibaba.otter.canal.instance.core.CanalInstance;
-import com.alibaba.otter.canal.instance.core.CanalInstanceSupport;
+import com.alibaba.otter.canal.instance.core.AbstractCanalInstance;
 import com.alibaba.otter.canal.instance.manager.model.Canal;
 import com.alibaba.otter.canal.instance.manager.model.CanalParameter;
 import com.alibaba.otter.canal.instance.manager.model.CanalParameter.DataSourcing;
@@ -28,7 +27,6 @@ import com.alibaba.otter.canal.instance.manager.model.CanalParameter.MetaMode;
 import com.alibaba.otter.canal.instance.manager.model.CanalParameter.SourcingType;
 import com.alibaba.otter.canal.instance.manager.model.CanalParameter.StorageMode;
 import com.alibaba.otter.canal.instance.manager.model.CanalParameter.StorageScavengeMode;
-import com.alibaba.otter.canal.meta.CanalMetaManager;
 import com.alibaba.otter.canal.meta.MemoryMetaManager;
 import com.alibaba.otter.canal.meta.PeriodMixedMetaManager;
 import com.alibaba.otter.canal.meta.ZooKeeperMetaManager;
@@ -46,17 +44,12 @@ import com.alibaba.otter.canal.parse.index.MetaLogPositionManager;
 import com.alibaba.otter.canal.parse.index.PeriodMixedLogPositionManager;
 import com.alibaba.otter.canal.parse.index.ZooKeeperLogPositionManager;
 import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
-import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
-import com.alibaba.otter.canal.protocol.ClientIdentity;
 import com.alibaba.otter.canal.protocol.position.EntryPosition;
-import com.alibaba.otter.canal.sink.CanalEventSink;
 import com.alibaba.otter.canal.sink.entry.EntryEventSink;
 import com.alibaba.otter.canal.sink.entry.group.GroupEventSink;
 import com.alibaba.otter.canal.store.AbstractCanalStoreScavenge;
-import com.alibaba.otter.canal.store.CanalEventStore;
 import com.alibaba.otter.canal.store.memory.MemoryEventStoreWithBuffer;
 import com.alibaba.otter.canal.store.model.BatchMode;
-import com.alibaba.otter.canal.store.model.Event;
 
 /**
  * 单个canal实例，比如一个destination会独立一个实例
@@ -64,24 +57,11 @@ import com.alibaba.otter.canal.store.model.Event;
  * @author jianghang 2012-7-11 下午09:26:51
  * @version 1.0.0
  */
-public class CanalInstanceWithManager extends CanalInstanceSupport implements CanalInstance {
+public class CanalInstanceWithManager extends AbstractCanalInstance {
 
-    private static final Logger           logger = LoggerFactory.getLogger(CanalInstanceWithManager.class);
-    protected Long                        canalId;                                                         // 和manager交互唯一标示
-    protected String                      destination;                                                     // 队列名字
-    protected String                      filter;                                                          // 过滤表达式
-    protected CanalParameter              parameters;                                                      // 对应参数
-    protected CanalMetaManager            metaManager;                                                     // 消费信息管理器
-    protected CanalEventStore<Event>      eventStore;                                                      // 有序队列
-
-    protected CanalEventParser            eventParser;                                                     // 解析对应的数据信息
-    protected CanalEventSink<List<Entry>> eventSink;                                                       // 链接parse和store的桥接器
-    protected CanalAlarmHandler           alarmHandler;                                                    // alarm报警机制
-    protected ZkClientx                   zkClientx;
-
-    public CanalInstanceWithManager(Canal canal){
-        this(canal, null);
-    }
+    private static final Logger logger = LoggerFactory.getLogger(CanalInstanceWithManager.class);
+    protected String            filter;                                                          // 过滤表达式
+    protected CanalParameter    parameters;                                                      // 对应参数
 
     public CanalInstanceWithManager(Canal canal, String filter){
         this.parameters = canal.getCanalParameter();
@@ -89,7 +69,7 @@ public class CanalInstanceWithManager extends CanalInstanceSupport implements Ca
         this.destination = canal.getName();
         this.filter = filter;
 
-        logger.info("init CannalInstance for {}-{} with parameters:{}", canalId, destination, parameters);
+        logger.info("init CanalInstance for {}-{} with parameters:{}", canalId, destination, parameters);
         // 初始化报警机制
         initAlarmHandler();
         // 初始化metaManager
@@ -101,7 +81,7 @@ public class CanalInstanceWithManager extends CanalInstanceSupport implements Ca
         // 初始化eventParser;
         initEventParser();
 
-        // 基础工具，需要提前start，会有先订阅再根据filter条件启动paser的需求
+        // 基础工具，需要提前start，会有先订阅再根据filter条件启动parse的需求
         if (!alarmHandler.isStart()) {
             alarmHandler.start();
         }
@@ -113,98 +93,9 @@ public class CanalInstanceWithManager extends CanalInstanceSupport implements Ca
     }
 
     public void start() {
-        super.start();
         // 初始化metaManager
         logger.info("start CannalInstance for {}-{} with parameters:{}", canalId, destination, parameters);
-
-        if (!metaManager.isStart()) {
-            metaManager.start();
-        }
-
-        if (!alarmHandler.isStart()) {
-            alarmHandler.start();
-        }
-
-        if (!eventStore.isStart()) {
-            eventStore.start();
-        }
-
-        if (!eventSink.isStart()) {
-            eventSink.start();
-        }
-
-        if (!eventParser.isStart()) {
-            beforeStartEventParser(eventParser);
-            eventParser.start();
-        }
-
-        logger.info("start successful....");
-    }
-
-    public void stop() {
-        logger.info("stop CannalInstance for {}-{} ", new Object[] { canalId, destination });
-
-        if (eventParser.isStart()) {
-            eventParser.stop();
-            afterStopEventParser(eventParser);
-        }
-
-        if (eventSink.isStart()) {
-            eventSink.stop();
-        }
-
-        if (eventStore.isStart()) {
-            eventStore.stop();
-        }
-
-        if (metaManager.isStart()) {
-            metaManager.stop();
-        }
-
-        if (alarmHandler.isStart()) {
-            alarmHandler.stop();
-        }
-
-        // if (zkClientx != null) {
-        // zkClientx.close();
-        // }
-
-        super.stop();
-        logger.info("stop successful....");
-    }
-
-    public boolean subscribeChange(ClientIdentity identity) {
-        if (StringUtils.isNotEmpty(identity.getFilter())) {
-            AviaterRegexFilter aviaterFilter = new AviaterRegexFilter(identity.getFilter());
-
-            boolean isGroup = (eventParser instanceof GroupEventParser);
-            if (isGroup) {
-                // 处理group的模式
-                List<CanalEventParser> eventParsers = ((GroupEventParser) eventParser).getEventParsers();
-                for (CanalEventParser singleEventParser : eventParsers) {// 需要遍历启动
-                    ((AbstractEventParser) singleEventParser).setEventFilter(aviaterFilter);
-                }
-            } else {
-                ((AbstractEventParser) eventParser).setEventFilter(aviaterFilter);
-            }
-
-        }
-
-        // filter的处理规则
-        // a. parser处理数据过滤处理
-        // b. sink处理数据的路由&分发,一份parse数据经过sink后可以分发为多份，每份的数据可以根据自己的过滤规则不同而有不同的数据
-        // 后续内存版的一对多分发，可以考虑
-        return true;
-    }
-
-    protected void afterStartEventParser(CanalEventParser eventParser) {
-        super.afterStartEventParser(eventParser);
-
-        // 读取一下历史订阅的filter信息
-        List<ClientIdentity> clientIdentitys = metaManager.listAllSubscribeInfo(destination);
-        for (ClientIdentity clientIdentity : clientIdentitys) {
-            subscribeChange(clientIdentity);
-        }
+        super.start();
     }
 
     protected void initAlarmHandler() {
@@ -515,32 +406,6 @@ public class CanalInstanceWithManager extends CanalInstanceSupport implements Ca
         return ZkClientx.getZkClient(StringUtils.join(zkClusters, ";"));
     }
 
-    // =====================================
-
-    public String getDestination() {
-        return destination;
-    }
-
-    public CanalMetaManager getMetaManager() {
-        return metaManager;
-    }
-
-    public CanalEventStore<Event> getEventStore() {
-        return eventStore;
-    }
-
-    public CanalEventParser getEventParser() {
-        return eventParser;
-    }
-
-    public CanalEventSink<List<Entry>> getEventSink() {
-        return eventSink;
-    }
-
-    public CanalAlarmHandler getAlarmHandler() {
-        return alarmHandler;
-    }
-
     public void setAlarmHandler(CanalAlarmHandler alarmHandler) {
         this.alarmHandler = alarmHandler;
     }

instance/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.0.21</version>
+		<version>1.0.24-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.21</version>
+		<version>1.0.24-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,20 +2,14 @@ package com.alibaba.otter.canal.instance.spring;
 
 import java.util.List;
 
-import org.apache.commons.lang.StringUtils;
+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.filter.aviater.AviaterRegexFilter;
-import com.alibaba.otter.canal.instance.core.CanalInstance;
-import com.alibaba.otter.canal.instance.core.CanalInstanceSupport;
 import com.alibaba.otter.canal.meta.CanalMetaManager;
 import com.alibaba.otter.canal.parse.CanalEventParser;
-import com.alibaba.otter.canal.parse.inbound.AbstractEventParser;
-import com.alibaba.otter.canal.parse.inbound.group.GroupEventParser;
 import com.alibaba.otter.canal.protocol.CanalEntry;
-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;
@@ -27,130 +21,13 @@ import com.alibaba.otter.canal.store.model.Event;
  * @author zebin.xuzb
  * @version 1.0.0
  */
-public class CanalInstanceWithSpring extends CanalInstanceSupport implements CanalInstance {
+public class CanalInstanceWithSpring extends AbstractCanalInstance {
 
     private static final Logger                    logger = LoggerFactory.getLogger(CanalInstanceWithSpring.class);
-    private String                                 destination;
-    private CanalEventParser                       eventParser;
-    private CanalEventSink<List<CanalEntry.Entry>> eventSink;
-    private CanalEventStore<Event>                 eventStore;
-    private CanalMetaManager                       metaManager;
-    private CanalAlarmHandler                      alarmHandler;
-
-    public String getDestination() {
-        return this.destination;
-    }
-
-    public CanalEventParser getEventParser() {
-        return this.eventParser;
-    }
-
-    public CanalEventSink<List<CanalEntry.Entry>> getEventSink() {
-        return this.eventSink;
-    }
-
-    public CanalEventStore<Event> getEventStore() {
-        return this.eventStore;
-    }
-
-    public CanalMetaManager getMetaManager() {
-        return this.metaManager;
-    }
-
-    public CanalAlarmHandler getAlarmHandler() {
-        return alarmHandler;
-    }
-
-    public boolean subscribeChange(ClientIdentity identity) {
-        if (StringUtils.isNotEmpty(identity.getFilter())) {
-            logger.info("subscribe filter change to " + identity.getFilter());
-            AviaterRegexFilter aviaterFilter = new AviaterRegexFilter(identity.getFilter());
-
-            boolean isGroup = (eventParser instanceof GroupEventParser);
-            if (isGroup) {
-                // 处理group的模式
-                List<CanalEventParser> eventParsers = ((GroupEventParser) eventParser).getEventParsers();
-                for (CanalEventParser singleEventParser : eventParsers) {// 需要遍历启动
-                    ((AbstractEventParser) singleEventParser).setEventFilter(aviaterFilter);
-                }
-            } else {
-                ((AbstractEventParser) eventParser).setEventFilter(aviaterFilter);
-            }
-
-        }
-
-        // filter的处理规则
-        // a. parser处理数据过滤处理
-        // b. sink处理数据的路由&分发,一份parse数据经过sink后可以分发为多份，每份的数据可以根据自己的过滤规则不同而有不同的数据
-        // 后续内存版的一对多分发，可以考虑
-        return true;
-    }
-
-    protected void afterStartEventParser(CanalEventParser eventParser) {
-        super.afterStartEventParser(eventParser);
-
-        // 读取一下历史订阅的filter信息
-        List<ClientIdentity> clientIdentitys = metaManager.listAllSubscribeInfo(destination);
-        for (ClientIdentity clientIdentity : clientIdentitys) {
-            subscribeChange(clientIdentity);
-        }
-    }
 
     public void start() {
-        super.start();
-
         logger.info("start CannalInstance for {}-{} ", new Object[] { 1, destination });
-        if (!metaManager.isStart()) {
-            metaManager.start();
-        }
-
-        if (!eventStore.isStart()) {
-            eventStore.start();
-        }
-
-        if (!eventSink.isStart()) {
-            eventSink.start();
-        }
-
-        if (!eventParser.isStart()) {
-            beforeStartEventParser(eventParser);
-            eventParser.start();
-            afterStartEventParser(eventParser);
-        }
-
-        logger.info("start successful....");
-    }
-
-    public void stop() {
-        logger.info("stop CannalInstance for {}-{} ", new Object[] { 1, destination });
-        if (eventParser.isStart()) {
-            beforeStopEventParser(eventParser);
-            eventParser.stop();
-            afterStopEventParser(eventParser);
-        }
-
-        if (eventSink.isStart()) {
-            eventSink.stop();
-        }
-
-        if (eventStore.isStart()) {
-            eventStore.stop();
-        }
-
-        if (metaManager.isStart()) {
-            metaManager.stop();
-        }
-
-        if (alarmHandler.isStart()) {
-            alarmHandler.stop();
-        }
-
-        // if (zkClientx != null) {
-        // zkClientx.close();
-        // }
-
-        super.stop();
-        logger.info("stop successful....");
+        super.start();
     }
 
     // ======== setter ========

meta/pom.xml

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

meta/src/main/java/com/alibaba/otter/canal/meta/PeriodMixedMetaManager.java

@@ -142,8 +142,8 @@ public class PeriodMixedMetaManager extends MemoryMetaManager implements CanalMe
     }
 
     public void updateCursor(ClientIdentity clientIdentity, Position position) throws CanalMetaManagerException {
-        updateCursorTasks.add(clientIdentity);// 添加到任务队列中进行触发
         super.updateCursor(clientIdentity, position);
+        updateCursorTasks.add(clientIdentity);// 添加到任务队列中进行触发
     }
 
     public Position getCursor(ClientIdentity clientIdentity) throws CanalMetaManagerException {

parse/pom.xml

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

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

@@ -22,7 +22,6 @@ import com.alibaba.otter.canal.parse.CanalEventParser;
 import com.alibaba.otter.canal.parse.exception.CanalParseException;
 import com.alibaba.otter.canal.parse.exception.TableIdNotFoundException;
 import com.alibaba.otter.canal.parse.inbound.EventTransactionBuffer.TransactionFlushCallback;
-import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser;
 import com.alibaba.otter.canal.parse.index.CanalLogPositionManager;
 import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
 import com.alibaba.otter.canal.protocol.CanalEntry;
@@ -120,7 +119,7 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
                     throw new CanalParseException("consume failed!");
                 }
 
-                LogPosition position = buildLastTranasctionPosition(transaction);
+                LogPosition position = buildLastTransactionPosition(transaction);
                 if (position != null) { // 可能position为空
                     logPositionManager.persistLogPosition(AbstractEventParser.this.destination, position);
                 }
@@ -159,7 +158,8 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
 
                         erosaConnection.connect();// 链接
                         // 4. 获取最后的位置信息
-                        final EntryPosition startPosition = findStartPosition(erosaConnection);
+                        EntryPosition position = findStartPosition(erosaConnection);
+                        final EntryPosition startPosition = position;
                         if (startPosition == null) {
                             throw new CanalParseException("can't find start position for " + destination);
                         }
@@ -190,9 +190,9 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
                                     return running;
                                 } catch (TableIdNotFoundException e) {
                                     throw e;
-                                } catch (Exception e) {
+                                } catch (Throwable e) {
                                     // 记录一下，出错的位点信息
-                                    processError(e,
+                                    processSinkError(e,
                                         this.lastPosition,
                                         startPosition.getJournalName(),
                                         startPosition.getPosition());
@@ -219,6 +219,7 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
                         logger.error(String.format("dump address %s has an error, retrying. caused by ",
                             runningInfo.getAddress().toString()), e);
                     } catch (Throwable e) {
+                        processDumpError(e);
                         exception = e;
                         if (!running) {
                             if (!(e instanceof java.nio.channels.ClosedByInterruptException || e.getCause() instanceof java.nio.channels.ClosedByInterruptException)) {
@@ -231,6 +232,8 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
                             sendAlarm(destination, ExceptionUtils.getFullStackTrace(e));
                         }
                     } finally {
+                        // 重新置为中断状态
+                        Thread.interrupted();
                         // 关闭一下链接
                         afterDump(erosaConnection);
                         try {
@@ -321,7 +324,6 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
             startTs = System.currentTimeMillis();
         }
         CanalEntry.Entry event = binlogParser.parse(bod);
-
         if (enabled) {
             this.parsingInterval = System.currentTimeMillis() - startTs;
         }
@@ -336,20 +338,11 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
         return profilingEnabled.get();
     }
 
-    protected LogPosition buildLastTranasctionPosition(List<CanalEntry.Entry> entries) { // 初始化一下
+    protected LogPosition buildLastTransactionPosition(List<CanalEntry.Entry> entries) { // 初始化一下
         for (int i = entries.size() - 1; i > 0; i--) {
             CanalEntry.Entry entry = entries.get(i);
             if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {// 尽量记录一个事务做为position
-                LogPosition logPosition = new LogPosition();
-                EntryPosition position = new EntryPosition();
-                position.setJournalName(entry.getHeader().getLogfileName());
-                position.setPosition(entry.getHeader().getLogfileOffset());
-                position.setTimestamp(entry.getHeader().getExecuteTime());
-                logPosition.setPostion(position);
-
-                LogIdentity identity = new LogIdentity(runningInfo.getAddress(), -1L);
-                logPosition.setIdentity(identity);
-                return logPosition;
+                return buildLastPosition(entry);
             }
         }
 
@@ -362,6 +355,8 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
         position.setJournalName(entry.getHeader().getLogfileName());
         position.setPosition(entry.getHeader().getLogfileOffset());
         position.setTimestamp(entry.getHeader().getExecuteTime());
+        // add serverId at 2016-06-28
+        position.setServerId(entry.getHeader().getServerId());
         logPosition.setPostion(position);
 
         LogIdentity identity = new LogIdentity(runningInfo.getAddress(), -1L);
@@ -369,7 +364,7 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
         return logPosition;
     }
 
-    protected void processError(Exception e, LogPosition lastPosition, String startBinlogFile, long startPosition) {
+    protected void processSinkError(Throwable e, LogPosition lastPosition, String startBinlogFile, long startPosition) {
         if (lastPosition != null) {
             logger.warn(String.format("ERROR ## parse this event has an error , last position : [%s]",
                 lastPosition.getPostion()),
@@ -381,13 +376,21 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
         }
     }
 
+    protected void processDumpError(Throwable e) {
+        // do nothing
+    }
+
     protected void startHeartBeat(ErosaConnection connection) {
         lastEntryTime = 0L; // 初始化
         if (timer == null) {// lazy初始化一下
             String name = String.format("destination = %s , address = %s , HeartBeatTimeTask",
                 destination,
                 runningInfo == null ? null : runningInfo.getAddress().toString());
-            synchronized (MysqlEventParser.class) {
+            synchronized (AbstractEventParser.class) {
+                // synchronized (MysqlEventParser.class) {
+                // why use MysqlEventParser.class, u know, MysqlEventParser is
+                // the child class 4 AbstractEventParser,
+                // do this is ...
                 if (timer == null) {
                     timer = new Timer(name, true);
                 }

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

@@ -1,5 +1,6 @@
 package com.alibaba.otter.canal.parse.inbound;
 
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.commons.lang.StringUtils;
@@ -44,6 +45,17 @@ public class TableMeta {
         this.fileds = fileds;
     }
 
+    public List<FieldMeta> getPrimaryFields() {
+        List<FieldMeta> primarys = new ArrayList<TableMeta.FieldMeta>();
+        for (FieldMeta meta : fileds) {
+            if (meta.isKey()) {
+                primarys.add(meta);
+            }
+        }
+
+        return primarys;
+    }
+
     public static class FieldMeta {
 
         private String columnName;

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

@@ -22,6 +22,7 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
     protected boolean           filterQueryDcl          = false;
     protected boolean           filterQueryDml          = false;
     protected boolean           filterQueryDdl          = false;
+    protected boolean           filterRows              = false;
     protected boolean           filterTableError        = false;
 
     protected BinlogParser buildParser() {
@@ -38,6 +39,7 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
         convert.setFilterQueryDcl(filterQueryDcl);
         convert.setFilterQueryDml(filterQueryDml);
         convert.setFilterQueryDdl(filterQueryDdl);
+        convert.setFilterRows(filterRows);
         convert.setFilterTableError(filterTableError);
         return convert;
     }
@@ -87,6 +89,10 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
         this.filterQueryDdl = filterQueryDdl;
     }
 
+    public void setFilterRows(boolean filterRows) {
+        this.filterRows = filterRows;
+    }
+
     public void setFilterTableError(boolean filterTableError) {
         this.filterTableError = filterTableError;
     }

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

@@ -8,7 +8,6 @@ import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import com.alibaba.otter.canal.parse.exception.CanalParseException;
 import com.alibaba.otter.canal.parse.inbound.ErosaConnection;
 import com.alibaba.otter.canal.parse.inbound.SinkFunction;
 import com.alibaba.otter.canal.parse.inbound.mysql.local.BinLogFileQueue;
@@ -18,6 +17,7 @@ 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.QueryLogEvent;
+import com.taobao.tddl.dbsync.binlog.event.RotateLogEvent;
 
 /**
  * local bin log connection (not real connection)
@@ -83,9 +83,9 @@ public class LocalBinLogConnection implements ErosaConnection {
             context.setLogPosition(new LogPosition(binlogfilename, binlogPosition));
             while (running) {
                 boolean needContinue = true;
-                LogEvent event;
-                while (fetcher.fetch()) {
-                    event = decoder.decode(fetcher, context);
+				LogEvent event = null;
+                L:while (fetcher.fetch()) {
+                    /*event = decoder.decode(fetcher, context);
                     if (event == null) {
                         throw new CanalParseException("parse failed");
                     }
@@ -93,15 +93,20 @@ public class LocalBinLogConnection implements ErosaConnection {
                     if (!func.sink(event)) {
                         needContinue = false;
                         break;
-                    }
+                    }*/
 
-                    // do {
-                    // event = decoder.decode(fetcher, context);
-                    // if (event != null && !func.sink(event)) {
-                    // needContinue = false;
-                    // break;
-                    // }
-                    // } while (event != null);
+					do {
+						if (event == null) {
+							event = new RotateLogEvent(context.getLogPosition().getFileName(), context.getLogPosition().getPosition());
+						} else {
+							event = decoder.decode(fetcher, context);
+						}
+												
+						if (event != null && !func.sink(event)) {
+							needContinue = false;
+							break L;
+						}
+					} while (event != null);
                 }
 
                 if (needContinue) {// 读取下一个
@@ -145,6 +150,7 @@ public class LocalBinLogConnection implements ErosaConnection {
 
         FileLogFetcher fetcher = new FileLogFetcher(bufferSize);
         LogDecoder decoder = new LogDecoder();
+        decoder.handle(LogEvent.FORMAT_DESCRIPTION_EVENT);
         decoder.handle(LogEvent.QUERY_EVENT);
         decoder.handle(LogEvent.XID_EVENT);
         LogContext context = new LogContext();

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

@@ -1,10 +1,14 @@
 package com.alibaba.otter.canal.parse.inbound.mysql;
 
+import java.io.IOException;
+
 import org.apache.commons.lang.StringUtils;
 
 import com.alibaba.otter.canal.parse.CanalEventParser;
 import com.alibaba.otter.canal.parse.exception.CanalParseException;
 import com.alibaba.otter.canal.parse.inbound.ErosaConnection;
+import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert;
+import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.TableMetaCache;
 import com.alibaba.otter.canal.parse.index.CanalLogPositionManager;
 import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
 import com.alibaba.otter.canal.protocol.position.EntryPosition;
@@ -21,7 +25,9 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements
     // 数据库信息
     private AuthenticationInfo masterInfo;
     private EntryPosition      masterPosition;        // binlog信息
-
+    private MysqlConnection    metaConnection;        // 查询meta信息的链接
+    private TableMetaCache     tableMetaCache;        // 对应meta
+    
     private String             directory;
     private boolean            needWait   = false;
     private int                bufferSize = 16 * 1024;
@@ -33,17 +39,60 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements
     @Override
     protected ErosaConnection buildErosaConnection() {
         return buildLocalBinLogConnection();
-    }
+    }   
+    
+    @Override
+	protected void preDump(ErosaConnection connection) {
+    	metaConnection = buildMysqlConnection();
+        try {
+            metaConnection.connect();
+        } catch (IOException e) {
+            throw new CanalParseException(e);
+        }
+        
+        tableMetaCache = new TableMetaCache(metaConnection);
+        ((LogEventConvert) binlogParser).setTableMetaCache(tableMetaCache);
+	}
 
-    public void start() throws CanalParseException {
+	@Override
+	protected void afterDump(ErosaConnection connection) {
+		if (metaConnection != null) {
+            try {
+                metaConnection.disconnect();
+            } catch (IOException e) {
+                logger.error("ERROR # disconnect meta connection for address:{}", metaConnection.getConnector()
+                    .getAddress(), e);
+            }
+        }
+	}
+
+	public void start() throws CanalParseException {
         if (runningInfo == null) { // 第一次链接主库
             runningInfo = masterInfo;
         }
 
         super.start();
     }
+	
+    @Override
+	public void stop() {
+    	if (metaConnection != null) {
+            try {
+                metaConnection.disconnect();
+            } catch (IOException e) {
+                logger.error("ERROR # disconnect meta connection for address:{}", metaConnection.getConnector()
+                    .getAddress(), e);
+            }
+        }
 
-    private ErosaConnection buildLocalBinLogConnection() {
+        if (tableMetaCache != null) {
+            tableMetaCache.clearTableMeta();
+        }
+
+        super.stop();
+	}
+
+	private ErosaConnection buildLocalBinLogConnection() {
         LocalBinLogConnection connection = new LocalBinLogConnection();
 
         connection.setBufferSize(this.bufferSize);
@@ -53,6 +102,19 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements
         return connection;
     }
 
+    private MysqlConnection buildMysqlConnection() {
+        MysqlConnection connection = new MysqlConnection(runningInfo.getAddress(),
+            runningInfo.getUsername(),
+            runningInfo.getPassword(),
+            connectionCharsetNumber,
+            runningInfo.getDefaultDatabaseName());
+        connection.getConnector().setReceiveBufferSize(64 * 1024);
+        connection.getConnector().setSendBufferSize(64 * 1024);
+        connection.getConnector().setSoTimeout(30 * 1000);
+        connection.setCharset(connectionCharset);
+        return connection;
+    }
+    
     @Override
     protected EntryPosition findStartPosition(ErosaConnection connection) {
         // 处理逻辑

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

@@ -6,6 +6,7 @@ import java.nio.ByteBuffer;
 import java.nio.charset.Charset;
 import java.util.List;
 
+import org.apache.commons.lang.StringUtils;
 import org.apache.commons.lang.exception.ExceptionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -24,6 +25,8 @@ 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 {
 
@@ -34,6 +37,7 @@ public class MysqlConnection implements ErosaConnection {
     private Charset             charset = Charset.forName("UTF-8");
     private BinlogFormat        binlogFormat;
     private BinlogImage         binlogImage;
+    private int                 binlogChecksum;
 
     public MysqlConnection(){
     }
@@ -89,6 +93,7 @@ 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);
@@ -105,11 +110,14 @@ 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);
@@ -141,6 +149,8 @@ public class MysqlConnection implements ErosaConnection {
         binlogDumpHeader.setPacketSequenceNumber((byte) 0x00);
         PacketManager.write(connector.getChannel(), new ByteBuffer[] { ByteBuffer.wrap(binlogDumpHeader.toBytes()),
                 ByteBuffer.wrap(cmdBody) });
+
+        connector.setDumping(true);
     }
 
     public MysqlConnection fork() {
@@ -194,9 +204,12 @@ public class MysqlConnection implements ErosaConnection {
             // 如果不设置会出现错误： Slave can not handle replication events with the
             // checksum that master is configured to log
             // 但也不能乱设置，需要和mysql server的checksum配置一致，不然RotateLogEvent会出现乱码
-            update("set @master_binlog_checksum= '@@global.binlog_checksum'");
+            // '@@global.binlog_checksum'需要去掉单引号,在mysql 5.6.29下导致master退出
+            update("set @master_binlog_checksum= @@global.binlog_checksum");
         } catch (Exception e) {
-            logger.warn(ExceptionUtils.getFullStackTrace(e));
+            if (!StringUtils.contains(e.getMessage(), "Unknown system variable")) {
+                logger.warn(ExceptionUtils.getFullStackTrace(e));
+            }
         }
 
         try {
@@ -254,6 +267,27 @@ 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

@@ -13,6 +13,7 @@ import org.apache.commons.lang.StringUtils;
 import org.apache.commons.lang.exception.ExceptionUtils;
 import org.springframework.util.CollectionUtils;
 
+import com.alibaba.otter.canal.common.utils.JsonUtils;
 import com.alibaba.otter.canal.parse.CanalEventParser;
 import com.alibaba.otter.canal.parse.CanalHASwitchable;
 import com.alibaba.otter.canal.parse.driver.mysql.packets.server.FieldPacket;
@@ -28,7 +29,6 @@ import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert;
 import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.TableMetaCache;
 import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
 import com.alibaba.otter.canal.protocol.CanalEntry;
-import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
 import com.alibaba.otter.canal.protocol.position.EntryPosition;
 import com.alibaba.otter.canal.protocol.position.LogPosition;
 import com.taobao.tddl.dbsync.binlog.LogEvent;
@@ -67,7 +67,9 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
     private BinlogFormat[]     supportBinlogFormats;                         // 支持的binlogFormat,如果设置会执行强校验
     private BinlogImage[]      supportBinlogImages;                          // 支持的binlogImage,如果设置会执行强校验
 
-    // 心跳检查
+    // update by yishun.chen,特殊异常处理参数
+    private int                dumpErrorCount                    = 0;        // binlogDump失败异常计数
+    private int                dumpErrorCountThreshold           = 2;        // binlogDump失败异常计数阀值
 
     protected ErosaConnection buildErosaConnection() {
         return buildMysqlConnection(this.runningInfo);
@@ -92,6 +94,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
                 for (BinlogFormat supportFormat : supportBinlogFormats) {
                     if (supportFormat != null && format == supportFormat) {
                         found = true;
+                        break;
                     }
                 }
 
@@ -106,6 +109,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
                 for (BinlogImage supportImage : supportBinlogImages) {
                     if (supportImage != null && image == supportImage) {
                         found = true;
+                        break;
                     }
                 }
 
@@ -176,10 +180,9 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
     }
 
     protected void stopHeartBeat() {
+        TimerTask heartBeatTimerTask = this.heartBeatTimerTask;
         super.stopHeartBeat();
-
         if (heartBeatTimerTask != null) {
-
             MysqlConnection mysqlConnection = ((MysqlDetectingTimeTask) heartBeatTimerTask).getMysqlConnection();
             try {
                 mysqlConnection.disconnect();
@@ -324,6 +327,12 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
         return startPosition;
     }
 
+    protected EntryPosition findEndPosition(ErosaConnection connection) throws IOException {
+        MysqlConnection mysqlConnection = (MysqlConnection) connection;
+        EntryPosition endPosition = findEndPosition(mysqlConnection);
+        return endPosition;
+    }
+
     protected EntryPosition findStartPositionInternal(ErosaConnection connection) {
         MysqlConnection mysqlConnection = (MysqlConnection) connection;
         LogPosition logPosition = logPositionManager.getLatestIndexBy(destination);
@@ -384,7 +393,27 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
             }
         } else {
             if (logPosition.getIdentity().getSourceAddress().equals(mysqlConnection.getConnector().getAddress())) {
-                logger.warn("prepare to find start position just last position");
+                if (dumpErrorCountThreshold >= 0 && dumpErrorCount > dumpErrorCountThreshold) {
+                    // binlog定位位点失败,可能有两个原因:
+                    // 1. binlog位点被删除
+                    // 2.vip模式的mysql,发生了主备切换,判断一下serverId是否变化,针对这种模式可以发起一次基于时间戳查找合适的binlog位点
+                    boolean case2 = (standbyInfo == null || standbyInfo.getAddress() == null)
+                                    && logPosition.getPostion().getServerId() != null
+                                    && !logPosition.getPostion().getServerId().equals(findServerId(mysqlConnection));
+                    if (case2) {
+                        long timestamp = logPosition.getPostion().getTimestamp();
+                        long newStartTimestamp = timestamp - fallbackIntervalInSeconds * 1000;
+                        logger.warn("prepare to find start position by last position {}:{}:{}", new Object[] { "", "",
+                                logPosition.getPostion().getTimestamp() });
+                        EntryPosition findPosition = findByStartTimeStamp(mysqlConnection, newStartTimestamp);
+                        // 重新置为一下
+                        dumpErrorCount = 0;
+                        return findPosition;
+                    }
+                }
+                // 其余情况
+                logger.warn("prepare to find start position just last position\n {}",
+                    JsonUtils.marshalToString(logPosition));
                 return logPosition.getPostion();
             } else {
                 // 针对切换的情况，考虑回退时间
@@ -426,7 +455,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
                     }
                 } catch (Exception e) {
                     // 上一次记录的poistion可能为一条update/insert/delete变更事件，直接进行dump的话，会缺少tableMap事件，导致tableId未进行解析
-                    processError(e, lastPosition, entryPosition.getJournalName(), entryPosition.getPosition());
+                    processSinkError(e, lastPosition, entryPosition.getJournalName(), entryPosition.getPosition());
                     reDump.set(true);
                     return false;
                 }
@@ -460,7 +489,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
 
                         lastPosition = buildLastPosition(entry);
                     } catch (Exception e) {
-                        processError(e, lastPosition, entryPosition.getJournalName(), entryPosition.getPosition());
+                        processSinkError(e, lastPosition, entryPosition.getJournalName(), entryPosition.getPosition());
                         return false;
                     }
 
@@ -540,6 +569,22 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
         return null;
     }
 
+    /**
+     * 查询当前db的serverId信息
+     */
+    private Long findServerId(MysqlConnection mysqlConnection) {
+        try {
+            ResultSetPacket packet = mysqlConnection.query("show variables like 'server_id'");
+            List<String> fields = packet.getFieldValues();
+            if (CollectionUtils.isEmpty(fields)) {
+                throw new CanalParseException("command : show variables like 'server_id' has an error! pls check. you need (at least one of) the SUPER,REPLICATION CLIENT privilege(s) for this operation");
+            }
+            return Long.valueOf(fields.get(1));
+        } catch (IOException e) {
+            throw new CanalParseException("command : show variables like 'server_id' has an error!", e);
+        }
+    }
+
     /**
      * 查询当前的binlog位置
      */
@@ -681,8 +726,8 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
                         }
 
                         lastPosition = buildLastPosition(entry);
-                    } catch (Exception e) {
-                        processError(e, lastPosition, searchBinlogFile, 4L);
+                    } catch (Throwable e) {
+                        processSinkError(e, lastPosition, searchBinlogFile, 4L);
                     }
 
                     return running;
@@ -700,21 +745,17 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
         }
     }
 
-    protected Entry parseAndProfilingIfNecessary(LogEvent bod) throws Exception {
-        long startTs = -1;
-        boolean enabled = getProfilingEnabled();
-        if (enabled) {
-            startTs = System.currentTimeMillis();
-        }
-        CanalEntry.Entry event = binlogParser.parse(bod);
-        if (enabled) {
-            this.parsingInterval = System.currentTimeMillis() - startTs;
+    @Override
+    protected void processDumpError(Throwable e) {
+        if (e instanceof IOException) {
+            String message = e.getMessage();
+            if (StringUtils.contains(message, "errno = 1236")) {
+                // 1236 errorCode代表ER_MASTER_FATAL_ERROR_READING_BINLOG
+                dumpErrorCount++;
+            }
         }
 
-        if (parsedEventCount.incrementAndGet() < 0) {
-            parsedEventCount.set(0);
-        }
-        return event;
+        super.processDumpError(e);
     }
 
     public void setSupportBinlogFormats(String formatStrs) {
@@ -800,4 +841,9 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
     public void setHaController(CanalHAController haController) {
         this.haController = haController;
     }
+
+    public void setDumpErrorCountThreshold(int dumpErrorCountThreshold) {
+        this.dumpErrorCountThreshold = dumpErrorCountThreshold;
+    }
+
 }

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

@@ -84,6 +84,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
     private boolean                     filterQueryDdl      = false;
     // 是否跳过table相关的解析异常,比如表不存在或者列数量不匹配,issue 92
     private boolean                     filterTableError    = false;
+    // 新增rows过滤，用于仅订阅除rows以外的数据
+    private boolean                     filterRows          = false;
 
     public Entry parse(LogEvent logEvent) throws CanalParseException {
         if (logEvent == null || logEvent instanceof UnknownLogEvent) {
@@ -305,6 +307,9 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
     }
 
     private Entry parseRowsEvent(RowsLogEvent event) {
+        if (filterRows) {
+            return null;
+        }
         try {
             TableMapLogEvent table = event.getTable();
             if (table == null) {
@@ -321,6 +326,11 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
                 return null;
             }
 
+            if (tableMetaCache.isOnRDS() && "mysql.ha_health_check".equals(fullname)) {
+                // 忽略rds模式的mysql.ha_health_check心跳数据
+                return null;
+            }
+
             EventType eventType = null;
             int type = event.getHeader().getType();
             if (LogEvent.WRITE_ROWS_EVENT_V1 == type || LogEvent.WRITE_ROWS_EVENT == type) {
@@ -345,7 +355,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
             rowChangeBuider.setEventType(eventType);
             RowsLogBuffer buffer = event.getRowsBuf(charset.name());
             BitSet columns = event.getColumns();
-            BitSet changeColumns = event.getColumns();
+            BitSet changeColumns = event.getChangeColumns();
             boolean tableError = false;
             TableMeta tableMeta = null;
             if (tableMetaCache != null) {// 入错存在table meta cache
@@ -375,7 +385,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
                         break;
                     }
 
-                    tableError |= parseOneRow(rowDataBuilder, event, buffer, event.getChangeColumns(), true, tableMeta);
+                    tableError |= parseOneRow(rowDataBuilder, event, buffer, changeColumns, true, tableMeta);
                 }
 
                 rowChangeBuider.addRowDatas(rowDataBuilder.build());
@@ -397,34 +407,51 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
 
     private boolean parseOneRow(RowData.Builder rowDataBuilder, RowsLogEvent event, RowsLogBuffer buffer, BitSet cols,
                                 boolean isAfter, TableMeta tableMeta) throws UnsupportedEncodingException {
-        final int columnCnt = event.getTable().getColumnCnt();
-        final ColumnInfo[] columnInfo = event.getTable().getColumnInfo();
+        int columnCnt = event.getTable().getColumnCnt();
+        ColumnInfo[] columnInfo = event.getTable().getColumnInfo();
 
         boolean tableError = false;
         // check table fileds count，只能处理加字段
+        boolean existRDSNoPrimaryKey = false;
         if (tableMeta != null && columnInfo.length > tableMeta.getFileds().size()) {
-            // online ddl增加字段操作步骤：
-            // 1. 新增一张临时表，将需要做ddl表的数据全量导入
-            // 2. 在老表上建立I/U/D的trigger，增量的将数据插入到临时表
-            // 3. 锁住应用请求，将临时表rename为老表的名字，完成增加字段的操作
-            // 尝试做一次reload，可能因为ddl没有正确解析，或者使用了类似online ddl的操作
-            // 因为online ddl没有对应表名的alter语法，所以不会有clear cache的操作
-            tableMeta = getTableMeta(event.getTable().getDbName(), event.getTable().getTableName(), false);// 强制重新获取一次
-            if (tableMeta == null) {
-                tableError = true;
-                if (!filterTableError) {
-                    throw new CanalParseException("not found [" + event.getTable().getDbName() + "."
-                                                  + event.getTable().getTableName() + "] in db , pls check!");
+            if (tableMetaCache.isOnRDS()) {
+                // 特殊处理下RDS的场景
+                List<FieldMeta> primaryKeys = tableMeta.getPrimaryFields();
+                if (primaryKeys == null || primaryKeys.isEmpty()) {
+                    if (columnInfo.length == tableMeta.getFileds().size() + 1
+                        && columnInfo[columnInfo.length - 1].type == LogEvent.MYSQL_TYPE_LONGLONG) {
+                        existRDSNoPrimaryKey = true;
+                    }
                 }
             }
 
-            // 在做一次判断
-            if (tableMeta != null && columnInfo.length > tableMeta.getFileds().size()) {
-                tableError = true;
-                if (!filterTableError) {
-                    throw new CanalParseException("column size is not match for table:" + tableMeta.getFullName() + ","
-                                                  + columnInfo.length + " vs " + tableMeta.getFileds().size());
+            if (!existRDSNoPrimaryKey) {
+                // online ddl增加字段操作步骤：
+                // 1. 新增一张临时表，将需要做ddl表的数据全量导入
+                // 2. 在老表上建立I/U/D的trigger，增量的将数据插入到临时表
+                // 3. 锁住应用请求，将临时表rename为老表的名字，完成增加字段的操作
+                // 尝试做一次reload，可能因为ddl没有正确解析，或者使用了类似online ddl的操作
+                // 因为online ddl没有对应表名的alter语法，所以不会有clear cache的操作
+                tableMeta = getTableMeta(event.getTable().getDbName(), event.getTable().getTableName(), false);// 强制重新获取一次
+                if (tableMeta == null) {
+                    tableError = true;
+                    if (!filterTableError) {
+                        throw new CanalParseException("not found [" + event.getTable().getDbName() + "."
+                                                      + event.getTable().getTableName() + "] in db , pls check!");
+                    }
                 }
+
+                // 在做一次判断
+                if (tableMeta != null && columnInfo.length > tableMeta.getFileds().size()) {
+                    tableError = true;
+                    if (!filterTableError) {
+                        throw new CanalParseException("column size is not match for table:" + tableMeta.getFullName()
+                                                      + "," + columnInfo.length + " vs " + tableMeta.getFileds().size());
+                    }
+                }
+            } else {
+                logger.warn("[" + event.getTable().getDbName() + "." + event.getTable().getTableName()
+                            + "] is no primary key , skip alibaba_rds_row_id column");
             }
         }
 
@@ -435,6 +462,12 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
                 continue;
             }
 
+            if (existRDSNoPrimaryKey && i == columnCnt - 1 && info.type == LogEvent.MYSQL_TYPE_LONGLONG) {
+                // 不解析最后一列
+                buffer.nextValue(info.type, info.meta, false);
+                continue;
+            }
+
             Column.Builder columnBuilder = Column.newBuilder();
 
             FieldMeta fieldMeta = null;
@@ -460,6 +493,10 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
                 }
             }
             buffer.nextValue(info.type, info.meta, isBinary);
+            if (existRDSNoPrimaryKey && i == columnCnt - 1 && info.type == LogEvent.MYSQL_TYPE_LONGLONG) {
+                // 不解析最后一列
+                continue;
+            }
 
             int javaType = buffer.getJavaType();
             if (buffer.isNull()) {
@@ -714,4 +751,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
         this.filterTableError = filterTableError;
     }
 
+    public void setFilterRows(boolean filterRows) {
+        this.filterRows = filterRows;
+    }
+
 }

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

@@ -32,6 +32,7 @@ public class TableMetaCache {
     public static final String     COLUMN_DEFAULT = "COLUMN_DEFAULT";
     public static final String     EXTRA          = "EXTRA";
     private MysqlConnection        connection;
+    private boolean                isOnRDS        = false;
 
     // 第一层tableId,第二层schema.table,解决tableId重复，对应多张表
     private Map<String, TableMeta> tableMetaCache;
@@ -56,6 +57,13 @@ public class TableMetaCache {
 
         });
 
+        try {
+            ResultSetPacket packet = connection.query("show global variables  like 'rds\\_%'");
+            if (packet.getFieldValues().size() > 0) {
+                isOnRDS = true;
+            }
+        } catch (IOException e) {
+        }
     }
 
     public TableMeta getTableMeta(String schema, String table) {
@@ -136,4 +144,13 @@ public class TableMetaCache {
             .append('`')
             .toString();
     }
+
+    public boolean isOnRDS() {
+        return isOnRDS;
+    }
+
+    public void setOnRDS(boolean isOnRDS) {
+        this.isOnRDS = isOnRDS;
+    }
+
 }

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

@@ -9,6 +9,8 @@ import java.util.Timer;
 import java.util.TimerTask;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 import org.apache.commons.io.FileUtils;
 import org.apache.commons.io.filefilter.IOFileFilter;
@@ -188,7 +190,9 @@ public class BinLogFileQueue {
         files.addAll(FileUtils.listFiles(directory, new IOFileFilter() {
 
             public boolean accept(File file) {
-                return file.getName().startsWith(baseName);
+				Pattern pattern = Pattern.compile("\\d+$");
+				Matcher matcher = pattern.matcher(file.getName());
+				return file.getName().startsWith(baseName) && matcher.find();
             }
 
             public boolean accept(File dir, String name) {

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

@@ -13,6 +13,7 @@ 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;
 import com.alibaba.otter.canal.protocol.position.EntryPosition;
 import com.alibaba.otter.canal.protocol.position.LogIdentity;
 import com.alibaba.otter.canal.protocol.position.LogPosition;
@@ -22,8 +23,8 @@ public class MysqlEventParserTest {
 
     private static final String DETECTING_SQL = "insert into retl.xdual values(1,now()) on duplicate key update x=now()";
     private static final String MYSQL_ADDRESS = "127.0.0.1";
-    private static final String USERNAME      = "xxxxx";
-    private static final String PASSWORD      = "xxxxx";
+    private static final String USERNAME      = "root";
+    private static final String PASSWORD      = "xxxxxx";
 
     @Test
     public void test_position() throws InterruptedException {
@@ -45,20 +46,23 @@ public class MysqlEventParserTest {
             public boolean sink(List<Entry> entrys, InetSocketAddress remoteAddress, String destination)
                                                                                                         throws CanalSinkException {
                 for (Entry entry : entrys) {
-                    entryCount.incrementAndGet();
-                    String logfilename = entry.getHeader().getLogfileName();
-                    long logfileoffset = entry.getHeader().getLogfileOffset();
-                    long executeTime = entry.getHeader().getExecuteTime();
-                    entryPosition.setJournalName(logfilename);
-                    entryPosition.setPosition(logfileoffset);
-                    entryPosition.setTimestamp(executeTime);
-
-                    break;
+                    if (entry.getEntryType() != EntryType.HEARTBEAT) {
+                        entryCount.incrementAndGet();
+                        String logfilename = entry.getHeader().getLogfileName();
+                        long logfileoffset = entry.getHeader().getLogfileOffset();
+                        long executeTime = entry.getHeader().getExecuteTime();
+                        entryPosition.setJournalName(logfilename);
+                        entryPosition.setPosition(logfileoffset);
+                        entryPosition.setTimestamp(executeTime);
+                        break;
+                    }
                 }
 
-                controller.stop();
-                timeoutChecker.stop();
-                timeoutChecker.touch();
+                if (entryCount.get() > 0) {
+                    controller.stop();
+                    timeoutChecker.stop();
+                    timeoutChecker.touch();
+                }
                 return true;
             }
         });
@@ -92,14 +96,14 @@ public class MysqlEventParserTest {
 
     @Test
     public void test_timestamp() throws InterruptedException {
-        final TimeoutChecker timeoutChecker = new TimeoutChecker(30 * 1000);
+        final TimeoutChecker timeoutChecker = new TimeoutChecker(3000 * 1000);
         final AtomicLong entryCount = new AtomicLong(0);
         final EntryPosition entryPosition = new EntryPosition();
 
         final MysqlEventParser controller = new MysqlEventParser();
-        final EntryPosition defaultPosition = buildPosition(null, null, 1322803601000L);
+        final EntryPosition defaultPosition = buildPosition(null, null, 1475116855000L);
         controller.setSlaveId(3344L);
-        controller.setDetectingEnable(true);
+        controller.setDetectingEnable(false);
         controller.setDetectingSQL(DETECTING_SQL);
         controller.setMasterInfo(buildAuthentication());
         controller.setMasterPosition(defaultPosition);
@@ -109,22 +113,25 @@ public class MysqlEventParserTest {
             public boolean sink(List<Entry> entrys, InetSocketAddress remoteAddress, String destination)
                                                                                                         throws CanalSinkException {
                 for (Entry entry : entrys) {
-                    entryCount.incrementAndGet();
+                    if (entry.getEntryType() != EntryType.HEARTBEAT) {
+                        entryCount.incrementAndGet();
 
-                    String logfilename = entry.getHeader().getLogfileName();
-                    long logfileoffset = entry.getHeader().getLogfileOffset();
-                    long executeTime = entry.getHeader().getExecuteTime();
+                        String logfilename = entry.getHeader().getLogfileName();
+                        long logfileoffset = entry.getHeader().getLogfileOffset();
+                        long executeTime = entry.getHeader().getExecuteTime();
 
-                    entryPosition.setJournalName(logfilename);
-                    entryPosition.setPosition(logfileoffset);
-                    entryPosition.setTimestamp(executeTime);
-
-                    break;
+                        entryPosition.setJournalName(logfilename);
+                        entryPosition.setPosition(logfileoffset);
+                        entryPosition.setTimestamp(executeTime);
+                        break;
+                    }
                 }
 
-                controller.stop();
-                timeoutChecker.stop();
-                timeoutChecker.touch();
+                if (entryCount.get() > 0) {
+                    controller.stop();
+                    timeoutChecker.stop();
+                    timeoutChecker.touch();
+                }
                 return true;
             }
         });
@@ -174,21 +181,26 @@ public class MysqlEventParserTest {
             public boolean sink(List<Entry> entrys, InetSocketAddress remoteAddress, String destination)
                                                                                                         throws CanalSinkException {
                 for (Entry entry : entrys) {
-                    entryCount.incrementAndGet();
+                    if (entry.getEntryType() != EntryType.HEARTBEAT) {
 
-                    String logfilename = entry.getHeader().getLogfileName();
-                    long logfileoffset = entry.getHeader().getLogfileOffset();
-                    long executeTime = entry.getHeader().getExecuteTime();
+                        entryCount.incrementAndGet();
 
-                    entryPosition.setJournalName(logfilename);
-                    entryPosition.setPosition(logfileoffset);
-                    entryPosition.setTimestamp(executeTime);
-                    break;
+                        String logfilename = entry.getHeader().getLogfileName();
+                        long logfileoffset = entry.getHeader().getLogfileOffset();
+                        long executeTime = entry.getHeader().getExecuteTime();
+
+                        entryPosition.setJournalName(logfilename);
+                        entryPosition.setPosition(logfileoffset);
+                        entryPosition.setTimestamp(executeTime);
+                        break;
+                    }
                 }
 
-                controller.stop();
-                timeoutChecker.stop();
-                timeoutChecker.touch();
+                if (entryCount.get() > 0) {
+                    controller.stop();
+                    timeoutChecker.stop();
+                    timeoutChecker.touch();
+                }
                 return true;
             }
         });
@@ -244,22 +256,27 @@ public class MysqlEventParserTest {
             public boolean sink(List<Entry> entrys, InetSocketAddress remoteAddress, String destination)
                                                                                                         throws CanalSinkException {
                 for (Entry entry : entrys) {
-                    entryCount.incrementAndGet();
+                    if (entry.getEntryType() != EntryType.HEARTBEAT) {
+                        entryCount.incrementAndGet();
 
-                    // String logfilename = entry.getHeader().getLogfileName();
-                    // long logfileoffset =
-                    // entry.getHeader().getLogfileOffset();
-                    long executeTime = entry.getHeader().getExecuteTime();
+                        // String logfilename =
+                        // entry.getHeader().getLogfileName();
+                        // long logfileoffset =
+                        // entry.getHeader().getLogfileOffset();
+                        long executeTime = entry.getHeader().getExecuteTime();
 
-                    // entryPosition.setJournalName(logfilename);
-                    // entryPosition.setPosition(logfileoffset);
-                    entryPosition.setTimestamp(executeTime);
-                    break;
+                        // entryPosition.setJournalName(logfilename);
+                        // entryPosition.setPosition(logfileoffset);
+                        entryPosition.setTimestamp(executeTime);
+                        break;
+                    }
                 }
 
-                controller.stop();
-                timeoutChecker.stop();
-                timeoutChecker.touch();
+                if (entryCount.get() > 0) {
+                    controller.stop();
+                    timeoutChecker.stop();
+                    timeoutChecker.touch();
+                }
                 return true;
             }
         });

pom.xml

@@ -4,7 +4,7 @@
     <artifactId>canal</artifactId>
     <packaging>pom</packaging>
     <name>canal module for otter ${project.version}</name>
-    <version>1.0.21</version>
+    <version>1.0.24-SNAPSHOT</version>
     <url>https://github.com/alibaba/canal</url>
     <parent>
         <groupId>org.sonatype.oss</groupId>
@@ -202,7 +202,7 @@
             <dependency>
                 <groupId>com.google.protobuf</groupId>
                 <artifactId>protobuf-java</artifactId>
-                <version>2.4.1</version>
+                <version>2.6.1</version>
             </dependency>
             <!-- log -->
             <dependency>

protocol/pom.xml

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

protocol/src/main/java/com/alibaba/otter/canal/protocol/position/EntryPosition.java

@@ -15,6 +15,8 @@ public class EntryPosition extends TimePosition {
     private boolean           included              = false;
     private String            journalName;
     private Long              position;
+    // add by agapple at 2016-06-28
+    private Long              serverId              = null;              // 记录一下位点对应的serverId
 
     public EntryPosition(){
         super(null);
@@ -34,6 +36,11 @@ public class EntryPosition extends TimePosition {
         this.position = position;
     }
 
+    public EntryPosition(String journalName, Long position, Long timestamp, Long serverId){
+        this(journalName, position, timestamp);
+        this.serverId = serverId;
+    }
+
     public String getJournalName() {
         return journalName;
     }
@@ -58,6 +65,14 @@ public class EntryPosition extends TimePosition {
         this.included = included;
     }
 
+    public Long getServerId() {
+        return serverId;
+    }
+
+    public void setServerId(Long serverId) {
+        this.serverId = serverId;
+    }
+
     @Override
     public int hashCode() {
         final int prime = 31;

server/pom.xml

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

server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java

@@ -21,6 +21,7 @@ import com.alibaba.otter.canal.protocol.position.LogPosition;
 import com.alibaba.otter.canal.protocol.position.Position;
 import com.alibaba.otter.canal.protocol.position.PositionRange;
 import com.alibaba.otter.canal.server.CanalServer;
+import com.alibaba.otter.canal.server.CanalService;
 import com.alibaba.otter.canal.server.exception.CanalServerException;
 import com.alibaba.otter.canal.store.CanalEventStore;
 import com.alibaba.otter.canal.store.model.Event;
@@ -37,24 +38,39 @@ import com.google.common.collect.MigrateMap;
  * @author zebin.xuzb
  * @version 1.0.0
  */
-public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements CanalServer, com.alibaba.otter.canal.server.CanalService {
+public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements CanalServer, CanalService {
 
     private static final Logger        logger = LoggerFactory.getLogger(CanalServerWithEmbedded.class);
     private Map<String, CanalInstance> canalInstances;
     // private Map<ClientIdentity, Position> lastRollbackPostions;
     private CanalInstanceGenerator     canalInstanceGenerator;
 
+    private static class SingletonHolder {
+
+        private static final CanalServerWithEmbedded CANAL_SERVER_WITH_EMBEDDED = new CanalServerWithEmbedded();
+    }
+
+    public CanalServerWithEmbedded(){
+        // 希望也保留用户new单独实例的需求,兼容历史
+    }
+
+    public static CanalServerWithEmbedded instance() {
+        return SingletonHolder.CANAL_SERVER_WITH_EMBEDDED;
+    }
+
     public void start() {
-        super.start();
+        if (!isStart()) {
+            super.start();
 
-        canalInstances = MigrateMap.makeComputingMap(new Function<String, CanalInstance>() {
+            canalInstances = MigrateMap.makeComputingMap(new Function<String, CanalInstance>() {
 
-            public CanalInstance apply(String destination) {
-                return canalInstanceGenerator.generate(destination);
-            }
-        });
+                public CanalInstance apply(String destination) {
+                    return canalInstanceGenerator.generate(destination);
+                }
+            });
 
-        // lastRollbackPostions = new MapMaker().makeMap();
+            // lastRollbackPostions = new MapMaker().makeMap();
+        }
     }
 
     public void stop() {
@@ -115,6 +131,8 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
      */
     @Override
     public void subscribe(ClientIdentity clientIdentity) throws CanalServerException {
+    	checkStart(clientIdentity.getDestination());
+    	
         CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
         if (!canalInstance.getMetaManager().isStart()) {
             canalInstance.getMetaManager().start();

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

@@ -33,11 +33,17 @@ public class CanalServerWithNetty extends AbstractCanalLifeCycle implements Cana
     private Channel                 serverChannel = null;
     private ServerBootstrap         bootstrap     = null;
 
-    public CanalServerWithNetty(){
+    private static class SingletonHolder {
+
+        private static final CanalServerWithNetty CANAL_SERVER_WITH_NETTY = new CanalServerWithNetty();
     }
 
-    public CanalServerWithNetty(CanalServerWithEmbedded embeddedServer){
-        this.embeddedServer = embeddedServer;
+    private CanalServerWithNetty(){
+        this.embeddedServer = CanalServerWithEmbedded.instance();
+    }
+
+    public static CanalServerWithNetty instance() {
+        return SingletonHolder.CANAL_SERVER_WITH_NETTY;
     }
 
     public void start() {

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

@@ -66,7 +66,6 @@ public class SessionHandler extends SimpleChannelHandler {
                             Short.valueOf(sub.getClientId()),
                             sub.getFilter());
                         MDC.put("destination", clientIdentity.getDestination());
-                        embeddedServer.subscribe(clientIdentity);
 
                         // 尝试启动，如果已经启动，忽略
                         if (!embeddedServer.isStart(clientIdentity.getDestination())) {
@@ -76,6 +75,7 @@ public class SessionHandler extends SimpleChannelHandler {
                             }
                         }
 
+                        embeddedServer.subscribe(clientIdentity);
                         ctx.setAttachment(clientIdentity);// 设置状态数据
                         NettyUtils.ack(ctx.getChannel(), null);
                     } else {

server/src/test/java/com/alibaba/otter/canal/server/BaseCanalServerWithEmbededTest.java

@@ -31,7 +31,7 @@ public abstract class BaseCanalServerWithEmbededTest {
 
     @Before
     public void setUp() {
-        server = new CanalServerWithEmbedded();
+        server = CanalServerWithEmbedded.instance();
         server.setCanalInstanceGenerator(new CanalInstanceGenerator() {
 
             public CanalInstance generate(String destination) {

server/src/test/java/com/alibaba/otter/canal/server/CanalServerWithNettyTest.java

@@ -58,7 +58,8 @@ public class CanalServerWithNettyTest {
             }
         });
 
-        nettyServer = new CanalServerWithNetty(embeddedServer);
+        nettyServer = CanalServerWithNetty.instance();
+        nettyServer.setEmbeddedServer(embeddedServer);
         nettyServer.setPort(1088);
         nettyServer.start();
     }

sink/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.0.21</version>
+		<version>1.0.24-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.21</version>
+		<version>1.0.24-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>

store/src/main/java/com/alibaba/otter/canal/store/helper/CanalEventUtils.java

@@ -51,6 +51,8 @@ public class CanalEventUtils {
         position.setJournalName(event.getEntry().getHeader().getLogfileName());
         position.setPosition(event.getEntry().getHeader().getLogfileOffset());
         position.setTimestamp(event.getEntry().getHeader().getExecuteTime());
+        // add serverId at 2016-06-28
+        position.setServerId(event.getEntry().getHeader().getServerId());
 
         LogPosition logPosition = new LogPosition();
         logPosition.setPostion(position);