fixed issue #626, support xa rollback/commit event

在CanalConnector接口加入stopRunning方法，用于优雅停止client

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

@@ -150,4 +150,9 @@ public interface CanalConnector {
      */
     void rollback() throws CanalClientException;
 
+    /**
+     * 中断的阻塞，用于优雅停止client
+     * @throws CanalClientException
+     */
+    void stopRunning() throws CanalClientException;
 }

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

@@ -23,7 +23,7 @@ public class ClusterCanalConnector implements CanalConnector {
     private String                  username;
     private String                  password;
     private int                     soTimeout     = 10000;
-    private int                     retryTimes    = 3;
+    private int                     retryTimes    = 3;                                       // 设置-1时可以subscribe阻塞等待时优雅停机
     private int                     retryInterval = 5000;                                    // 重试的时间间隔，默认5秒
     private CanalNodeAccessStrategy accessStrategy;
     private SimpleCanalConnector    currentConnector;
@@ -106,13 +106,19 @@ public class ClusterCanalConnector implements CanalConnector {
                 this.filter = filter;
                 return;
             } catch (Throwable t) {
-                logger.warn(String.format(
-                        "something goes wrong when subscribing from server: %s",
-                        currentConnector != null ? currentConnector.getAddress() : "null"),
-                        t);
-                times++;
-                restart();
-                logger.info("restart the connector for next round retry.");
+                if (retryTimes == -1 && t.getCause() instanceof InterruptedException) {
+                    logger.info("block waiting interrupted by other thread.");
+                    return;
+                } else {
+                    logger.warn(String.format(
+                            "something goes wrong when subscribing from server: %s",
+                            currentConnector != null ? currentConnector.getAddress() : "null"),
+                            t);
+                    times++;
+                    restart();
+                    logger.info("restart the connector for next round retry.");
+                }
+
             }
         }
 
@@ -322,4 +328,8 @@ public class ClusterCanalConnector implements CanalConnector {
         return currentConnector;
     }
 
+    public void stopRunning() {
+        currentConnector.stopRunning();
+    }
+
 }

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

@@ -44,7 +44,7 @@ import com.google.protobuf.ByteString;
 
 /**
  * 基于{@linkplain CanalServerWithNetty}定义的网络协议接口，对于canal数据进行get/rollback/ack等操作
- * 
+ *
  * @author jianghang 2012-10-24 下午05:37:20
  * @version 1.0.0
  */
@@ -75,6 +75,8 @@ public class SimpleCanalConnector implements CanalConnector {
     private Object               readDataLock          = new Object();
     private Object               writeDataLock         = new Object();
 
+    private boolean              running               = false;
+
     public SimpleCanalConnector(SocketAddress address, String username, String password, String destination){
         this(address, username, password, destination, 60000);
     }
@@ -99,6 +101,9 @@ public class SimpleCanalConnector implements CanalConnector {
             }
         } else {
             waitClientRunning();
+            if (!running) {
+                return;
+            }
             doConnect();
             if (filter != null) { // 如果存在条件，说明是自动切换，基于上一次的条件订阅一次
                 subscribe(filter);
@@ -208,6 +213,9 @@ public class SimpleCanalConnector implements CanalConnector {
 
     public void subscribe(String filter) throws CanalClientException {
         waitClientRunning();
+        if (!running) {
+            return;
+        }
         try {
             writeWithHeader(Packet.newBuilder()
                 .setType(PacketType.SUBSCRIPTION)
@@ -234,6 +242,9 @@ public class SimpleCanalConnector implements CanalConnector {
 
     public void unsubscribe() throws CanalClientException {
         waitClientRunning();
+        if (!running) {
+            return;
+        }
         try {
             writeWithHeader(Packet.newBuilder()
                 .setType(PacketType.UNSUBSCRIPTION)
@@ -445,7 +456,11 @@ public class SimpleCanalConnector implements CanalConnector {
                     throw new CanalClientException("should connect first");
                 }
 
+                running = true;
                 mutex.get();// 阻塞等待
+            } else {
+                // 单机模式直接设置为running
+                running = true;
             }
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
@@ -502,4 +517,13 @@ public class SimpleCanalConnector implements CanalConnector {
         this.filter = filter;
     }
 
+    public void stopRunning() {
+        if (running) {
+            running = false;  //设置为非running状态
+            if (!mutex.state()) {
+                mutex.set(true);  //中断阻塞
+            }
+        }
+    }
+
 }

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

@@ -103,8 +103,8 @@ public final class CharsetConversion {
         putEntry(42, "latin7", "latin7_general_cs", "ISO8859_7");
         putEntry(43, "macce", "macce_bin", "MacCentralEurope");
         putEntry(44, "cp1250", "cp1250_croatian_ci", "Cp1250");
-        putEntry(45, "utf8mb4", "utf8mb4_general_ci", "MacCentralEurope");
-        putEntry(46, "utf8mb4", "utf8mb4_bin", "MacCentralEurope");
+        putEntry(45, "utf8mb4", "utf8mb4_general_ci", "UTF-8");
+        putEntry(46, "utf8mb4", "utf8mb4_bin", "UTF-8");
         putEntry(47, "latin1", "latin1_bin", "ISO8859_1");
         putEntry(48, "latin1", "latin1_general_ci", "ISO8859_1");
         putEntry(49, "latin1", "latin1_general_cs", "ISO8859_1");

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

@@ -75,7 +75,9 @@ public final class FileLogFetcher extends LogFetcher {
         fin = new FileInputStream(file);
 
         ensureCapacity(BIN_LOG_HEADER_SIZE);
-        if (BIN_LOG_HEADER_SIZE != fin.read(buffer, 0, BIN_LOG_HEADER_SIZE)) throw new IOException("No binlog file header");
+        if (BIN_LOG_HEADER_SIZE != fin.read(buffer, 0, BIN_LOG_HEADER_SIZE)) {
+            throw new IOException("No binlog file header");
+        }
 
         if (buffer[0] != BINLOG_MAGIC[0] || buffer[1] != BINLOG_MAGIC[1] || buffer[2] != BINLOG_MAGIC[2]
             || buffer[3] != BINLOG_MAGIC[3]) {
@@ -128,6 +130,19 @@ public final class FileLogFetcher extends LogFetcher {
                 return true;
             }
         } else if (limit > 0) {
+            if (limit >= FormatDescriptionLogEvent.LOG_EVENT_HEADER_LEN) {
+                int lenPosition = position + 4 + 1 + 4;
+                long eventLen = ((long) (0xff & buffer[lenPosition++])) | ((long) (0xff & buffer[lenPosition++]) << 8)
+                                | ((long) (0xff & buffer[lenPosition++]) << 16)
+                                | ((long) (0xff & buffer[lenPosition++]) << 24);
+
+                if (limit >= eventLen) {
+                    return true;
+                } else {
+                    ensureCapacity((int) eventLen);
+                }
+            }
+
             System.arraycopy(buffer, origin, buffer, 0, limit);
             position -= origin;
             origin = 0;

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

@@ -30,15 +30,19 @@ import com.taobao.tddl.dbsync.binlog.event.RowsQueryLogEvent;
 import com.taobao.tddl.dbsync.binlog.event.StartLogEventV3;
 import com.taobao.tddl.dbsync.binlog.event.StopLogEvent;
 import com.taobao.tddl.dbsync.binlog.event.TableMapLogEvent;
+import com.taobao.tddl.dbsync.binlog.event.TransactionContextLogEvent;
 import com.taobao.tddl.dbsync.binlog.event.UnknownLogEvent;
 import com.taobao.tddl.dbsync.binlog.event.UpdateRowsLogEvent;
 import com.taobao.tddl.dbsync.binlog.event.UserVarLogEvent;
+import com.taobao.tddl.dbsync.binlog.event.ViewChangeEvent;
 import com.taobao.tddl.dbsync.binlog.event.WriteRowsLogEvent;
+import com.taobao.tddl.dbsync.binlog.event.XaPrepareLogEvent;
 import com.taobao.tddl.dbsync.binlog.event.XidLogEvent;
 import com.taobao.tddl.dbsync.binlog.event.mariadb.AnnotateRowsEvent;
 import com.taobao.tddl.dbsync.binlog.event.mariadb.BinlogCheckPointLogEvent;
 import com.taobao.tddl.dbsync.binlog.event.mariadb.MariaGtidListLogEvent;
 import com.taobao.tddl.dbsync.binlog.event.mariadb.MariaGtidLogEvent;
+import com.taobao.tddl.dbsync.binlog.event.mariadb.StartEncryptionLogEvent;
 
 /**
  * Implements a binary-log decoder.
@@ -366,6 +370,24 @@ public final class LogDecoder {
                 logPosition.position = header.getLogPos();
                 return event;
             }
+            case LogEvent.TRANSACTION_CONTEXT_EVENT: {
+                TransactionContextLogEvent event = new TransactionContextLogEvent(header, buffer, descriptionEvent);
+                /* updating position in context */
+                logPosition.position = header.getLogPos();
+                return event;
+            }
+            case LogEvent.VIEW_CHANGE_EVENT: {
+                ViewChangeEvent event = new ViewChangeEvent(header, buffer, descriptionEvent);
+                /* updating position in context */
+                logPosition.position = header.getLogPos();
+                return event;
+            }
+            case LogEvent.XA_PREPARE_LOG_EVENT: {
+                XaPrepareLogEvent event = new XaPrepareLogEvent(header, buffer, descriptionEvent);
+                /* updating position in context */
+                logPosition.position = header.getLogPos();
+                return event;
+            }
             case LogEvent.ANNOTATE_ROWS_EVENT: {
                 AnnotateRowsEvent event = new AnnotateRowsEvent(header, buffer, descriptionEvent);
                 /* updating position in context */
@@ -390,6 +412,12 @@ public final class LogDecoder {
                 logPosition.position = header.getLogPos();
                 return event;
             }
+            case LogEvent.START_ENCRYPTION_EVENT: {
+                StartEncryptionLogEvent event = new StartEncryptionLogEvent(header, buffer, descriptionEvent);
+                /* updating position in context */
+                logPosition.position = header.getLogPos();
+                return event;
+            }
             default:
                 /*
                  * Create an object of Ignorable_log_event for unrecognized
@@ -402,9 +430,10 @@ public final class LogDecoder {
                     logPosition.position = header.getLogPos();
                     return event;
                 } else {
-                    if (logger.isWarnEnabled()) logger.warn("Skipping unrecognized binlog event "
-                                                            + LogEvent.getTypeName(header.getType()) + " from: "
-                                                            + context.getLogPosition());
+                    if (logger.isWarnEnabled()) {
+                        logger.warn("Skipping unrecognized binlog event " + LogEvent.getTypeName(header.getType())
+                                    + " from: " + context.getLogPosition());
+                    }
                 }
         }
 

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

@@ -161,9 +161,17 @@ public abstract class LogEvent {
 
     public static final int    PREVIOUS_GTIDS_LOG_EVENT                 = 35;
 
+    /* MySQL 5.7 events */
+    public static final int    TRANSACTION_CONTEXT_EVENT                = 36;
+
+    public static final int    VIEW_CHANGE_EVENT                        = 37;
+
+    /* Prepared XA transaction terminal event similar to Xid */
+    public static final int    XA_PREPARE_LOG_EVENT                     = 38;
+
     // mariaDb 5.5.34
     /* New MySQL/Sun events are to be added right above this comment */
-    public static final int    MYSQL_EVENTS_END                         = 36;
+    public static final int    MYSQL_EVENTS_END                         = 39;
 
     public static final int    MARIA_EVENTS_BEGIN                       = 160;
     /* New Maria event numbers start from here */
@@ -189,8 +197,10 @@ public abstract class LogEvent {
      */
     public static final int    GTID_LIST_EVENT                          = 163;
 
+    public static final int    START_ENCRYPTION_EVENT                   = 164;
+
     /** end marker */
-    public static final int    ENUM_END_EVENT                           = 164;
+    public static final int    ENUM_END_EVENT                           = 165;
 
     /**
      * 1 byte length, 1 byte format Length is total length in bytes, including 2
@@ -359,7 +369,7 @@ public abstract class LogEvent {
     protected static final Log logger = LogFactory.getLog(LogEvent.class);
 
     protected final LogHeader  header;
-    
+
     /**
      * mysql半同步semi标识
      * 
@@ -369,18 +379,16 @@ public abstract class LogEvent {
      * </pre>
      */
     protected int              semival;
-    
-    
 
     public int getSemival() {
-		return semival;
-	}
+        return semival;
+    }
 
-	public void setSemival(int semival) {
-		this.semival = semival;
-	}
+    public void setSemival(int semival) {
+        this.semival = semival;
+    }
 
-	protected LogEvent(LogHeader header){
+    protected LogEvent(LogHeader header){
         this.header = header;
     }
 

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

@@ -59,10 +59,15 @@ public final class FormatDescriptionLogEvent extends StartLogEventV3 {
     public static final int   HEARTBEAT_HEADER_LEN                = 0;
     public static final int   IGNORABLE_HEADER_LEN                = 0;
     public static final int   ROWS_HEADER_LEN_V2                  = 10;
+    public static final int   TRANSACTION_CONTEXT_HEADER_LEN      = 18;
+    public static final int   VIEW_CHANGE_HEADER_LEN              = 52;
+    public static final int   XA_PREPARE_HEADER_LEN               = 0;
+
     public static final int   ANNOTATE_ROWS_HEADER_LEN            = 0;
     public static final int   BINLOG_CHECKPOINT_HEADER_LEN        = 4;
     public static final int   GTID_HEADER_LEN                     = 19;
     public static final int   GTID_LIST_HEADER_LEN                = 4;
+    public static final int   START_ENCRYPTION_HEADER_LEN         = 0;
 
     public static final int   POST_HEADER_LENGTH                  = 11;
 
@@ -202,11 +207,17 @@ public final class FormatDescriptionLogEvent extends StartLogEventV3 {
                 postHeaderLen[GTID_LOG_EVENT - 1] = POST_HEADER_LENGTH;
                 postHeaderLen[ANONYMOUS_GTID_LOG_EVENT - 1] = POST_HEADER_LENGTH;
                 postHeaderLen[PREVIOUS_GTIDS_LOG_EVENT - 1] = IGNORABLE_HEADER_LEN;
+
+                postHeaderLen[TRANSACTION_CONTEXT_EVENT - 1] = TRANSACTION_CONTEXT_HEADER_LEN;
+                postHeaderLen[VIEW_CHANGE_EVENT - 1] = VIEW_CHANGE_HEADER_LEN;
+                postHeaderLen[XA_PREPARE_LOG_EVENT - 1] = XA_PREPARE_HEADER_LEN;
+
                 // mariadb 10
                 postHeaderLen[ANNOTATE_ROWS_EVENT - 1] = ANNOTATE_ROWS_HEADER_LEN;
                 postHeaderLen[BINLOG_CHECKPOINT_EVENT - 1] = BINLOG_CHECKPOINT_HEADER_LEN;
                 postHeaderLen[GTID_EVENT - 1] = GTID_HEADER_LEN;
                 postHeaderLen[GTID_LIST_EVENT - 1] = GTID_LIST_HEADER_LEN;
+                postHeaderLen[START_ENCRYPTION_EVENT - 1] = START_ENCRYPTION_HEADER_LEN;
                 break;
 
             case 3: /* 4.0.x x>=2 */

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

@@ -1,5 +1,8 @@
 package com.taobao.tddl.dbsync.binlog.event;
 
+import java.nio.ByteBuffer;
+import java.util.UUID;
+
 import com.taobao.tddl.dbsync.binlog.LogBuffer;
 import com.taobao.tddl.dbsync.binlog.LogEvent;
 
@@ -16,6 +19,8 @@ public class GtidLogEvent extends LogEvent {
     public static final int ENCODED_SID_LENGTH  = 16;
 
     private boolean         commitFlag;
+    private UUID            sid;
+    private long            gno;
 
     public GtidLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){
         super(header);
@@ -27,6 +32,14 @@ public class GtidLogEvent extends LogEvent {
         buffer.position(commonHeaderLen);
         commitFlag = (buffer.getUint8() != 0); // ENCODED_FLAG_LENGTH
 
+        byte[] bs = buffer.getData(ENCODED_SID_LENGTH);
+        ByteBuffer bb = ByteBuffer.wrap(bs);
+        long high = bb.getLong();
+        long low = bb.getLong();
+        sid = new UUID(high, low);
+
+        gno = buffer.getLong64();
+
         // ignore gtid info read
         // sid.copy_from((uchar *)ptr_buffer);
         // ptr_buffer+= ENCODED_SID_LENGTH;
@@ -42,4 +55,11 @@ public class GtidLogEvent extends LogEvent {
         return commitFlag;
     }
 
+    public UUID getSid() {
+        return sid;
+    }
+
+    public long getGno() {
+        return gno;
+    }
 }

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

@@ -480,13 +480,27 @@ public final class RowsLogBuffer {
                     // cal.set(d / 10000, (d % 10000) / 100 - 1, d % 100, t /
                     // 10000, (t % 10000) / 100, t % 100);
                     // value = new Timestamp(cal.getTimeInMillis());
-                    value = String.format("%04d-%02d-%02d %02d:%02d:%02d",
-                        d / 10000,
-                        (d % 10000) / 100,
-                        d % 100,
-                        t / 10000,
-                        (t % 10000) / 100,
-                        t % 100);
+                    // value = String.format("%04d-%02d-%02d %02d:%02d:%02d",
+                    // d / 10000,
+                    // (d % 10000) / 100,
+                    // d % 100,
+                    // t / 10000,
+                    // (t % 10000) / 100,
+                    // t % 100);
+
+                    StringBuilder builder = new StringBuilder();
+                    builder.append(formatNumber(d / 10000, 4))
+                        .append('-')
+                        .append(formatNumber((d % 10000) / 100, 2))
+                        .append('-')
+                        .append(formatNumber(d % 100, 2))
+                        .append(' ')
+                        .append(formatNumber(t / 10000, 2))
+                        .append(':')
+                        .append(formatNumber((t % 10000) / 100, 2))
+                        .append(':')
+                        .append(formatNumber(t % 100, 2));
+                    value = builder.toString();
                 }
                 javaType = Types.TIMESTAMP;
                 length = 8;
@@ -540,13 +554,27 @@ public final class RowsLogBuffer {
                     // % (1 << 5)), (int) (hms >> 12),
                     // (int) ((hms >> 6) % (1 << 6)), (int) (hms % (1 << 6)));
                     // value = new Timestamp(cal.getTimeInMillis());
-                    second = 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)));
+                    // second = 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)));
+
+                    StringBuilder builder = new StringBuilder();
+                    builder.append(formatNumber((int) (ym / 13), 4))
+                        .append('-')
+                        .append(formatNumber((int) (ym % 13), 2))
+                        .append('-')
+                        .append(formatNumber((int) (ymd % (1 << 5)), 2))
+                        .append(' ')
+                        .append(formatNumber((int) (hms >> 12), 2))
+                        .append(':')
+                        .append(formatNumber((int) ((hms >> 6) % (1 << 6)), 2))
+                        .append(':')
+                        .append(formatNumber((int) (hms % (1 << 6)), 2));
+                    second = builder.toString();
                 }
 
                 if (meta >= 1) {
@@ -575,11 +603,22 @@ public final class RowsLogBuffer {
                     // cal.set(70, 0, 1, i32 / 10000, (i32 % 10000) / 100, i32 %
                     // 100);
                     // value = new Time(cal.getTimeInMillis());
-                    value = String.format("%s%02d:%02d:%02d",
-                        (i32 >= 0) ? "" : "-",
-                        u32 / 10000,
-                        (u32 % 10000) / 100,
-                        u32 % 100);
+                    // value = String.format("%s%02d:%02d:%02d",
+                    // (i32 >= 0) ? "" : "-",
+                    // u32 / 10000,
+                    // (u32 % 10000) / 100,
+                    // u32 % 100);
+
+                    StringBuilder builder = new StringBuilder();
+                    if (i32 < 0) {
+                        builder.append('-');
+                    }
+                    builder.append(formatNumber(u32 / 10000, 2))
+                        .append(':')
+                        .append(formatNumber((u32 % 10000) / 100, 2))
+                        .append(':')
+                        .append(formatNumber(u32 % 100, 2));
+                    value = builder.toString();
                 }
                 javaType = Types.TIME;
                 length = 3;
@@ -673,11 +712,22 @@ public final class RowsLogBuffer {
                     // value = new Time(cal.getTimeInMillis());
                     long ultime = Math.abs(ltime);
                     intpart = ultime >> 24;
-                    second = String.format("%s%02d:%02d:%02d",
-                        ltime >= 0 ? "" : "-",
-                        (int) ((intpart >> 12) % (1 << 10)),
-                        (int) ((intpart >> 6) % (1 << 6)),
-                        (int) (intpart % (1 << 6)));
+                    // second = String.format("%s%02d:%02d:%02d",
+                    // ltime >= 0 ? "" : "-",
+                    // (int) ((intpart >> 12) % (1 << 10)),
+                    // (int) ((intpart >> 6) % (1 << 6)),
+                    // (int) (intpart % (1 << 6)));
+
+                    StringBuilder builder = new StringBuilder();
+                    if (ltime < 0) {
+                        builder.append('-');
+                    }
+                    builder.append(formatNumber((int) ((intpart >> 12) % (1 << 10)), 2))
+                        .append(':')
+                        .append(formatNumber((int) ((intpart >> 6) % (1 << 6)), 2))
+                        .append(':')
+                        .append(formatNumber((int) (intpart % (1 << 6)), 2));
+                    second = builder.toString();
                 }
 
                 if (meta >= 1) {
@@ -717,7 +767,16 @@ public final class RowsLogBuffer {
                     // cal.set((i32 / (16 * 32)), (i32 / 32 % 16) - 1, (i32 %
                     // 32));
                     // value = new java.sql.Date(cal.getTimeInMillis());
-                    value = String.format("%04d-%02d-%02d", i32 / (16 * 32), i32 / 32 % 16, i32 % 32);
+                    // value = String.format("%04d-%02d-%02d", i32 / (16 * 32),
+                    // i32 / 32 % 16, i32 % 32);
+
+                    StringBuilder builder = new StringBuilder();
+                    builder.append(formatNumber(i32 / (16 * 32), 4))
+                        .append('-')
+                        .append(formatNumber(i32 / 32 % 16, 2))
+                        .append('-')
+                        .append(formatNumber(i32 % 32, 2));
+                    value = builder.toString();
                 }
                 javaType = Types.DATE;
                 length = 3;
@@ -970,7 +1029,8 @@ public final class RowsLogBuffer {
                         throw new IllegalArgumentException("!! Unknown JSON packlen = " + meta);
                 }
                 if (0 == len) {
-                    // fixed issue #1 by lava, json column of zero length has no value, value parsing should be skipped
+                    // fixed issue #1 by lava, json column of zero length has no
+                    // value, value parsing should be skipped
                     value = "";
                 } else {
                     int position = buffer.position();
@@ -1066,4 +1126,29 @@ public final class RowsLogBuffer {
         return sec.substring(0, meta);
     }
 
+    private String formatNumber(int d, int size) {
+        return leftPad(String.valueOf(d), size, '0');
+    }
+
+    private String leftPad(String str, int size, char padChar) {
+        if (str == null) {
+            return null;
+        }
+        int pads = size - str.length();
+        if (pads <= 0) {
+            return str; // returns original String when possible
+        }
+        return padding(pads, padChar).concat(str);
+    }
+
+    private String padding(int repeat, char padChar) throws IndexOutOfBoundsException {
+        if (repeat < 0) {
+            throw new IndexOutOfBoundsException("Cannot pad a negative amount: " + repeat);
+        }
+        final char[] buf = new char[repeat];
+        for (int i = 0; i < buf.length; i++) {
+            buf[i] = padChar;
+        }
+        return new String(buf);
+    }
 }

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

@@ -0,0 +1,16 @@
+package com.taobao.tddl.dbsync.binlog.event;
+
+import com.taobao.tddl.dbsync.binlog.LogBuffer;
+import com.taobao.tddl.dbsync.binlog.LogEvent;
+
+/**
+ * @author agapple 2018年5月7日 下午7:05:39
+ * @version 1.0.26
+ * @since mysql 5.7
+ */
+public class TransactionContextLogEvent extends LogEvent {
+
+    public TransactionContextLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){
+        super(header);
+    }
+}

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

@@ -0,0 +1,16 @@
+package com.taobao.tddl.dbsync.binlog.event;
+
+import com.taobao.tddl.dbsync.binlog.LogBuffer;
+import com.taobao.tddl.dbsync.binlog.LogEvent;
+
+/**
+ * @author agapple 2018年5月7日 下午7:05:39
+ * @version 1.0.26
+ * @since mysql 5.7
+ */
+public class ViewChangeEvent extends LogEvent {
+
+    public ViewChangeEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){
+        super(header);
+    }
+}

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

@@ -0,0 +1,65 @@
+package com.taobao.tddl.dbsync.binlog.event;
+
+import com.taobao.tddl.dbsync.binlog.LogBuffer;
+import com.taobao.tddl.dbsync.binlog.LogEvent;
+
+/**
+ * @author agapple 2018年5月7日 下午7:05:39
+ * @version 1.0.26
+ * @since mysql 5.7
+ */
+public class XaPrepareLogEvent extends LogEvent {
+
+    private boolean onePhase;
+    private int     formatId;
+    private int     gtridLength;
+    private int     bqualLength;
+    private byte[]  data;
+
+    public XaPrepareLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){
+        super(header);
+
+        final int commonHeaderLen = descriptionEvent.getCommonHeaderLen();
+        final int postHeaderLen = descriptionEvent.getPostHeaderLen()[header.getType() - 1];
+
+        int offset = commonHeaderLen + postHeaderLen;
+        buffer.position(offset);
+
+        onePhase = (buffer.getInt8() == 0x00 ? false : true);
+
+        formatId = buffer.getInt32();
+        gtridLength = buffer.getInt32();
+        bqualLength = buffer.getInt32();
+
+        int MY_XIDDATASIZE = 128;
+        if (MY_XIDDATASIZE >= gtridLength + bqualLength && gtridLength >= 0 && gtridLength <= 64 && bqualLength >= 0
+            && bqualLength <= 64) {
+            data = buffer.getData(gtridLength + bqualLength);
+        } else {
+            formatId = -1;
+            gtridLength = 0;
+            bqualLength = 0;
+        }
+    }
+
+    public boolean isOnePhase() {
+        return onePhase;
+    }
+
+    public int getFormatId() {
+        return formatId;
+    }
+
+    public int getGtridLength() {
+        return gtridLength;
+    }
+
+    public int getBqualLength() {
+        return bqualLength;
+    }
+
+    public byte[] getData() {
+        return data;
+    }
+
+}

dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/mariadb/StartEncryptionLogEvent.java

@@ -0,0 +1,19 @@
+package com.taobao.tddl.dbsync.binlog.event.mariadb;
+
+import com.taobao.tddl.dbsync.binlog.LogBuffer;
+import com.taobao.tddl.dbsync.binlog.LogEvent;
+import com.taobao.tddl.dbsync.binlog.event.FormatDescriptionLogEvent;
+import com.taobao.tddl.dbsync.binlog.event.LogHeader;
+
+/**
+ * mariadb的Start_encryption_log_event
+ * 
+ * @author agapple 2018年5月7日 下午7:23:02
+ * @version 1.0.26
+ */
+public class StartEncryptionLogEvent extends LogEvent {
+
+    public StartEncryptionLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){
+        super(header);
+    }
+}

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

@@ -454,6 +454,10 @@ public class CanalController {
         // 释放canal的工作节点
         releaseCid(ZookeeperPathUtils.getCanalClusterNode(ip + ":" + port));
         logger.info("## stop the canal server[{}:{}]", ip, port);
+        
+        if (zkclientx != null) {
+            zkclientx.close();
+        }
     }
 
     private void initCid(String path) {

deployer/src/main/resources/example/instance.properties

@@ -1,12 +1,15 @@
 #################################################
 ## mysql serverId
 canal.instance.mysql.slaveId=0
+
 # position info
 canal.instance.master.address=127.0.0.1:3306
+# enable gtid use true/false
+canal.instance.gtidon=false
 canal.instance.master.journal.name=
 canal.instance.master.position=
 canal.instance.master.timestamp=
-
+canal.instance.master.gtid=
 
 # table meta tsdb info
 canal.instance.tsdb.enable=true
@@ -20,7 +23,8 @@ canal.instance.tsdb.dbPassword=canal
 #canal.instance.standby.address =
 #canal.instance.standby.journal.name =
 #canal.instance.standby.position = 
-#canal.instance.standby.timestamp = 
+#canal.instance.standby.timestamp =
+#canal.instance.standby.gtid=
 # username/password
 canal.instance.dbUsername=canal
 canal.instance.dbPassword=canal

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

@@ -166,6 +166,7 @@
 				<property name="journalName" value="${canal.instance.master.journal.name}" />
 				<property name="position" value="${canal.instance.master.position}" />
 				<property name="timestamp" value="${canal.instance.master.timestamp}" />
+				<property name="gtid" value="${canal.instance.master.gtid}" />
 			</bean>
 		</property>
 		<property name="standbyPosition">
@@ -173,6 +174,7 @@
 				<property name="journalName" value="${canal.instance.standby.journal.name}" />
 				<property name="position" value="${canal.instance.standby.position}" />
 				<property name="timestamp" value="${canal.instance.standby.timestamp}" />
+				<property name="gtid" value="${canal.instance.standby.gtid}" />
 			</bean>
 		</property>
 		<property name="filterQueryDml" value="${canal.instance.filter.query.dml:false}" />
@@ -187,5 +189,8 @@
 		<!--表结构相关-->
 		<property name="enableTsdb" value="${canal.instance.tsdb.enable:true}"/>
 		<property name="tsdbSpringXml" value="${canal.instance.tsdb.spring.xml:}"/>
+		
+		<!--是否启用GTID模式-->
+		<property name="isGTIDMode" value="${canal.instance.gtidon}"/>
 	</bean>
 </beans>

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

@@ -151,6 +151,7 @@
 				<property name="journalName" value="${canal.instance.master.journal.name}" />
 				<property name="position" value="${canal.instance.master.position}" />
 				<property name="timestamp" value="${canal.instance.master.timestamp}" />
+				<property name="gtid" value="${canal.instance.master.gtid}" />
 			</bean>
 		</property>
 		<property name="standbyPosition">
@@ -158,6 +159,7 @@
 				<property name="journalName" value="${canal.instance.standby.journal.name}" />
 				<property name="position" value="${canal.instance.standby.position}" />
 				<property name="timestamp" value="${canal.instance.standby.timestamp}" />
+				<property name="gtid" value="${canal.instance.standby.gtid}" />
 			</bean>
 		</property>
 		<property name="filterQueryDml" value="${canal.instance.filter.query.dml:false}" />
@@ -172,5 +174,8 @@
 		<!--表结构相关-->
 		<property name="enableTsdb" value="${canal.instance.tsdb.enable:true}"/>
 		<property name="tsdbSpringXml" value="${canal.instance.tsdb.spring.xml:}"/>
+
+		<!--是否启用GTID模式-->
+		<property name="isGTIDMode" value="${canal.instance.gtidon}"/>
 	</bean>
 </beans>

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

@@ -148,6 +148,7 @@
 				<property name="journalName" value="${canal.instance.master1.journal.name}" />
 				<property name="position" value="${canal.instance.master1.position}" />
 				<property name="timestamp" value="${canal.instance.master1.timestamp}" />
+				<property name="gtid" value="${canal.instance.master1.gtid}" />
 			</bean>
 		</property>
 		<property name="standbyPosition">
@@ -155,6 +156,7 @@
 				<property name="journalName" value="${canal.instance.standby1.journal.name}" />
 				<property name="position" value="${canal.instance.standby1.position}" />
 				<property name="timestamp" value="${canal.instance.standby1.timestamp}" />
+				<property name="gtid" value="${canal.instance.standby1.gtid}" />
 			</bean>
 		</property>
 		<property name="filterQueryDml" value="${canal.instance.filter.query.dml:false}" />
@@ -239,6 +241,7 @@
 				<property name="journalName" value="${canal.instance.master2.journal.name}" />
 				<property name="position" value="${canal.instance.master2.position}" />
 				<property name="timestamp" value="${canal.instance.master2.timestamp}" />
+				<property name="gtid" value="${canal.instance.master2.gtid}" />
 			</bean>
 		</property>
 		<property name="standbyPosition">
@@ -246,6 +249,7 @@
 				<property name="journalName" value="${canal.instance.standby2.journal.name}" />
 				<property name="position" value="${canal.instance.standby2.position}" />
 				<property name="timestamp" value="${canal.instance.standby2.timestamp}" />
+				<property name="gtid" value="${canal.instance.standby2.gtid}" />
 			</bean>
 		</property>
 		<property name="filterQueryDml" value="${canal.instance.filter.query.dml:false}" />

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

@@ -139,6 +139,7 @@
 				<property name="journalName" value="${canal.instance.master.journal.name}" />
 				<property name="position" value="${canal.instance.master.position}" />
 				<property name="timestamp" value="${canal.instance.master.timestamp}" />
+				<property name="gtid" value="${canal.instance.master.gtid}" />
 			</bean>
 		</property>
 		<property name="standbyPosition">
@@ -146,6 +147,7 @@
 				<property name="journalName" value="${canal.instance.standby.journal.name}" />
 				<property name="position" value="${canal.instance.standby.position}" />
 				<property name="timestamp" value="${canal.instance.standby.timestamp}" />
+				<property name="gtid" value="${canal.instance.standby.gtid}" />
 			</bean>
 		</property>
 		<property name="filterQueryDml" value="${canal.instance.filter.query.dml:false}" />
@@ -160,5 +162,8 @@
 		<!--表结构相关-->
 		<property name="enableTsdb" value="${canal.instance.tsdb.enable:false}"/>
 		<property name="tsdbSpringXml" value="${canal.instance.tsdb.spring.xml:}"/>
+		
+		<!--是否启用GTID模式-->
+		<property name="isGTIDMode" value="${canal.instance.gtidon}"/>
 	</bean>
 </beans>

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

@@ -0,0 +1,24 @@
+package com.alibaba.otter.canal.parse.driver.mysql.packets;
+
+import java.io.IOException;
+
+/**
+ * Created by hiwjd on 2018/4/23.
+ * hiwjd0@gmail.com
+ */
+public interface GTIDSet {
+
+    /**
+     * 序列化成字节数组
+     *
+     * @return
+     */
+    byte[] encode() throws IOException;
+
+    /**
+     * 更新当前实例
+     * @param str
+     * @throws Exception
+     */
+    void update(String str);
+}

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

@@ -0,0 +1,151 @@
+package com.alibaba.otter.canal.parse.driver.mysql.packets;
+
+import com.alibaba.otter.canal.parse.driver.mysql.utils.ByteHelper;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.*;
+
+/**
+ * Created by hiwjd on 2018/4/23.
+ * hiwjd0@gmail.com
+ */
+public class MysqlGTIDSet implements GTIDSet {
+
+    public Map<String, UUIDSet> sets;
+
+    @Override
+    public byte[] encode() throws IOException {
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        ByteHelper.writeUnsignedInt64LittleEndian(sets.size(), out);
+
+        for (Map.Entry<String, UUIDSet> entry : sets.entrySet()) {
+            out.write(entry.getValue().encode());
+        }
+
+        return out.toByteArray();
+    }
+
+    @Override
+    public void update(String str) {
+        UUIDSet us = UUIDSet.parse(str);
+        String sid = us.SID.toString();
+        if (sets.containsKey(sid)) {
+            sets.get(sid).intervals.addAll(us.intervals);
+            sets.get(sid).intervals = UUIDSet.combine(sets.get(sid).intervals);
+        } else {
+            sets.put(sid, us);
+        }
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (o == null) return false;
+        if (this == o) return true;
+
+        MysqlGTIDSet gs = (MysqlGTIDSet) o;
+        if (gs.sets == null) return false;
+
+        for (Map.Entry<String, UUIDSet> entry : sets.entrySet()) {
+            if (!entry.getValue().equals(gs.sets.get(entry.getKey()))) {
+                return false;
+            }
+        }
+
+        return true;
+    }
+
+    /**
+     * 解析如下格式的字符串为MysqlGTIDSet:
+     *
+     * 726757ad-4455-11e8-ae04-0242ac110002:1 =>
+     *   MysqlGTIDSet{
+     *     sets: {
+     *       726757ad-4455-11e8-ae04-0242ac110002: UUIDSet{
+     *         SID: 726757ad-4455-11e8-ae04-0242ac110002,
+     *         intervals: [{start:1, stop:2}]
+     *       }
+     *     }
+     *   }
+     *
+     * 726757ad-4455-11e8-ae04-0242ac110002:1-3 =>
+     *   MysqlGTIDSet{
+     *     sets: {
+     *       726757ad-4455-11e8-ae04-0242ac110002: UUIDSet{
+     *         SID: 726757ad-4455-11e8-ae04-0242ac110002,
+     *         intervals: [{start:1, stop:4}]
+     *       }
+     *     }
+     *   }
+     *
+     * 726757ad-4455-11e8-ae04-0242ac110002:1-3:4 =>
+     *   MysqlGTIDSet{
+     *     sets: {
+     *       726757ad-4455-11e8-ae04-0242ac110002: UUIDSet{
+     *         SID: 726757ad-4455-11e8-ae04-0242ac110002,
+     *         intervals: [{start:1, stop:5}]
+     *       }
+     *     }
+     *   }
+     *
+     * 726757ad-4455-11e8-ae04-0242ac110002:1-3:7-9 =>
+     *   MysqlGTIDSet{
+     *     sets: {
+     *       726757ad-4455-11e8-ae04-0242ac110002: UUIDSet{
+     *         SID: 726757ad-4455-11e8-ae04-0242ac110002,
+     *         intervals: [{start:1, stop:4}, {start:7, stop: 10}]
+     *       }
+     *     }
+     *   }
+     *
+     * 726757ad-4455-11e8-ae04-0242ac110002:1-3,726757ad-4455-11e8-ae04-0242ac110003:4 =>
+     *   MysqlGTIDSet{
+     *     sets: {
+     *       726757ad-4455-11e8-ae04-0242ac110002: UUIDSet{
+     *         SID: 726757ad-4455-11e8-ae04-0242ac110002,
+     *         intervals: [{start:1, stop:4}]
+     *       },
+     *       726757ad-4455-11e8-ae04-0242ac110003: UUIDSet{
+     *         SID: 726757ad-4455-11e8-ae04-0242ac110002,
+     *         intervals: [{start:4, stop:5}]
+     *       }
+     *     }
+     *   }
+     *
+     * @param gtidData
+     * @return
+     */
+    public static MysqlGTIDSet parse(String gtidData) {
+        Map<String, UUIDSet> m;
+
+        if (gtidData == null || gtidData.length() < 1) {
+            m = new HashMap<String, UUIDSet>();
+        } else {
+            String[] uuidStrs = gtidData.split(",");
+            m = new HashMap<String, UUIDSet>(uuidStrs.length);
+            for (int i = 0; i < uuidStrs.length; i++) {
+                UUIDSet uuidSet = UUIDSet.parse(uuidStrs[i]);
+                m.put(uuidSet.SID.toString(), uuidSet);
+            }
+        }
+
+        MysqlGTIDSet gs = new MysqlGTIDSet();
+        gs.sets = m;
+
+        return gs;
+    }
+
+    @Override
+    public String toString() {
+        StringBuilder sb = new StringBuilder();
+
+        for (Map.Entry<String, UUIDSet> entry : sets.entrySet()) {
+            if (sb.length() > 0) {
+                sb.append(",");
+            }
+            sb.append(entry.getValue().toString());
+        }
+
+        return sb.toString();
+    }
+}

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

@@ -0,0 +1,202 @@
+package com.alibaba.otter.canal.parse.driver.mysql.packets;
+
+import com.alibaba.otter.canal.parse.driver.mysql.utils.ByteHelper;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+
+/**
+ * Created by hiwjd on 2018/4/23.
+ * hiwjd0@gmail.com
+ */
+public class UUIDSet {
+
+    public UUID SID;
+    public List<Interval> intervals;
+
+    public byte[] encode() throws IOException {
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+
+        ByteBuffer bb = ByteBuffer.wrap(new byte[16]);
+        bb.putLong(SID.getMostSignificantBits());
+        bb.putLong(SID.getLeastSignificantBits());
+
+        out.write(bb.array());
+
+        ByteHelper.writeUnsignedInt64LittleEndian(intervals.size(), out);
+
+        for (Interval interval : intervals) {
+            ByteHelper.writeUnsignedInt64LittleEndian(interval.start, out);
+            ByteHelper.writeUnsignedInt64LittleEndian(interval.stop, out);
+        }
+
+        return out.toByteArray();
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (o == null) return false;
+        if (this == o) return true;
+
+        UUIDSet us = (UUIDSet) o;
+        Collections.sort(intervals);
+        Collections.sort(us.intervals);
+        if (SID.equals(us.SID) && intervals.equals(us.intervals)) {
+            return true;
+        }
+
+        return false;
+    }
+
+    public static class Interval implements Comparable<Interval> {
+        public long start;
+        public long stop;
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+
+            Interval interval = (Interval) o;
+
+            if (start != interval.start) return false;
+            return stop == interval.stop;
+        }
+
+        @Override
+        public int hashCode() {
+            int result = (int) (start ^ (start >>> 32));
+            result = 31 * result + (int) (stop ^ (stop >>> 32));
+            return result;
+        }
+
+        @Override
+        public int compareTo(Interval o) {
+            if (equals(o)) {
+                return 1;
+            }
+            return start > o.start ? 1 : (start == o.start ? 0 : -1);
+        }
+    }
+
+    /**
+     * 解析如下格式字符串为UUIDSet:
+     *
+     * 726757ad-4455-11e8-ae04-0242ac110002:1 =>
+     *   UUIDSet{SID: 726757ad-4455-11e8-ae04-0242ac110002, intervals: [{start:1, stop:2}]}
+     *
+     * 726757ad-4455-11e8-ae04-0242ac110002:1-3 =>
+     *   UUIDSet{SID: 726757ad-4455-11e8-ae04-0242ac110002, intervals: [{start:1, stop:4}]}
+     *
+     * 726757ad-4455-11e8-ae04-0242ac110002:1-3:4
+     *   UUIDSet{SID: 726757ad-4455-11e8-ae04-0242ac110002, intervals: [{start:1, stop:5}]}
+     *
+     * 726757ad-4455-11e8-ae04-0242ac110002:1-3:7-9
+     *   UUIDSet{SID: 726757ad-4455-11e8-ae04-0242ac110002, intervals: [{start:1, stop:4}, {start:7, stop:10}]}
+     *
+     * @param str
+     * @return
+     */
+    public static UUIDSet parse(String str) {
+        String[] ss = str.split(":");
+
+        if (ss.length < 2) {
+            throw new RuntimeException(String.format("parseUUIDSet failed due to wrong format: %s", str));
+        }
+
+        List<Interval> intervals = new ArrayList<Interval>();
+        for (int i=1; i<ss.length; i++) {
+            intervals.add(parseInterval(ss[i]));
+        }
+
+        UUIDSet uuidSet = new UUIDSet();
+        uuidSet.SID = UUID.fromString(ss[0]);
+        uuidSet.intervals = combine(intervals);
+
+        return uuidSet;
+    }
+
+    @Override
+    public String toString() {
+        StringBuilder sb = new StringBuilder();
+
+        sb.append(SID.toString());
+        for (Interval interval : intervals) {
+            if (interval.start == interval.stop - 1) {
+                sb.append(":");
+                sb.append(interval.start);
+            } else {
+                sb.append(":");
+                sb.append(interval.start);
+                sb.append("-");
+                sb.append(interval.stop-1);
+            }
+        }
+
+        return sb.toString();
+    }
+
+    /**
+     * 解析如下格式字符串为Interval:
+     *
+     * 1 => Interval{start:1, stop:2}
+     * 1-3 => Interval{start:1, stop:4}
+     *
+     * 注意！字符串格式表达时[n,m]是两侧都包含的，Interval表达时[n,m)右侧开
+     *
+     * @param str
+     * @return
+     */
+    public static Interval parseInterval(String str) {
+        String[] ss = str.split("-");
+
+        Interval interval = new Interval();
+        switch (ss.length) {
+            case 1:
+                interval.start = Long.parseLong(ss[0]);
+                interval.stop = interval.start + 1;
+                break;
+            case 2:
+                interval.start = Long.parseLong(ss[0]);
+                interval.stop = Long.parseLong(ss[1]) + 1;
+                break;
+            default:
+                throw new RuntimeException(String.format("parseInterval failed due to wrong format: %s", str));
+        }
+
+        return interval;
+    }
+
+    /**
+     * 把{start,stop}连续的合并掉:
+     * [{start:1, stop:4},{start:4, stop:5}] => [{start:1, stop:5}]
+     *
+     * @param intervals
+     * @return
+     */
+    public static List<Interval> combine(List<Interval> intervals) {
+        List<Interval> combined = new ArrayList<Interval>();
+        Collections.sort(intervals);
+        int len = intervals.size();
+        for (int i=0; i<len; i++) {
+            combined.add(intervals.get(i));
+
+            int j;
+            for (j=i+1; j<len; j++) {
+                if (intervals.get(i).stop >= intervals.get(j).start) {
+                    intervals.get(i).stop = intervals.get(j).stop;
+                } else {
+                    break;
+                }
+            }
+            i = j - 1;
+        }
+
+        return combined;
+    }
+}

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

@@ -0,0 +1,68 @@
+package com.alibaba.otter.canal.parse.driver.mysql.packets.client;
+
+import com.alibaba.otter.canal.parse.driver.mysql.packets.CommandPacket;
+import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet;
+import com.alibaba.otter.canal.parse.driver.mysql.utils.ByteHelper;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+
+/**
+ * Created by hiwjd on 2018/4/24.
+ * hiwjd0@gmail.com
+ *
+ * https://dev.mysql.com/doc/internals/en/com-binlog-dump-gtid.html
+ */
+public class BinlogDumpGTIDCommandPacket extends CommandPacket {
+
+    public static final int BINLOG_DUMP_NON_BLOCK = 0x01;
+    public static final int BINLOG_THROUGH_POSITION = 0x02;
+    public static final int BINLOG_THROUGH_GTID = 0x04;
+
+    public long    slaveServerId;
+    public GTIDSet gtidSet;
+
+    public BinlogDumpGTIDCommandPacket() {
+        setCommand((byte) 0x1e);
+    }
+
+    @Override
+    public void fromBytes(byte[] data) throws IOException {
+    }
+
+    @Override
+    public byte[] toBytes() throws IOException {
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+
+        // 0. [1] write command number
+        out.write(getCommand());
+        // 1. [2] flags
+        ByteHelper.writeUnsignedShortLittleEndian(BINLOG_THROUGH_GTID, out);
+        // 2. [4] server-id
+        ByteHelper.writeUnsignedIntLittleEndian(slaveServerId, out);
+        // 3. [4] binlog-filename-len
+        ByteHelper.writeUnsignedIntLittleEndian(0, out);
+        // 4. [] binlog-filename
+        // skip
+        // 5. [8] binlog-pos
+        ByteHelper.writeUnsignedInt64LittleEndian(4, out);
+        // if flags & BINLOG_THROUGH_GTID {
+        byte[] bs = gtidSet.encode();
+        // 6. [4] data-size
+        ByteHelper.writeUnsignedIntLittleEndian(bs.length, out);
+        // 7, [] data
+        //       [8] n_sids // 文档写的是4个字节，其实是8个字节
+        //       for n_sids {
+        //          [16] SID
+        //          [8] n_intervals
+        //          for n_intervals {
+        //             [8] start (signed)
+        //             [8] end (signed)
+        //          }
+        //       }
+        out.write(bs);
+        // }
+
+        return out.toByteArray();
+    }
+}

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

@@ -127,6 +127,17 @@ public abstract class ByteHelper {
         out.write((byte) (data >>> 24));
     }
 
+    public static void writeUnsignedInt64LittleEndian(long data, ByteArrayOutputStream out) {
+        out.write((byte) (data & 0xFF));
+        out.write((byte) (data >>> 8));
+        out.write((byte) (data >>> 16));
+        out.write((byte) (data >>> 24));
+        out.write((byte) (data >>> 32));
+        out.write((byte) (data >>> 40));
+        out.write((byte) (data >>> 48));
+        out.write((byte) (data >>> 56));
+    }
+
     public static void writeUnsignedShortLittleEndian(int data, ByteArrayOutputStream out) {
         out.write((byte) (data & 0xFF));
         out.write((byte) ((data >>> 8) & 0xFF));

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

@@ -0,0 +1,129 @@
+package com.alibaba.otter.canal.parse.driver.mysql;
+
+import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet;
+import com.alibaba.otter.canal.parse.driver.mysql.packets.MysqlGTIDSet;
+import com.alibaba.otter.canal.parse.driver.mysql.packets.UUIDSet;
+import org.junit.Test;
+import static org.junit.Assert.*;
+
+import java.io.IOException;
+import java.util.*;
+
+/**
+ * Created by hiwjd on 2018/4/25.
+ * hiwjd0@gmail.com
+ */
+public class MysqlGTIDSetTest {
+
+    @Test
+    public void testEncode() throws IOException {
+        GTIDSet gtidSet = MysqlGTIDSet.parse("726757ad-4455-11e8-ae04-0242ac110002:1");
+        byte[] bytes = gtidSet.encode();
+
+        byte[] expected = {
+                0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x72, 0x67, 0x57, (byte)0xad,
+                0x44, 0x55, 0x11, (byte)0xe8, (byte)0xae, 0x04, 0x02, 0x42, (byte)0xac, 0x11, 0x00, 0x02,
+                0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00,
+                0x00, 0x00, 0x00, 0x00, 0x02, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00
+        };
+
+        for (int i=0; i<bytes.length; i++) {
+            assertEquals(expected[i], bytes[i]);
+        }
+    }
+
+    @Test
+    public void testParse() {
+        Map<String, MysqlGTIDSet> cases = new HashMap<String, MysqlGTIDSet>(5);
+        cases.put(
+                "726757ad-4455-11e8-ae04-0242ac110002:1",
+                buildForTest(new Material("726757ad-4455-11e8-ae04-0242ac110002", 1, 2))
+        );
+        cases.put(
+                "726757ad-4455-11e8-ae04-0242ac110002:1-3",
+                buildForTest(new Material("726757ad-4455-11e8-ae04-0242ac110002", 1, 4))
+        );
+        cases.put(
+                "726757ad-4455-11e8-ae04-0242ac110002:1-3:4",
+                buildForTest(new Material("726757ad-4455-11e8-ae04-0242ac110002", 1, 5))
+        );
+        cases.put(
+                "726757ad-4455-11e8-ae04-0242ac110002:1-3:7-9",
+                buildForTest(new Material("726757ad-4455-11e8-ae04-0242ac110002", 1, 4, 7, 10))
+        );
+        cases.put(
+                "726757ad-4455-11e8-ae04-0242ac110002:1-3,726757ad-4455-11e8-ae04-0242ac110003:4",
+                buildForTest(
+                        Arrays.asList(
+                            new Material("726757ad-4455-11e8-ae04-0242ac110002", 1, 4),
+                                new Material("726757ad-4455-11e8-ae04-0242ac110003", 4, 5)
+                        )
+                )
+        );
+
+        for (Map.Entry<String, MysqlGTIDSet> entry : cases.entrySet()) {
+            MysqlGTIDSet expected = entry.getValue();
+            MysqlGTIDSet actual = MysqlGTIDSet.parse(entry.getKey());
+
+            assertEquals(expected, actual);
+        }
+    }
+
+    private static class Material {
+
+        public Material(String uuid, long start, long stop) {
+            this.uuid = uuid;
+            this.start = start;
+            this.stop = stop;
+            this.start1 = 0;
+            this.stop1 = 0;
+        }
+
+        public Material(String uuid, long start, long stop, long start1, long stop1) {
+            this.uuid = uuid;
+            this.start = start;
+            this.stop = stop;
+            this.start1 = start1;
+            this.stop1 = stop1;
+        }
+
+        public String uuid;
+        public long start;
+        public long stop;
+        public long start1;
+        public long stop1;
+    }
+
+    private MysqlGTIDSet buildForTest(Material material) {
+        return buildForTest(Arrays.asList(material));
+    }
+
+    private MysqlGTIDSet buildForTest(List<Material> materials) {
+        Map<String, UUIDSet> sets = new HashMap<String, UUIDSet>();
+        for (Material a : materials) {
+            UUIDSet.Interval interval = new UUIDSet.Interval();
+            interval.start = a.start;
+            interval.stop = a.stop;
+            List<UUIDSet.Interval> intervals = new ArrayList<UUIDSet.Interval>();
+            intervals.add(interval);
+
+            if (a.start1 > 0 && a.stop1 > 0) {
+                UUIDSet.Interval interval1 = new UUIDSet.Interval();
+                interval1.start = a.start1;
+                interval1.stop = a.stop1;
+                intervals.add(interval1);
+            }
+
+            UUIDSet us = new UUIDSet();
+            us.SID = UUID.fromString(a.uuid);
+            us.intervals = intervals;
+
+            sets.put(a.uuid, us);
+        }
+
+        MysqlGTIDSet gs = new MysqlGTIDSet();
+        gs.sets = sets;
+
+        return gs;
+    }
+}

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

@@ -0,0 +1,30 @@
+package com.alibaba.otter.canal.parse.driver.mysql;
+
+import com.alibaba.otter.canal.parse.driver.mysql.packets.UUIDSet;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.*;
+
+/**
+ * Created by hiwjd on 2018/4/26.
+ * hiwjd0@gmail.com
+ */
+public class UUIDSetTest {
+
+    @Test
+    public void testToString() {
+        Map<String, String> cases = new HashMap<String, String>(4);
+        cases.put("726757ad-4455-11e8-ae04-0242ac110002:1", "726757ad-4455-11e8-ae04-0242ac110002:1");
+        cases.put("726757ad-4455-11e8-ae04-0242ac110002:1-3", "726757ad-4455-11e8-ae04-0242ac110002:1-3");
+        cases.put("726757ad-4455-11e8-ae04-0242ac110002:1-3:4-6", "726757ad-4455-11e8-ae04-0242ac110002:1-6");
+        cases.put("726757ad-4455-11e8-ae04-0242ac110002:1-3:5-7", "726757ad-4455-11e8-ae04-0242ac110002:1-3:5-7");
+
+        for (Map.Entry<String, String> entry : cases.entrySet()) {
+            String expected = entry.getValue();
+            assertEquals(expected, UUIDSet.parse(entry.getKey()).toString());
+        }
+    }
+}

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

@@ -4,6 +4,7 @@ import java.text.SimpleDateFormat;
 import java.util.Date;
 import java.util.List;
 
+import org.apache.commons.lang.StringUtils;
 import org.apache.commons.lang.SystemUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -12,10 +13,12 @@ import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 
 import com.alibaba.otter.canal.client.CanalConnector;
+import com.alibaba.otter.canal.client.impl.ClusterCanalConnector;
 import com.alibaba.otter.canal.protocol.CanalEntry.Column;
 import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
 import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
 import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
+import com.alibaba.otter.canal.protocol.CanalEntry.Pair;
 import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
 import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
 import com.alibaba.otter.canal.protocol.CanalEntry.TransactionBegin;
@@ -35,6 +38,7 @@ public class AbstractCanalClientTest {
     protected static final String             SEP                = SystemUtils.LINE_SEPARATOR;
     protected static final String             DATE_FORMAT        = "yyyy-MM-dd HH:mm:ss";
     protected volatile boolean                running            = false;
+    private volatile boolean                  waiting            = true;
     protected Thread.UncaughtExceptionHandler handler            = new Thread.UncaughtExceptionHandler() {
 
                                                                      public void uncaughtException(Thread t, Throwable e) {
@@ -56,10 +60,12 @@ public class AbstractCanalClientTest {
         context_format += "****************************************************" + SEP;
 
         row_format = SEP
-                     + "----------------> binlog[{}:{}] , name[{},{}] , eventType : {} , executeTime : {}({}) , delay : {} ms"
+                     + "----------------> binlog[{}:{}] , name[{},{}] , eventType : {} , executeTime : {}({}) , gtid : ({}) , delay : {} ms"
                      + SEP;
 
-        transaction_format = SEP + "================> binlog[{}:{}] , executeTime : {}({}) , delay : {}ms" + SEP;
+        transaction_format = SEP
+                             + "================> binlog[{}:{}] , executeTime : {}({}) , gtid : ({}) , delay : {}ms"
+                             + SEP;
 
     }
 
@@ -82,8 +88,8 @@ public class AbstractCanalClientTest {
         });
 
         thread.setUncaughtExceptionHandler(handler);
-        thread.start();
         running = true;
+        thread.start();
     }
 
     protected void stop() {
@@ -91,6 +97,12 @@ public class AbstractCanalClientTest {
             return;
         }
         running = false;
+        if (waiting) {
+            if (connector instanceof ClusterCanalConnector) {
+                ((ClusterCanalConnector) connector).setRetryTimes(-1);
+            }
+            thread.interrupt();
+        }
         if (thread != null) {
             try {
                 thread.join();
@@ -109,6 +121,7 @@ public class AbstractCanalClientTest {
                 MDC.put("destination", destination);
                 connector.connect();
                 connector.subscribe();
+                waiting = false;
                 while (running) {
                     Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                     long batchId = message.getId();
@@ -157,8 +170,12 @@ public class AbstractCanalClientTest {
         long time = entry.getHeader().getExecuteTime();
         Date date = new Date(time);
         SimpleDateFormat format = new SimpleDateFormat(DATE_FORMAT);
-        return entry.getHeader().getLogfileName() + ":" + entry.getHeader().getLogfileOffset() + ":"
-               + entry.getHeader().getExecuteTime() + "(" + format.format(date) + ")";
+        String position = entry.getHeader().getLogfileName() + ":" + entry.getHeader().getLogfileOffset() + ":"
+                          + entry.getHeader().getExecuteTime() + "(" + format.format(date) + ")";
+        if (StringUtils.isNotEmpty(entry.getHeader().getGtid())) {
+            position += " gtid(" + entry.getHeader().getGtid() + ")";
+        }
+        return position;
     }
 
     protected void printEntry(List<Entry> entrys) {
@@ -181,8 +198,9 @@ public class AbstractCanalClientTest {
                         new Object[] { entry.getHeader().getLogfileName(),
                                 String.valueOf(entry.getHeader().getLogfileOffset()),
                                 String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date),
-                                String.valueOf(delayTime) });
+                                entry.getHeader().getGtid(), String.valueOf(delayTime) });
                     logger.info(" BEGIN ----> Thread id: {}", begin.getThreadId());
+                    printXAInfo(begin.getPropsList());
                 } else if (entry.getEntryType() == EntryType.TRANSACTIONEND) {
                     TransactionEnd end = null;
                     try {
@@ -193,11 +211,12 @@ public class AbstractCanalClientTest {
                     // 打印事务提交信息，事务id
                     logger.info("----------------\n");
                     logger.info(" END ----> transaction id: {}", end.getTransactionId());
+                    printXAInfo(end.getPropsList());
                     logger.info(transaction_format,
                         new Object[] { entry.getHeader().getLogfileName(),
                                 String.valueOf(entry.getHeader().getLogfileOffset()),
                                 String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date),
-                                String.valueOf(delayTime) });
+                                entry.getHeader().getGtid(), String.valueOf(delayTime) });
                 }
 
                 continue;
@@ -218,13 +237,14 @@ public class AbstractCanalClientTest {
                             String.valueOf(entry.getHeader().getLogfileOffset()), entry.getHeader().getSchemaName(),
                             entry.getHeader().getTableName(), eventType,
                             String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date),
-                            String.valueOf(delayTime) });
+                            entry.getHeader().getGtid(), String.valueOf(delayTime) });
 
                 if (eventType == EventType.QUERY || rowChage.getIsDdl()) {
                     logger.info(" sql ----> " + rowChage.getSql() + SEP);
                     continue;
                 }
 
+                printXAInfo(rowChage.getPropsList());
                 for (RowData rowData : rowChage.getRowDatasList()) {
                     if (eventType == EventType.DELETE) {
                         printColumn(rowData.getBeforeColumnsList());
@@ -251,6 +271,27 @@ public class AbstractCanalClientTest {
         }
     }
 
+    protected void printXAInfo(List<Pair> pairs) {
+        if (pairs == null) {
+            return;
+        }
+
+        String xaType = null;
+        String xaXid = null;
+        for (Pair pair : pairs) {
+            String key = pair.getKey();
+            if (StringUtils.endsWithIgnoreCase(key, "XA_TYPE")) {
+                xaType = pair.getValue();
+            } else if (StringUtils.endsWithIgnoreCase(key, "XA_XID")) {
+                xaXid = pair.getValue();
+            }
+        }
+
+        if (xaType != null && xaXid != null) {
+            logger.info(" ------> " + xaType + " " + xaXid);
+        }
+    }
+
     public void setConnector(CanalConnector connector) {
         this.connector = connector;
     }

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

@@ -263,7 +263,7 @@ public class CanalInstanceWithManager extends AbstractCanalInstance {
                 mysqlEventParser.setMasterPosition(masterPosition);
 
                 if (parameters.getPositions().size() > 1) {
-                    EntryPosition standbyPosition = JsonUtils.unmarshalFromString(parameters.getPositions().get(0),
+                    EntryPosition standbyPosition = JsonUtils.unmarshalFromString(parameters.getPositions().get(1),
                         EntryPosition.class);
                     mysqlEventParser.setStandbyPosition(standbyPosition);
                 }

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

@@ -109,10 +109,11 @@ public class FileMixedMetaManager extends MemoryMetaManager implements CanalMeta
                         // 定时将内存中的最新值刷到file中，多次变更只刷一次
                         if (logger.isInfoEnabled()) {
                             LogPosition cursor = (LogPosition) getCursor(clientIdentity);
-                            logger.info("clientId:{} cursor:[{},{},{}] address[{}]",
-                                new Object[] { clientIdentity.getClientId(), cursor.getPostion().getJournalName(),
-                                        cursor.getPostion().getPosition(), cursor.getPostion().getTimestamp(),
-                                        cursor.getIdentity().getSourceAddress().toString() });
+                            logger.info("clientId:{} cursor:[{},{},{},{},{}] address[{}]", new Object[] {
+                                    clientIdentity.getClientId(), cursor.getPostion().getJournalName(),
+                                    cursor.getPostion().getPosition(), cursor.getPostion().getTimestamp(),
+                                    cursor.getPostion().getServerId(), cursor.getPostion().getGtid(),
+                                    cursor.getIdentity().getSourceAddress().toString() });
                         }
                         flushDataToFile(clientIdentity.getDestination());
                         updateCursorTasks.remove(clientIdentity);

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

@@ -83,7 +83,8 @@ public class MemoryMetaManager extends AbstractCanalLifeCycle implements CanalMe
     }
 
     public synchronized List<ClientIdentity> listAllSubscribeInfo(String destination) throws CanalMetaManagerException {
-        return destinations.get(destination);
+        // fixed issue #657, fixed ConcurrentModificationException
+        return Lists.newArrayList(destinations.get(destination));
     }
 
     public Position getCursor(ClientIdentity clientIdentity) throws CanalMetaManagerException {

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

@@ -19,6 +19,7 @@ import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
 import com.alibaba.otter.canal.common.alarm.CanalAlarmHandler;
 import com.alibaba.otter.canal.filter.CanalEventFilter;
 import com.alibaba.otter.canal.parse.CanalEventParser;
+import com.alibaba.otter.canal.parse.driver.mysql.packets.MysqlGTIDSet;
 import com.alibaba.otter.canal.parse.exception.CanalParseException;
 import com.alibaba.otter.canal.parse.exception.TableIdNotFoundException;
 import com.alibaba.otter.canal.parse.inbound.EventTransactionBuffer.TransactionFlushCallback;
@@ -87,6 +88,8 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
     protected TimerTask                              heartBeatTimerTask;
     protected Throwable                              exception                  = null;
 
+    protected boolean                                isGTIDMode                 = false; // 是否是GTID模式
+
     protected abstract BinlogParser buildParser();
 
     protected abstract ErosaConnection buildErosaConnection();
@@ -214,12 +217,17 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
                         };
 
                         // 4. 开始dump数据
-                        if (StringUtils.isEmpty(startPosition.getJournalName()) && startPosition.getTimestamp() != null) {
-                            erosaConnection.dump(startPosition.getTimestamp(), sinkHandler);
+                        // 判断所属instance是否启用GTID模式，是的话调用ErosaConnection中GTID对应方法dump数据
+                        if (isGTIDMode()) {
+                            erosaConnection.dump(MysqlGTIDSet.parse(startPosition.getGtid()), sinkHandler);
                         } else {
-                            erosaConnection.dump(startPosition.getJournalName(),
-                                startPosition.getPosition(),
-                                sinkHandler);
+                            if (StringUtils.isEmpty(startPosition.getJournalName()) && startPosition.getTimestamp() != null) {
+                                erosaConnection.dump(startPosition.getTimestamp(), sinkHandler);
+                            } else {
+                                erosaConnection.dump(startPosition.getJournalName(),
+                                        startPosition.getPosition(),
+                                        sinkHandler);
+                            }
                         }
 
                     } catch (TableIdNotFoundException e) {
@@ -368,6 +376,9 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
         position.setTimestamp(entry.getHeader().getExecuteTime());
         // add serverId at 2016-06-28
         position.setServerId(entry.getHeader().getServerId());
+        // set gtid
+        position.setGtid(entry.getHeader().getGtid());
+
         logPosition.setPostion(position);
 
         LogIdentity identity = new LogIdentity(runningInfo.getAddress(), -1L);
@@ -530,4 +541,11 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
         return exception;
     }
 
+    public boolean isGTIDMode() {
+        return isGTIDMode;
+    }
+
+    public void setIsGTIDMode(boolean isGTIDMode) {
+        this.isGTIDMode = isGTIDMode;
+    }
 }

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

@@ -1,5 +1,7 @@
 package com.alibaba.otter.canal.parse.inbound;
 
+import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet;
+
 import java.io.IOException;
 
 /**
@@ -24,5 +26,14 @@ public interface ErosaConnection {
 
     public void dump(long timestamp, SinkFunction func) throws IOException;
 
+    /**
+     * 通过GTID同步binlog
+     *
+     * @param gtidSet
+     * @param func
+     * @throws IOException
+     */
+    public void dump(GTIDSet gtidSet, SinkFunction func) throws IOException;
+
     ErosaConnection fork();
 }

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

@@ -7,6 +7,7 @@ import org.slf4j.LoggerFactory;
 
 import com.alibaba.otter.canal.filter.CanalEventFilter;
 import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter;
+import com.alibaba.otter.canal.parse.driver.mysql.packets.MysqlGTIDSet;
 import com.alibaba.otter.canal.parse.exception.CanalParseException;
 import com.alibaba.otter.canal.parse.inbound.AbstractEventParser;
 import com.alibaba.otter.canal.parse.inbound.BinlogParser;
@@ -68,7 +69,18 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
      * @return
      */
     protected boolean processTableMeta(EntryPosition position) {
+        if (isGTIDMode()) {
+            if (binlogParser instanceof LogEventConvert) {
+                // 记录gtid
+                ((LogEventConvert) binlogParser).setGtidSet(MysqlGTIDSet.parse(position.getGtid()));
+            }
+        }
+
         if (tableMetaTSDB != null) {
+            if (position.getTimestamp() == null || position.getTimestamp() <= 0) {
+                throw new CanalParseException("use gtid and TableMeta TSDB should be config timestamp > 0");
+            }
+
             return tableMetaTSDB.rollback(position);
         }
 

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

@@ -4,11 +4,12 @@ import java.io.File;
 import java.io.IOException;
 import java.util.List;
 
+import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet;
+import org.apache.commons.lang.NotImplementedException;
 import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import com.alibaba.otter.canal.parse.exception.CanalParseException;
 import com.alibaba.otter.canal.parse.inbound.ErosaConnection;
 import com.alibaba.otter.canal.parse.inbound.SinkFunction;
 import com.alibaba.otter.canal.parse.inbound.mysql.local.BinLogFileQueue;
@@ -91,7 +92,7 @@ public class LocalBinLogConnection implements ErosaConnection {
                 while (fetcher.fetch()) {
                     event = decoder.decode(fetcher, context);
                     if (event == null) {
-                        throw new CanalParseException("parse failed");
+                        continue;
                     }
 
                     if (!func.sink(event)) {
@@ -201,6 +202,11 @@ public class LocalBinLogConnection implements ErosaConnection {
         dump(binlogFilename, binlogFileOffset, func);
     }
 
+    @Override
+    public void dump(GTIDSet gtidSet, SinkFunction func) throws IOException {
+        throw new NotImplementedException();
+    }
+
     public ErosaConnection fork() {
         LocalBinLogConnection connection = new LocalBinLogConnection();
 

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

@@ -8,6 +8,8 @@ import java.nio.charset.Charset;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 
+import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet;
+import com.alibaba.otter.canal.parse.driver.mysql.packets.client.BinlogDumpGTIDCommandPacket;
 import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -159,6 +161,29 @@ public class MysqlConnection implements ErosaConnection {
         }
     }
 
+    @Override
+    public void dump(GTIDSet gtidSet, SinkFunction func) throws IOException {
+        updateSettings();
+        sendBinlogDumpGTID(gtidSet);
+
+        DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize());
+        fetcher.start(connector.getChannel());
+        LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT);
+        LogContext context = new LogContext();
+        while (fetcher.fetch()) {
+            LogEvent event = null;
+            event = decoder.decode(fetcher, context);
+
+            if (event == null) {
+                throw new CanalParseException("parse failed");
+            }
+
+            if (!func.sink(event)) {
+                break;
+            }
+        }
+    }
+
     public void dump(long timestamp, SinkFunction func) throws IOException {
         throw new NullPointerException("Not implement yet");
     }
@@ -222,6 +247,20 @@ public class MysqlConnection implements ErosaConnection {
         PacketManager.writePkg(connector.getChannel(), semiAckHeader.toBytes(), cmdBody);
     }
 
+    private void sendBinlogDumpGTID(GTIDSet gtidSet) throws IOException {
+        BinlogDumpGTIDCommandPacket binlogDumpCmd = new BinlogDumpGTIDCommandPacket();
+        binlogDumpCmd.slaveServerId = this.slaveId;
+        binlogDumpCmd.gtidSet = gtidSet;
+        byte[] cmdBody = binlogDumpCmd.toBytes();
+
+        logger.info("COM_BINLOG_DUMP_GTID:{}", binlogDumpCmd);
+        HeaderPacket binlogDumpHeader = new HeaderPacket();
+        binlogDumpHeader.setPacketBodyLength(cmdBody.length);
+        binlogDumpHeader.setPacketSequenceNumber((byte) 0x00);
+        PacketManager.writePkg(connector.getChannel(), binlogDumpHeader.toBytes(), cmdBody);
+        connector.setDumping(true);
+    }
+
     public MysqlConnection fork() {
         MysqlConnection connection = new MysqlConnection();
         connection.setCharset(getCharset());

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

@@ -344,6 +344,18 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
     }
 
     protected EntryPosition findStartPosition(ErosaConnection connection) throws IOException {
+        if (isGTIDMode()) {
+            // GTID模式下，CanalLogPositionManager里取最后的gtid，没有则取instanc配置中的
+            LogPosition logPosition = getLogPositionManager().getLatestIndexBy(destination);
+            if (logPosition != null) {
+                return logPosition.getPostion();
+            }
+
+            if (StringUtils.isNotEmpty(masterPosition.getGtid())) {
+                return masterPosition;
+            }
+        }
+
         EntryPosition startPosition = findStartPositionInternal(connection);
         if (needTransactionPosition.get()) {
             logger.warn("prepare to find last position : {}", startPosition.toString());

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

@@ -17,6 +17,7 @@ import org.slf4j.LoggerFactory;
 
 import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
 import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter;
+import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet;
 import com.alibaba.otter.canal.parse.exception.CanalParseException;
 import com.alibaba.otter.canal.parse.exception.TableIdNotFoundException;
 import com.alibaba.otter.canal.parse.inbound.BinlogParser;
@@ -40,6 +41,7 @@ import com.alibaba.otter.canal.protocol.position.EntryPosition;
 import com.google.protobuf.ByteString;
 import com.taobao.tddl.dbsync.binlog.LogEvent;
 import com.taobao.tddl.dbsync.binlog.event.DeleteRowsLogEvent;
+import com.taobao.tddl.dbsync.binlog.event.GtidLogEvent;
 import com.taobao.tddl.dbsync.binlog.event.IntvarLogEvent;
 import com.taobao.tddl.dbsync.binlog.event.LogHeader;
 import com.taobao.tddl.dbsync.binlog.event.QueryLogEvent;
@@ -65,6 +67,12 @@ import com.taobao.tddl.dbsync.binlog.event.mariadb.AnnotateRowsEvent;
  */
 public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogParser<LogEvent> {
 
+    public static final String          XA_XID              = "XA_XID";
+    public static final String          XA_TYPE             = "XA_TYPE";
+    public static final String          XA_START            = "XA START";
+    public static final String          XA_END              = "XA END";
+    public static final String          XA_COMMIT           = "XA COMMIT";
+    public static final String          XA_ROLLBACK         = "XA ROLLBACK";
     public static final String          ISO_8859_1          = "ISO-8859-1";
     public static final String          UTF_8               = "UTF-8";
     public static final int             TINYINT_MAX_VALUE   = 256;
@@ -92,6 +100,17 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
     private boolean                     filterRows          = false;
     private boolean                     useDruidDdlFilter   = true;
 
+    // latest gtid
+    private GTIDSet                     gtidSet;
+
+    public LogEventConvert(GTIDSet gtidSet){
+        this.gtidSet = gtidSet;
+    }
+
+    public LogEventConvert(){
+
+    }
+
     @Override
     public Entry parse(LogEvent logEvent, boolean isSeek) throws CanalParseException {
         if (logEvent == null || logEvent instanceof UnknownLogEvent) {
@@ -128,6 +147,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
                 return parseIntrvarLogEvent((IntvarLogEvent) logEvent);
             case LogEvent.RAND_EVENT:
                 return parseRandLogEvent((RandLogEvent) logEvent);
+            case LogEvent.GTID_LOG_EVENT:
+                return parseGTIDLogEvent((GtidLogEvent) logEvent);
             default:
                 break;
         }
@@ -143,9 +164,59 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
         }
     }
 
+    private Entry parseGTIDLogEvent(GtidLogEvent logEvent) {
+        LogHeader logHeader = logEvent.getHeader();
+        String value = logEvent.getSid().toString() + ":" + logEvent.getGno();
+        Pair.Builder builder = Pair.newBuilder();
+        builder.setKey("gtid");
+        builder.setValue(value);
+        if (gtidSet != null) {
+            gtidSet.update(value);
+        }
+
+        Header header = createHeader("", logHeader, "", "", EventType.GTID);
+        return createEntry(header, EntryType.GTIDLOG, builder.build().toByteString());
+    }
+
     private Entry parseQueryEvent(QueryLogEvent event, boolean isSeek) {
         String queryString = event.getQuery();
-        if (StringUtils.endsWithIgnoreCase(queryString, BEGIN)) {
+        if (StringUtils.startsWithIgnoreCase(queryString, XA_START)) {
+            // xa start use TransactionBegin
+            TransactionBegin.Builder beginBuilder = TransactionBegin.newBuilder();
+            beginBuilder.setThreadId(event.getSessionId());
+            beginBuilder.addProps(createSpecialPair(XA_TYPE, XA_START));
+            beginBuilder.addProps(createSpecialPair(XA_XID, getXaXid(queryString, XA_START)));
+            TransactionBegin transactionBegin = beginBuilder.build();
+            Header header = createHeader(binlogFileName, event.getHeader(), "", "", null);
+            return createEntry(header, EntryType.TRANSACTIONBEGIN, transactionBegin.toByteString());
+        } else if (StringUtils.startsWithIgnoreCase(queryString, XA_END)) {
+            // xa start use TransactionEnd
+            TransactionEnd.Builder endBuilder = TransactionEnd.newBuilder();
+            endBuilder.setTransactionId(String.valueOf(0L));
+            endBuilder.addProps(createSpecialPair(XA_TYPE, XA_END));
+            endBuilder.addProps(createSpecialPair(XA_XID, getXaXid(queryString, XA_END)));
+            TransactionEnd transactionEnd = endBuilder.build();
+            Header header = createHeader(binlogFileName, event.getHeader(), "", "", null);
+            return createEntry(header, EntryType.TRANSACTIONEND, transactionEnd.toByteString());
+        } else if (StringUtils.startsWithIgnoreCase(queryString, XA_COMMIT)) {
+            // xa commit
+            Header header = createHeader(binlogFileName, event.getHeader(), "", "", EventType.XACOMMIT);
+            RowChange.Builder rowChangeBuider = RowChange.newBuilder();
+            rowChangeBuider.setSql(queryString);
+            rowChangeBuider.addProps(createSpecialPair(XA_TYPE, XA_COMMIT));
+            rowChangeBuider.addProps(createSpecialPair(XA_XID, getXaXid(queryString, XA_COMMIT)));
+            rowChangeBuider.setEventType(EventType.XACOMMIT);
+            return createEntry(header, EntryType.ROWDATA, rowChangeBuider.build().toByteString());
+        } else if (StringUtils.startsWithIgnoreCase(queryString, XA_ROLLBACK)) {
+            // xa rollback
+            Header header = createHeader(binlogFileName, event.getHeader(), "", "", EventType.XAROLLBACK);
+            RowChange.Builder rowChangeBuider = RowChange.newBuilder();
+            rowChangeBuider.setSql(queryString);
+            rowChangeBuider.addProps(createSpecialPair(XA_TYPE, XA_ROLLBACK));
+            rowChangeBuider.addProps(createSpecialPair(XA_XID, getXaXid(queryString, XA_ROLLBACK)));
+            rowChangeBuider.setEventType(EventType.XAROLLBACK);
+            return createEntry(header, EntryType.ROWDATA, rowChangeBuider.build().toByteString());
+        } else if (StringUtils.endsWithIgnoreCase(queryString, BEGIN)) {
             TransactionBegin transactionBegin = createTransactionBegin(event.getSessionId());
             Header header = createHeader(binlogFileName, event.getHeader(), "", "", null);
             return createEntry(header, EntryType.TRANSACTIONBEGIN, transactionBegin.toByteString());
@@ -208,6 +279,10 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
         }
     }
 
+    private String getXaXid(String queryString, String type) {
+        return StringUtils.substringAfter(queryString, type);
+    }
+
     private boolean processFilter(String queryString, DdlResult result) {
         String schemaName = result.getSchemaName();
         String tableName = result.getTableName();
@@ -725,6 +800,11 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
             headerBuilder.setTableName(tableName);
         }
         headerBuilder.setEventLength(logHeader.getEventLen());
+        // enable gtid position
+        if (gtidSet != null) {
+            String gtid = gtidSet.toString();
+            headerBuilder.setGtid(gtid);
+        }
         return headerBuilder.build();
     }
 
@@ -773,7 +853,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
         return "LONGTEXT".equalsIgnoreCase(columnType) || "MEDIUMTEXT".equalsIgnoreCase(columnType)
                || "TEXT".equalsIgnoreCase(columnType) || "TINYTEXT".equalsIgnoreCase(columnType);
     }
-    
+
     private boolean isAliSQLHeartBeat(String schema, String table) {
         return "test".equalsIgnoreCase(schema) && "heartbeat".equalsIgnoreCase(table);
     }
@@ -845,4 +925,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
         this.filterRows = filterRows;
     }
 
+    public void setGtidSet(GTIDSet gtidSet) {
+        this.gtidSet = gtidSet;
+    }
+
 }

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

@@ -16,6 +16,7 @@ import com.alibaba.fastsql.sql.ast.SQLExpr;
 import com.alibaba.fastsql.sql.ast.SQLStatement;
 import com.alibaba.fastsql.sql.ast.expr.SQLCharExpr;
 import com.alibaba.fastsql.sql.ast.expr.SQLIdentifierExpr;
+import com.alibaba.fastsql.sql.ast.expr.SQLMethodInvokeExpr;
 import com.alibaba.fastsql.sql.ast.expr.SQLNullExpr;
 import com.alibaba.fastsql.sql.ast.expr.SQLPropertyExpr;
 import com.alibaba.fastsql.sql.ast.statement.SQLColumnConstraint;
@@ -243,6 +244,8 @@ public class MemoryTableMeta implements TableMetaTSDB {
             return DruidDdlParser.unescapeName(((SQLIdentifierExpr) sqlName).getName());
         } else if (sqlName instanceof SQLCharExpr) {
             return ((SQLCharExpr) sqlName).getText();
+        } else if (sqlName instanceof SQLMethodInvokeExpr) {
+            return DruidDdlParser.unescapeName(((SQLMethodInvokeExpr) sqlName).getMethodName());
         } else {
             return sqlName.toString();
         }

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

@@ -4,6 +4,7 @@ import java.net.InetSocketAddress;
 import java.nio.charset.Charset;
 import java.util.List;
 
+import org.apache.commons.lang.StringUtils;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -16,6 +17,7 @@ import com.alibaba.otter.canal.protocol.CanalEntry.Column;
 import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
 import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
 import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
+import com.alibaba.otter.canal.protocol.CanalEntry.Pair;
 import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
 import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
 import com.alibaba.otter.canal.protocol.position.EntryPosition;
@@ -27,8 +29,8 @@ public class MysqlDumpTest {
     @Test
     public void testSimple() {
         final MysqlEventParser controller = new MysqlEventParser();
-        final EntryPosition startPosition = new EntryPosition("mysql-bin.000001", 104606L);
-
+        final EntryPosition startPosition = new EntryPosition("mysql-bin.000012", 34051L, 100L);
+        // startPosition.setGtid("f1ceb61a-a5d5-11e7-bdee-107c3dbcf8a7:1-17");
         controller.setConnectionCharset(Charset.forName("UTF-8"));
         controller.setSlaveId(3344L);
         controller.setDetectingEnable(false);
@@ -39,6 +41,7 @@ public class MysqlDumpTest {
         controller.setTsdbSpringXml("classpath:tsdb/h2-tsdb.xml");
         controller.setEventFilter(new AviaterRegexFilter("test\\..*"));
         controller.setEventBlackFilter(new AviaterRegexFilter("canal_tsdb\\..*"));
+        controller.setIsGTIDMode(false);
         controller.setEventSink(new AbstractCanalEventSinkTest<List<Entry>>() {
 
             public boolean sink(List<Entry> entrys, InetSocketAddress remoteAddress, String destination)
@@ -72,6 +75,7 @@ public class MysqlDumpTest {
                         System.out.println(" sql ----> " + rowChage.getSql());
                     }
 
+                    printXAInfo(rowChage.getPropsList());
                     for (RowData rowData : rowChage.getRowDatasList()) {
                         if (eventType == EventType.DELETE) {
                             print(rowData.getBeforeColumnsList());
@@ -118,4 +122,25 @@ public class MysqlDumpTest {
             System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
         }
     }
+
+    private void printXAInfo(List<Pair> pairs) {
+        if (pairs == null) {
+            return;
+        }
+
+        String xaType = null;
+        String xaXid = null;
+        for (Pair pair : pairs) {
+            String key = pair.getKey();
+            if (StringUtils.endsWithIgnoreCase(key, "XA_TYPE")) {
+                xaType = pair.getValue();
+            } else if (StringUtils.endsWithIgnoreCase(key, "XA_XID")) {
+                xaXid = pair.getValue();
+            }
+        }
+
+        if (xaType != null && xaXid != null) {
+            System.out.println(" ------> " + xaType + " " + xaXid);
+        }
+    }
 }

pom.xml

@@ -254,7 +254,7 @@
             <dependency>
                 <groupId>com.alibaba.fastsql</groupId>
                 <artifactId>fastsql</artifactId>
-                <version>2.0.0_preview_159</version>
+                <version>2.0.0_preview_366</version>
             </dependency>
             <dependency>
                 <groupId>com.alibaba</groupId>

protocol/src/main/java/com/alibaba/otter/canal/protocol/EntryProtocol.proto

@@ -55,7 +55,10 @@ message Header {
 	optional EventType 				eventType			= 11 [default = UPDATE];
 	
 	/**预留扩展**/
-	repeated Pair					props				= 12;		
+	repeated Pair					props				= 12;
+
+    /**当前事务的gitd**/
+	optional string                 gtid                = 13;
 }
 
 /**每个字段的数据结构**/
@@ -169,7 +172,8 @@ enum EntryType{
 	ROWDATA					=		2;
 	TRANSACTIONEND			=		3;
 	/** 心跳类型，内部使用，外部暂不可见，可忽略 **/
-	HEARTBEAT				=		4; 
+	HEARTBEAT				=		4;
+	GTIDLOG                 =       5;
 }
 
 /** 事件类型 **/
@@ -186,6 +190,10 @@ enum EventType {
     /**CREATE INDEX**/
     CINDEX		= 		10;
     DINDEX 		= 		11;
+    GTID        =       12;
+    /** XA **/
+    XACOMMIT    =       13;
+    XAROLLBACK  =		14;
 }
 
 /**数据库类型**/

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

@@ -18,6 +18,7 @@ public class EntryPosition extends TimePosition {
     private Long              position;
     // add by agapple at 2016-06-28
     private Long              serverId              = null;              // 记录一下位点对应的serverId
+    private String            gtid                  = null;
 
     public EntryPosition(){
         super(null);
@@ -74,6 +75,14 @@ public class EntryPosition extends TimePosition {
         this.serverId = serverId;
     }
 
+    public String getGtid() {
+        return gtid;
+    }
+
+    public void setGtid(String gtid) {
+        this.gtid = gtid;
+    }
+
     @Override
     public int hashCode() {
         final int prime = 31;

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

@@ -54,6 +54,9 @@ public class CanalEventUtils {
         // add serverId at 2016-06-28
         position.setServerId(event.getEntry().getHeader().getServerId());
 
+        // add gtid
+        position.setGtid(event.getEntry().getHeader().getGtid());
+
         LogPosition logPosition = new LogPosition();
         logPosition.setPostion(position);
         logPosition.setIdentity(event.getLogIdentity());