[maven-release-plugin] prepare release canal-1.0.22

RDS主从库切换时候异常处理

.gitignore

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

README.md

@@ -36,7 +36,7 @@
 <li>slave重做中继日志中的事件，将改变反映它自己的数据。</li>
 </ol>
 <h3>canal的工作原理：</h3>
-<p><img width="590" src="http://dl.iteye.com/upload/attachment/0080/3107/c87b67ba-394c-3086-9577-9db05be04c95.jpg" alt="" height="273">
+<p><img width="590" src="https://camo.githubusercontent.com/46c626b4cde399db43b2634a7911a04aecf273a0/687474703a2f2f646c2e69746579652e636f6d2f75706c6f61642f6174746163686d656e742f303038302f333130372f63383762363762612d333934632d333038362d393537372d3964623035626530346339352e6a7067" alt="" height="273">
 <p>原理相对比较简单：</p>
 <ol>
 <li>canal模拟mysql slave的交互协议，伪装自己为mysql slave，向mysql master发送dump协议</li>
@@ -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.20</version>
+		<version>1.0.22</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.20</version>
+		<version>1.0.22</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<artifactId>canal.common</artifactId>

dbsync/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.0.20</version>
+		<version>1.0.22</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/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;
@@ -163,10 +166,13 @@ public final class RowsLogBuffer {
 
             case LogEvent.MYSQL_TYPE_TIMESTAMP:
             case LogEvent.MYSQL_TYPE_DATETIME:
+            case LogEvent.MYSQL_TYPE_TIMESTAMP2:
+            case LogEvent.MYSQL_TYPE_DATETIME2:
                 javaType = Types.TIMESTAMP;
                 break;
 
             case LogEvent.MYSQL_TYPE_TIME:
+            case LogEvent.MYSQL_TYPE_TIME2:
                 javaType = Types.TIME;
                 break;
 
@@ -942,6 +948,23 @@ 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.VARBINARY;
+                length = len;
+                break;
+            }
             case LogEvent.MYSQL_TYPE_GEOMETRY: {
                 /*
                  * MYSQL_TYPE_GEOMETRY: copy from BLOB or TEXT
@@ -1023,4 +1046,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.20</version>
+		<version>1.0.22</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>

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

@@ -88,9 +88,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);
 
@@ -231,7 +231,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();

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/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.20</version>
+		<version>1.0.22</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;
@@ -38,13 +39,15 @@ public class MysqlConnector {
     private int                 sendBufferSize    = 16 * 1024;
 
     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,29 @@ 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) {
+                    throw new IOException("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 +134,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 +172,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 +324,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();
+    }
+
+}

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

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

example/pom.xml

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

example/src/main/bin/init.sh

@@ -0,0 +1,90 @@
+#! /bin/bash
+current_path=`pwd`
+case "`uname`" in
+    Linux)
+                bin_abs_path=$(readlink -f $(dirname $0))
+                ;;
+        *)
+                bin_abs_path=`cd $(dirname $0); pwd`
+                ;;
+esac
+base=${bin_abs_path}/..
+
+# default values
+PORT=3306
+MYSQL_IP="0"
+
+# parse arguments
+while getopts "h:p:" opt; do
+    case $opt in
+        p)
+            if [[ $OPTARG -gt 0 ]]; then
+		PORT=$OPTARG
+            else
+                echo "Invalid -p option: -$OPTARG, should be positive integer >&2"
+                exit 1
+            fi
+            ;;
+       h)
+            MYSQL_IP=$OPTARG
+            ;;
+        \?)
+            printUsage
+            exit 1
+            ;;
+    esac
+done
+
+function printUsage
+{
+    cat << __Usage
+
+Initialise the instance config property file automatically. This tool will connect to Mysql master server and get the binlog file and position then update the instance config property file for you.
+
+Usage: $0 -p [port] -h [hostname]
+Options:
+    -p: mysql port, default to 3306
+    -h: mysql host ip.
+    For example:
+        $0 -p 3306 -h 172.17.0.2
+
+Enjoy!!!
+
+__Usage
+}
+
+if [[ $PORT -lt 0 ]]; then
+  echo "invalid port argument $PORT"
+  printUsage
+  exit 1
+fi
+
+# check if mysql ip is specified
+if [[ $MYSQL_IP = "0" ]]; then
+  echo '-h mysql host ip argument is required.'
+  printUsage
+  exit 1
+fi
+
+MYSQL_ADDR="$MYSQL_IP:$PORT"
+
+MASTER_STATUS=`mysql -ucanal -pcanal -h$MYSQL_IP -P$PORT -e 'show master status' 2>/dev/null | head -n 2 | tail -n 1`
+
+LOG_FILE=`echo "$MASTER_STATUS" | cut -f 1`
+LOG_POS=`echo "$MASTER_STATUS" | cut -f 2`
+
+INSTANCE_CONF_FILE=$base/conf/example/instance.properties
+
+if [ ! -f $INSTANCE_CONF_FILE ]; then
+  echo "Can not find instance config file: $INSTANCE_CONF_FILE"
+  exit 1
+fi
+
+# replace the binlog file and position
+sed -i.bak -e "s/canal\.instance\.master\.address = .*/canal\.instance\.master\.address = $MYSQL_ADDR/" -e "s/canal\.instance\.master\.journal\.name = .*/canal\.instance\.master\.journal\.name = $LOG_FILE/" -e "s/canal\.instance\.master\.position = .*/canal\.instance\.master\.position = $LOG_POS/" $INSTANCE_CONF_FILE
+
+cat $INSTANCE_CONF_FILE
+
+echo
+echo 'Above is your instance config file, check if the settings are correct'
+echo

filter/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.0.20</version>
+		<version>1.0.22</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.20</version>
+		<version>1.0.22</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.20</version>
+		<version>1.0.22</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.20</version>
+		<version>1.0.22</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.20</version>
+		<version>1.0.22</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.20</version>
+		<version>1.0.22</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>

parse/pom.xml

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

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

@@ -120,7 +120,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 +159,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 +191,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 +220,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 +233,8 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
                             sendAlarm(destination, ExceptionUtils.getFullStackTrace(e));
                         }
                     } finally {
+                        // 重新置为中断状态
+                        Thread.interrupted();
                         // 关闭一下链接
                         afterDump(erosaConnection);
                         try {
@@ -336,20 +340,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 +357,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 +366,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,6 +378,10 @@ 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初始化一下

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/LocalBinlogEventParser.java

@@ -52,7 +52,7 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements
 
         return connection;
     }
-
+    
     @Override
     protected EntryPosition findStartPosition(ErosaConnection connection) {
         // 处理逻辑

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

@@ -141,6 +141,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() {

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

@@ -28,7 +28,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 +66,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);
@@ -324,6 +325,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,6 +391,25 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
             }
         } else {
             if (logPosition.getIdentity().getSourceAddress().equals(mysqlConnection.getConnector().getAddress())) {
+                if (dumpErrorCountThreshold >= 0 && dumpErrorCount > dumpErrorCountThreshold) {
+                    // binlog定位位点失败,可能有两个原因:
+                    // 1. binlog位点被删除
+                    // 2.vip模式的mysql,发生了主备切换,判断一下serverId是否变化,针对这种模式可以发起一次基于时间戳查找合适的binlog位点
+                    boolean case2 = (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");
                 return logPosition.getPostion();
             } else {
@@ -426,7 +452,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 +486,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 +566,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 +723,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 +742,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 +838,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) {
@@ -215,12 +217,20 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
             // 更新下table meta cache
             if (tableMetaCache != null
                 && (result.getType() == EventType.ALTER || result.getType() == EventType.ERASE || result.getType() == EventType.RENAME)) {
-                if (StringUtils.isNotEmpty(tableName)) {
-                    // 如果解析到了正确的表信息，则根据全名进行清除
-                    tableMetaCache.clearTableMeta(schemaName, tableName);
-                } else {
-                    // 如果无法解析正确的表信息，则根据schema进行清除
-                    tableMetaCache.clearTableMetaWithSchemaName(schemaName);
+                for (DdlResult renameResult = result; renameResult != null; renameResult = renameResult.getRenameTableResult()) {
+                    String schemaName0 = event.getDbName(); // 防止rename语句后产生schema变更带来影响
+                    if (StringUtils.isNotEmpty(renameResult.getSchemaName())) {
+                        schemaName0 = renameResult.getSchemaName();
+                    }
+
+                    tableName = renameResult.getTableName();
+                    if (StringUtils.isNotEmpty(tableName)) {
+                        // 如果解析到了正确的表信息，则根据全名进行清除
+                        tableMetaCache.clearTableMeta(schemaName0, tableName);
+                    } else {
+                        // 如果无法解析正确的表信息，则根据schema进行清除
+                        tableMetaCache.clearTableMetaWithSchemaName(schemaName0);
+                    }
                 }
             }
 
@@ -297,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) {
@@ -337,7 +350,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
@@ -367,7 +380,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());
@@ -615,8 +628,9 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
                 if (column.getIsNull() && newValue == null) {
                     // 如果全是null
                     return false;
-                } else if (newValue != null && column.getValue().equals(newValue)) {
-                    // 如果不围null，并且相等
+                } else if (newValue != null && (!column.getIsNull() && column.getValue().equals(newValue))) {
+                    // fixed issue #135, old column is Null
+                    // 如果不为null，并且相等
                     return false;
                 }
             }
@@ -704,5 +718,9 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
     public void setFilterTableError(boolean filterTableError) {
         this.filterTableError = filterTableError;
     }
+    
+    public void setFilterRows(boolean filterRows) {
+        this.filterRows = filterRows;
+    }
 
 }

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

@@ -23,15 +23,17 @@ import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
  */
 public class SimpleDdlParser {
 
-    public static final String CREATE_PATTERN       = "^\\s*CREATE\\s*(TEMPORARY)?\\s*TABLE\\s*(.*)$";
-    public static final String DROP_PATTERN         = "^\\s*DROP\\s*(TEMPORARY)?\\s*TABLE\\s*(.*)$";
-    public static final String ALERT_PATTERN        = "^\\s*ALTER\\s*(IGNORE)?\\s*TABLE\\s*(.*)$";
-    public static final String TRUNCATE_PATTERN     = "^\\s*TRUNCATE\\s*(TABLE)?\\s*(.*)$";
-    public static final String TABLE_PATTERN        = "^(IF\\s*NOT\\s*EXISTS\\s*)?(IF\\s*EXISTS\\s*)?(`?.+?`?[;\\(\\s]+?)?.*$";         // 采用非贪婪模式
-    public static final String INSERT_PATTERN       = "^\\s*(INSERT|MERGE|REPLACE)(.*)$";
-    public static final String UPDATE_PATTERN       = "^\\s*UPDATE(.*)$";
-    public static final String DELETE_PATTERN       = "^\\s*DELETE(.*)$";
-    public static final String RENAME_PATTERN       = "^\\s*RENAME\\s*TABLE\\s*(.*?)\\s*TO\\s*(.*?)$";
+    public static final String CREATE_PATTERN         = "^\\s*CREATE\\s*(TEMPORARY)?\\s*TABLE\\s*(.*)$";
+    public static final String DROP_PATTERN           = "^\\s*DROP\\s*(TEMPORARY)?\\s*TABLE\\s*(.*)$";
+    public static final String ALERT_PATTERN          = "^\\s*ALTER\\s*(IGNORE)?\\s*TABLE\\s*(.*)$";
+    public static final String TRUNCATE_PATTERN       = "^\\s*TRUNCATE\\s*(TABLE)?\\s*(.*)$";
+    public static final String TABLE_PATTERN          = "^(IF\\s*NOT\\s*EXISTS\\s*)?(IF\\s*EXISTS\\s*)?(`?.+?`?[;\\(\\s]+?)?.*$";         // 采用非贪婪模式
+    public static final String INSERT_PATTERN         = "^\\s*(INSERT|MERGE|REPLACE)(.*)$";
+    public static final String UPDATE_PATTERN         = "^\\s*UPDATE(.*)$";
+    public static final String DELETE_PATTERN         = "^\\s*DELETE(.*)$";
+    public static final String RENAME_PATTERN         = "^\\s*RENAME\\s+TABLE\\s+(.+?)\\s+TO\\s+(.+?)$";
+    public static final String RENAME_REMNANT_PATTERN = "^\\s*(.+?)\\s+TO\\s+(.+?)$";
+
     /**
      * <pre>
      * CREATE [UNIQUE|FULLTEXT|SPATIAL] INDEX index_name
@@ -42,8 +44,8 @@ public class SimpleDdlParser {
      * http://dev.mysql.com/doc/refman/5.6/en/create-index.html
      * </pre>
      */
-    public static final String CREATE_INDEX_PATTERN = "^\\s*CREATE\\s*(UNIQUE)?(FULLTEXT)?(SPATIAL)?\\s*INDEX\\s*(.*?)\\s*ON\\s*(.*?)$";
-    public static final String DROP_INDEX_PATTERN   = "^\\s*DROP\\s*INDEX\\s*(.*?)\\s*ON\\s*(.*?)$";
+    public static final String CREATE_INDEX_PATTERN   = "^\\s*CREATE\\s*(UNIQUE)?(FULLTEXT)?(SPATIAL)?\\s*INDEX\\s*(.*?)\\s*ON\\s*(.*?)$";
+    public static final String DROP_INDEX_PATTERN     = "^\\s*DROP\\s*INDEX\\s*(.*?)\\s*ON\\s*(.*?)$";
 
     public static DdlResult parse(String queryString, String schmeaName) {
         queryString = removeComment(queryString); // 去除/* */的sql注释内容
@@ -74,6 +76,18 @@ public class SimpleDdlParser {
         result = parseRename(queryString, schmeaName, RENAME_PATTERN);
         if (result != null) {
             result.setType(EventType.RENAME);
+
+            String[] renameStrings = queryString.split(",");
+            if (renameStrings.length > 1) {
+                DdlResult lastResult = result;
+                for (int i = 1; i < renameStrings.length; i++) {
+                    DdlResult ddlResult = parseRename(renameStrings[i], schmeaName, RENAME_REMNANT_PATTERN);
+                    ddlResult.setType(EventType.RENAME);
+                    lastResult.setRenameTableResult(ddlResult);
+                    lastResult = ddlResult;
+                }
+            }
+
             return result;
         }
 
@@ -209,9 +223,15 @@ public class SimpleDdlParser {
 
         private String    schemaName;
         private String    tableName;
-        private String    oriSchemaName; // rename ddl中的源表
-        private String    oriTableName; // rename ddl中的目标表
+        private String    oriSchemaName;    // rename ddl中的源表
+        private String    oriTableName;     // rename ddl中的目标表
         private EventType type;
+        private DdlResult renameTableResult; // 多个rename table的存储
+
+        /*
+         * RENAME TABLE tbl_name TO new_tbl_name [, tbl_name2 TO new_tbl_name2]
+         * ...
+         */
 
         public DdlResult(){
         }
@@ -272,11 +292,28 @@ public class SimpleDdlParser {
             this.oriTableName = oriTableName;
         }
 
-        @Override
-        public String toString() {
-            return "DdlResult [schemaName=" + schemaName + ", tableName=" + tableName + ", oriSchemaName="
-                   + oriSchemaName + ", oriTableName=" + oriTableName + ", type=" + type + "]";
+        public DdlResult getRenameTableResult() {
+            return renameTableResult;
         }
 
+        public void setRenameTableResult(DdlResult renameTableResult) {
+            this.renameTableResult = renameTableResult;
+        }
+
+        @Override
+        public String toString() {
+            DdlResult ddlResult = this;
+            StringBuffer sb = new StringBuffer();
+            do {
+                sb.append(String.format("DdlResult [schemaName=%s , tableName=%s , oriSchemaName=%s , oriTableName=%s , type=%s ];",
+                    ddlResult.schemaName,
+                    ddlResult.tableName,
+                    ddlResult.oriSchemaName,
+                    ddlResult.oriTableName,
+                    ddlResult.type));
+                ddlResult = ddlResult.renameTableResult;
+            } while (ddlResult != null);
+            return sb.toString();
+        }
     }
 }

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

@@ -149,6 +149,61 @@ public class SimpleDdlParserTest {
         Assert.assertEquals("retl2", result.getSchemaName());
         Assert.assertEquals("retl_mark", result.getOriTableName());
         Assert.assertEquals("retl_mark2", result.getTableName());
+
+        queryString = "rename \n table \n `retl`.`retl_mark` to `retl2.retl_mark2` , `retl1`.`retl_mark1` to `retl3.retl_mark3`;";
+        result = SimpleDdlParser.parse(queryString, "retl");
+        Assert.assertNotNull(result);
+        Assert.assertEquals("retl", result.getOriSchemaName());
+        Assert.assertEquals("retl2", result.getSchemaName());
+        Assert.assertEquals("retl_mark", result.getOriTableName());
+        Assert.assertEquals("retl_mark2", result.getTableName());
+        result = result.getRenameTableResult();
+        Assert.assertNotNull(result);
+        Assert.assertEquals("retl1", result.getOriSchemaName());
+        Assert.assertEquals("retl3", result.getSchemaName());
+        Assert.assertEquals("retl_mark1", result.getOriTableName());
+        Assert.assertEquals("retl_mark3", result.getTableName());
+
+        //正则匹配test case
+
+        queryString = "rename table totl_mark to totl_mark2";
+        result = SimpleDdlParser.parse(queryString, "retl");
+        Assert.assertNotNull(result);
+        Assert.assertEquals("retl", result.getOriSchemaName());
+        Assert.assertEquals("retl", result.getSchemaName());
+        Assert.assertEquals("totl_mark", result.getOriTableName());
+        Assert.assertEquals("totl_mark2", result.getTableName());
+
+        queryString = "rename table totl.retl_mark to totl2.retl_mark2";
+        result = SimpleDdlParser.parse(queryString, "retl");
+        Assert.assertNotNull(result);
+        Assert.assertEquals("totl", result.getOriSchemaName());
+        Assert.assertEquals("totl2", result.getSchemaName());
+        Assert.assertEquals("retl_mark", result.getOriTableName());
+        Assert.assertEquals("retl_mark2", result.getTableName());
+
+        queryString = "rename \n table \n `totl`.`retl_mark` to `totl2.retl_mark2`;";
+        result = SimpleDdlParser.parse(queryString, "retl");
+        Assert.assertNotNull(result);
+        Assert.assertEquals("totl", result.getOriSchemaName());
+        Assert.assertEquals("totl2", result.getSchemaName());
+        Assert.assertEquals("retl_mark", result.getOriTableName());
+        Assert.assertEquals("retl_mark2", result.getTableName());
+
+        queryString = "rename \n table \n `totl`.`retl_mark` to `totl2.retl_mark2` , `totl1`.`retl_mark1` to `totl3.retl_mark3`;";
+        result = SimpleDdlParser.parse(queryString, "retl");
+        Assert.assertNotNull(result);
+        Assert.assertEquals("totl", result.getOriSchemaName());
+        Assert.assertEquals("totl2", result.getSchemaName());
+        Assert.assertEquals("retl_mark", result.getOriTableName());
+        Assert.assertEquals("retl_mark2", result.getTableName());
+        result = result.getRenameTableResult();
+        Assert.assertNotNull(result);
+        Assert.assertEquals("totl1", result.getOriSchemaName());
+        Assert.assertEquals("totl3", result.getSchemaName());
+        Assert.assertEquals("retl_mark1", result.getOriTableName());
+        Assert.assertEquals("retl_mark3", result.getTableName());
+
     }
 
     @Test

pom.xml

@@ -4,7 +4,7 @@
     <artifactId>canal</artifactId>
     <packaging>pom</packaging>
     <name>canal module for otter ${project.version}</name>
-    <version>1.0.20</version>
+    <version>1.0.22</version>
     <url>https://github.com/alibaba/canal</url>
     <parent>
         <groupId>org.sonatype.oss</groupId>

protocol/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.0.20</version>
+		<version>1.0.22</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.20</version>
+		<version>1.0.22</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() {

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/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.20</version>
+		<version>1.0.22</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.20</version>
+		<version>1.0.22</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);