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

fix bug for #202（canal.instance.filter.regex 修改后未生效）

README.md

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

client/pom.xml

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

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

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

common/pom.xml

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

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

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

dbsync/pom.xml

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

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

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

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

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

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

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

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

@@ -330,17 +330,14 @@ public class QueryLogEvent extends LogEvent {
                                                           * MariaDb type,
                                                           * sec_part of NOW()
                                                           */
-                                                         + 1 + (MAX_DBS_IN_EVENT_MTS * (1 + NAME_LEN)) + 3 + 1 + 16 + 1 + 60/*
-                                                                                                                             * type
-                                                                                                                             * ,
-                                                                                                                             * user_len
-                                                                                                                             * ,
-                                                                                                                             * user
-                                                                                                                             * ,
-                                                                                                                             * host_len
-                                                                                                                             * ,
-                                                                                                                             * host
-                                                                                                                             */);
+                                                         + 1 + (MAX_DBS_IN_EVENT_MTS * (1 + NAME_LEN)) + 3 + 1 + 32 * 3
+                                                         + 1 + 60/*
+                                                                  * type ,
+                                                                  * user_len ,
+                                                                  * user ,
+                                                                  * host_len ,
+                                                                  * host
+                                                                  */);
     /**
      * Fixed data part:
      * <ul>
@@ -397,7 +394,7 @@ public class QueryLogEvent extends LogEvent {
     // inspection by the DBA
     private final long      execTime;
     private final int       errorCode;
-    private final long      sessionId;                                                                                           /* thread_id */
+    private final long      sessionId;                                                            /* thread_id */
 
     /**
      * 'flags2' is a second set of flags (on top of those in Log_event), for
@@ -499,26 +496,26 @@ public class QueryLogEvent extends LogEvent {
     }
 
     /* query event post-header */
-    public static final int Q_THREAD_ID_OFFSET          = 0;
-    public static final int Q_EXEC_TIME_OFFSET          = 4;
-    public static final int Q_DB_LEN_OFFSET             = 8;
-    public static final int Q_ERR_CODE_OFFSET           = 9;
-    public static final int Q_STATUS_VARS_LEN_OFFSET    = 11;
-    public static final int Q_DATA_OFFSET               = QUERY_HEADER_LEN;
+    public static final int Q_THREAD_ID_OFFSET                = 0;
+    public static final int Q_EXEC_TIME_OFFSET                = 4;
+    public static final int Q_DB_LEN_OFFSET                   = 8;
+    public static final int Q_ERR_CODE_OFFSET                 = 9;
+    public static final int Q_STATUS_VARS_LEN_OFFSET          = 11;
+    public static final int Q_DATA_OFFSET                     = QUERY_HEADER_LEN;
 
     /* these are codes, not offsets; not more than 256 values (1 byte). */
-    public static final int Q_FLAGS2_CODE               = 0;
-    public static final int Q_SQL_MODE_CODE             = 1;
+    public static final int Q_FLAGS2_CODE                     = 0;
+    public static final int Q_SQL_MODE_CODE                   = 1;
 
     /**
      * Q_CATALOG_CODE is catalog with end zero stored; it is used only by MySQL
      * 5.0.x where 0<=x<=3. We have to keep it to be able to replicate these old
      * masters.
      */
-    public static final int Q_CATALOG_CODE              = 2;
-    public static final int Q_AUTO_INCREMENT            = 3;
-    public static final int Q_CHARSET_CODE              = 4;
-    public static final int Q_TIME_ZONE_CODE            = 5;
+    public static final int Q_CATALOG_CODE                    = 2;
+    public static final int Q_AUTO_INCREMENT                  = 3;
+    public static final int Q_CHARSET_CODE                    = 4;
+    public static final int Q_TIME_ZONE_CODE                  = 5;
 
     /**
      * Q_CATALOG_NZ_CODE is catalog withOUT end zero stored; it is used by MySQL
@@ -528,31 +525,47 @@ public class QueryLogEvent extends LogEvent {
      * would crash (segfault etc) because it would expect a 0 when there is
      * none.
      */
-    public static final int Q_CATALOG_NZ_CODE           = 6;
+    public static final int Q_CATALOG_NZ_CODE                 = 6;
 
-    public static final int Q_LC_TIME_NAMES_CODE        = 7;
+    public static final int Q_LC_TIME_NAMES_CODE              = 7;
 
-    public static final int Q_CHARSET_DATABASE_CODE     = 8;
+    public static final int Q_CHARSET_DATABASE_CODE           = 8;
 
-    public static final int Q_TABLE_MAP_FOR_UPDATE_CODE = 9;
+    public static final int Q_TABLE_MAP_FOR_UPDATE_CODE       = 9;
 
-    public static final int Q_MASTER_DATA_WRITTEN_CODE  = 10;
+    public static final int Q_MASTER_DATA_WRITTEN_CODE        = 10;
 
-    public static final int Q_INVOKER                   = 11;
+    public static final int Q_INVOKER                         = 11;
 
     /**
      * Q_UPDATED_DB_NAMES status variable collects of the updated databases
      * total number and their names to be propagated to the slave in order to
      * facilitate the parallel applying of the Query events.
      */
-    public static final int Q_UPDATED_DB_NAMES          = 12;
+    public static final int Q_UPDATED_DB_NAMES                = 12;
 
-    public static final int Q_MICROSECONDS              = 13;
+    public static final int Q_MICROSECONDS                    = 13;
+    /**
+     * A old (unused now) code for Query_log_event status similar to
+     * G_COMMIT_TS.
+     */
+    public static final int Q_COMMIT_TS                       = 14;
+    /**
+     * A code for Query_log_event status, similar to G_COMMIT_TS2.
+     */
+    public static final int Q_COMMIT_TS2                      = 15;
+    /**
+     * The master connection @@session.explicit_defaults_for_timestamp which is
+     * recorded for queries, CREATE and ALTER table that is defined with a
+     * TIMESTAMP column, that are dependent on that feature. For pre-WL6292
+     * master's the associated with this code value is zero.
+     */
+    public static final int Q_EXPLICIT_DEFAULTS_FOR_TIMESTAMP = 16;
 
     /**
      * FROM MariaDB 5.5.34
      */
-    public static final int Q_HRNOW                     = 128;
+    public static final int Q_HRNOW                           = 128;
 
     private final void unpackVariables(LogBuffer buffer, final int end) throws IOException {
         int code = -1;
@@ -632,6 +645,9 @@ public class QueryLogEvent extends LogEvent {
                             mtsAccessedDbNames[i] = buffer.getFixString(length < NAME_LEN ? length : NAME_LEN);
                         }
                         break;
+                    case Q_EXPLICIT_DEFAULTS_FOR_TIMESTAMP:
+                        buffer.forward(1);
+                        break;
                     case Q_HRNOW:
                         // int when_sec_part = buffer.getUint24();
                         buffer.forward(3);

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

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

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

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

deployer/pom.xml

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

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

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

driver/pom.xml

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

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

@@ -27,22 +27,22 @@ import com.alibaba.otter.canal.parse.driver.mysql.utils.PacketManager;
  */
 public class MysqlConnector {
 
-    private static final Logger logger = LoggerFactory.getLogger(MysqlConnector.class);
-    private InetSocketAddress address;
-    private String username;
-    private String password;
+    private static final Logger logger            = LoggerFactory.getLogger(MysqlConnector.class);
+    private InetSocketAddress   address;
+    private String              username;
+    private String              password;
 
-    private byte charsetNumber = 33;
-    private String defaultSchema = "retl";
-    private int soTimeout = 30 * 1000;
-    private int receiveBufferSize = 16 * 1024;
-    private int sendBufferSize = 16 * 1024;
+    private byte                charsetNumber     = 33;
+    private String              defaultSchema     = "retl";
+    private int                 soTimeout         = 30 * 1000;
+    private int                 receiveBufferSize = 16 * 1024;
+    private int                 sendBufferSize    = 16 * 1024;
 
-    private SocketChannel channel;
-    private volatile boolean dumping = false;
+    private SocketChannel       channel;
+    private volatile boolean    dumping           = false;
     // mysql connectinnId
-    private long connectionId = -1;
-    private AtomicBoolean connected = new AtomicBoolean(false);
+    private long                connectionId      = -1;
+    private AtomicBoolean       connected         = new AtomicBoolean(false);
 
     public MysqlConnector(){
     }

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

@@ -24,9 +24,9 @@ public class MysqlQueryExecutor {
 
     private SocketChannel channel;
 
-    public MysqlQueryExecutor(MysqlConnector connector){
+    public MysqlQueryExecutor(MysqlConnector connector) throws IOException{
         if (!connector.isConnected()) {
-            throw new RuntimeException("should execute connector.connect() first");
+            throw new IOException("should execute connector.connect() first");
         }
 
         this.channel = connector.getChannel();

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

@@ -23,9 +23,9 @@ public class MysqlUpdateExecutor {
 
     private SocketChannel       channel;
 
-    public MysqlUpdateExecutor(MysqlConnector connector){
+    public MysqlUpdateExecutor(MysqlConnector connector) throws IOException{
         if (!connector.isConnected()) {
-            throw new RuntimeException("should execute connector.connect() first");
+            throw new IOException("should execute connector.connect() first");
         }
 
         this.channel = connector.getChannel();

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

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

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

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

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

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

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

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

example/pom.xml

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

filter/pom.xml

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

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

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

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

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

instance/core/pom.xml

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

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

@@ -1,5 +1,11 @@
 package com.alibaba.otter.canal.instance.core;
 
+import java.util.List;
+
+import org.apache.commons.lang.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
 import com.alibaba.otter.canal.common.alarm.CanalAlarmHandler;
 import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter;
@@ -16,30 +22,22 @@ import com.alibaba.otter.canal.protocol.ClientIdentity;
 import com.alibaba.otter.canal.sink.CanalEventSink;
 import com.alibaba.otter.canal.store.CanalEventStore;
 import com.alibaba.otter.canal.store.model.Event;
-import org.apache.commons.lang.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.List;
 
 /**
- * Created with Intellig IDEA.
- * Author: yinxiu
- * Date: 2016-01-07
- * Time: 22:26
+ * Created with Intellig IDEA. Author: yinxiu Date: 2016-01-07 Time: 22:26
  */
 public class AbstractCanalInstance extends AbstractCanalLifeCycle implements CanalInstance {
 
-    private static final Logger logger = LoggerFactory.getLogger(AbstractCanalInstance.class);
+    private static final Logger                      logger = LoggerFactory.getLogger(AbstractCanalInstance.class);
 
-    protected Long                                   canalId;                               // 和manager交互唯一标示
-    protected String                                 destination;                           // 队列名字
-    protected CanalEventStore<Event>                 eventStore;                            // 有序队列
+    protected Long                                   canalId;                                                      // 和manager交互唯一标示
+    protected String                                 destination;                                                  // 队列名字
+    protected CanalEventStore<Event>                 eventStore;                                                   // 有序队列
 
-    protected CanalEventParser                       eventParser;                           // 解析对应的数据信息
-    protected CanalEventSink<List<CanalEntry.Entry>> eventSink;                             // 链接parse和store的桥接器
-    protected CanalMetaManager                       metaManager;                           // 消费信息管理器
-    protected CanalAlarmHandler                      alarmHandler;                          // alarm报警机制
+    protected CanalEventParser                       eventParser;                                                  // 解析对应的数据信息
+    protected CanalEventSink<List<CanalEntry.Entry>> eventSink;                                                    // 链接parse和store的桥接器
+    protected CanalMetaManager                       metaManager;                                                  // 消费信息管理器
+    protected CanalAlarmHandler                      alarmHandler;                                                 // alarm报警机制
 
     @Override
     public boolean subscribeChange(ClientIdentity identity) {
@@ -121,7 +119,6 @@ public class AbstractCanalInstance extends AbstractCanalLifeCycle implements Can
             alarmHandler.stop();
         }
 
-
         logger.info("stop successful....");
     }
 

instance/manager/pom.xml

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

instance/pom.xml

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

instance/spring/pom.xml

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

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

@@ -2,11 +2,11 @@ package com.alibaba.otter.canal.instance.spring;
 
 import java.util.List;
 
-import com.alibaba.otter.canal.instance.core.AbstractCanalInstance;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import com.alibaba.otter.canal.common.alarm.CanalAlarmHandler;
+import com.alibaba.otter.canal.instance.core.AbstractCanalInstance;
 import com.alibaba.otter.canal.meta.CanalMetaManager;
 import com.alibaba.otter.canal.parse.CanalEventParser;
 import com.alibaba.otter.canal.protocol.CanalEntry;
@@ -23,7 +23,7 @@ import com.alibaba.otter.canal.store.model.Event;
  */
 public class CanalInstanceWithSpring extends AbstractCanalInstance {
 
-    private static final Logger                    logger = LoggerFactory.getLogger(CanalInstanceWithSpring.class);
+    private static final Logger logger = LoggerFactory.getLogger(CanalInstanceWithSpring.class);
 
     public void start() {
         logger.info("start CannalInstance for {}-{} ", new Object[] { 1, destination });

meta/pom.xml

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

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

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

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

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

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

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

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

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

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

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

parse/pom.xml

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

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

@@ -83,30 +83,27 @@ public class LocalBinLogConnection implements ErosaConnection {
             context.setLogPosition(new LogPosition(binlogfilename, binlogPosition));
             while (running) {
                 boolean needContinue = true;
-				LogEvent event = null;
-                L:while (fetcher.fetch()) {
-                    /*event = decoder.decode(fetcher, context);
-                    if (event == null) {
-                        throw new CanalParseException("parse failed");
-                    }
+                LogEvent event = null;
+                L: while (fetcher.fetch()) {
+                    /*
+                     * event = decoder.decode(fetcher, context); if (event ==
+                     * null) { throw new CanalParseException("parse failed"); }
+                     * if (!func.sink(event)) { needContinue = false; break; }
+                     */
 
-                    if (!func.sink(event)) {
-                        needContinue = false;
-                        break;
-                    }*/
+                    do {
+                        if (event == null) {
+                            event = new RotateLogEvent(context.getLogPosition().getFileName(), context.getLogPosition()
+                                .getPosition());
+                        } else {
+                            event = decoder.decode(fetcher, context);
+                        }
 
-					do {
-						if (event == null) {
-							event = new RotateLogEvent(context.getLogPosition().getFileName(), context.getLogPosition().getPosition());
-						} else {
-							event = decoder.decode(fetcher, context);
-						}
-												
-						if (event != null && !func.sink(event)) {
-							needContinue = false;
-							break L;
-						}
-					} while (event != null);
+                        if (event != null && !func.sink(event)) {
+                            needContinue = false;
+                            break L;
+                        }
+                    } while (event != null);
                 }
 
                 if (needContinue) {// 读取下一个

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

@@ -27,7 +27,7 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements
     private EntryPosition      masterPosition;        // binlog信息
     private MysqlConnection    metaConnection;        // 查询meta信息的链接
     private TableMetaCache     tableMetaCache;        // 对应meta
-    
+
     private String             directory;
     private boolean            needWait   = false;
     private int                bufferSize = 16 * 1024;
@@ -39,24 +39,24 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements
     @Override
     protected ErosaConnection buildErosaConnection() {
         return buildLocalBinLogConnection();
-    }   
-    
+    }
+
     @Override
-	protected void preDump(ErosaConnection connection) {
-    	metaConnection = buildMysqlConnection();
+    protected void preDump(ErosaConnection connection) {
+        metaConnection = buildMysqlConnection();
         try {
             metaConnection.connect();
         } catch (IOException e) {
             throw new CanalParseException(e);
         }
-        
+
         tableMetaCache = new TableMetaCache(metaConnection);
         ((LogEventConvert) binlogParser).setTableMetaCache(tableMetaCache);
-	}
+    }
 
-	@Override
-	protected void afterDump(ErosaConnection connection) {
-		if (metaConnection != null) {
+    @Override
+    protected void afterDump(ErosaConnection connection) {
+        if (metaConnection != null) {
             try {
                 metaConnection.disconnect();
             } catch (IOException e) {
@@ -64,19 +64,19 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements
                     .getAddress(), e);
             }
         }
-	}
+    }
 
-	public void start() throws CanalParseException {
+    public void start() throws CanalParseException {
         if (runningInfo == null) { // 第一次链接主库
             runningInfo = masterInfo;
         }
 
         super.start();
     }
-	
+
     @Override
-	public void stop() {
-    	if (metaConnection != null) {
+    public void stop() {
+        if (metaConnection != null) {
             try {
                 metaConnection.disconnect();
             } catch (IOException e) {
@@ -90,9 +90,9 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements
         }
 
         super.stop();
-	}
+    }
 
-	private ErosaConnection buildLocalBinLogConnection() {
+    private ErosaConnection buildLocalBinLogConnection() {
         LocalBinLogConnection connection = new LocalBinLogConnection();
 
         connection.setBufferSize(this.bufferSize);
@@ -114,7 +114,7 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements
         connection.setCharset(connectionCharset);
         return connection;
     }
-    
+
     @Override
     protected EntryPosition findStartPosition(ErosaConnection connection) {
         // 处理逻辑

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

@@ -83,7 +83,7 @@ public class MysqlConnection implements ErosaConnection {
      */
     public void seek(String binlogfilename, Long binlogPosition, SinkFunction func) throws IOException {
         updateSettings();
-
+        loadBinlogChecksum();
         sendBinlogDump(binlogfilename, binlogPosition);
         DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize());
         fetcher.start(connector.getChannel());
@@ -94,6 +94,7 @@ public class MysqlConnection implements ErosaConnection {
         decoder.handle(LogEvent.XID_EVENT);
         LogContext context = new LogContext();
         context.setLogPosition(new LogPosition(binlogfilename));
+        context.setFormatDescription(new FormatDescriptionLogEvent(4, binlogChecksum));
         while (fetcher.fetch()) {
             LogEvent event = null;
             event = decoder.decode(fetcher, context);
@@ -212,6 +213,16 @@ public class MysqlConnection implements ErosaConnection {
             }
         }
 
+        try {
+            // 参考:https://github.com/alibaba/canal/issues/284
+            // mysql5.6需要设置slave_uuid避免被server kill链接
+            update("set @slave_uuid=uuid()");
+        } catch (Exception e) {
+            if (!StringUtils.contains(e.getMessage(), "Unknown system variable")) {
+                logger.warn(ExceptionUtils.getFullStackTrace(e));
+            }
+        }
+
         try {
             // mariadb针对特殊的类型，需要设置session变量
             update("SET @mariadb_slave_capability='" + LogEvent.MARIA_SLAVE_CAPABILITY_MINE + "'");
@@ -281,7 +292,8 @@ public class MysqlConnection implements ErosaConnection {
         }
 
         List<String> columnValues = rs.getFieldValues();
-        if (columnValues != null && columnValues.size() >= 1 && columnValues.get(0).toUpperCase().equals("CRC32")) {
+        if (columnValues != null && columnValues.size() >= 1 && columnValues.get(0) != null
+            && columnValues.get(0).toUpperCase().equals("CRC32")) {
             binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_CRC32;
         } else {
             binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_OFF;

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

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

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

@@ -190,9 +190,9 @@ public class BinLogFileQueue {
         files.addAll(FileUtils.listFiles(directory, new IOFileFilter() {
 
             public boolean accept(File file) {
-				Pattern pattern = Pattern.compile("\\d+$");
-				Matcher matcher = pattern.matcher(file.getName());
-				return file.getName().startsWith(baseName) && matcher.find();
+                Pattern pattern = Pattern.compile("\\d+$");
+                Matcher matcher = pattern.matcher(file.getName());
+                return file.getName().startsWith(baseName) && matcher.find();
             }
 
             public boolean accept(File dir, String name) {

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

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

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

@@ -5,7 +5,6 @@ import java.nio.charset.Charset;
 import java.util.List;
 
 import org.junit.Assert;
-
 import org.junit.Test;
 
 import com.alibaba.otter.canal.parse.stub.AbstractCanalEventSinkTest;

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

@@ -5,7 +5,6 @@ import java.nio.charset.Charset;
 import java.util.List;
 
 import org.junit.Assert;
-
 import org.junit.Test;
 
 import com.alibaba.otter.canal.parse.stub.AbstractCanalEventSinkTest;

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

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

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

@@ -3,9 +3,8 @@ package com.alibaba.otter.canal.parse.index;
 import java.io.File;
 import java.io.IOException;
 
-import org.junit.Assert;
-
 import org.apache.commons.io.FileUtils;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 

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

@@ -3,9 +3,8 @@ package com.alibaba.otter.canal.parse.index;
 import java.net.InetSocketAddress;
 import java.util.Date;
 
-import org.junit.Assert;
-
 import org.junit.After;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 

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

@@ -1,8 +1,7 @@
 package com.alibaba.otter.canal.parse.index;
 
-import org.junit.Assert;
-
 import org.junit.After;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 

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

@@ -1,8 +1,7 @@
 package com.alibaba.otter.canal.parse.index;
 
-import org.junit.Assert;
-
 import org.junit.After;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 

pom.xml

@@ -4,7 +4,7 @@
     <artifactId>canal</artifactId>
     <packaging>pom</packaging>
     <name>canal module for otter ${project.version}</name>
-    <version>1.0.24-SNAPSHOT</version>
+    <version>1.0.24</version>
     <url>https://github.com/alibaba/canal</url>
     <parent>
         <groupId>org.sonatype.oss</groupId>
@@ -177,7 +177,7 @@
             <dependency>
                 <groupId>com.alibaba</groupId>
                 <artifactId>fastjson</artifactId>
-                <version>1.1.35</version>
+                <version>1.2.28</version>
             </dependency>
             <dependency>
                 <groupId>com.google.guava</groupId>
@@ -199,6 +199,11 @@
                 <artifactId>netty</artifactId>
                 <version>3.2.5.Final</version>
             </dependency>
+            <dependency>
+                <groupId>io.netty</groupId>
+                <artifactId>netty-all</artifactId>
+                <version>4.1.6.Final</version>
+            </dependency>
             <dependency>
                 <groupId>com.google.protobuf</groupId>
                 <artifactId>protobuf-java</artifactId>
@@ -336,7 +341,7 @@
 				  <keywords>true</keywords>
 				</configuration>
 			</plugin>
-			
+
 			<plugin>
 				<groupId>org.apache.maven.plugins</groupId>
 				<artifactId>maven-jxr-plugin</artifactId>

protocol/pom.xml

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

server/pom.xml

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

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

@@ -131,8 +131,8 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
      */
     @Override
     public void subscribe(ClientIdentity clientIdentity) throws CanalServerException {
-    	checkStart(clientIdentity.getDestination());
-    	
+        checkStart(clientIdentity.getDestination());
+
         CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
         if (!canalInstance.getMetaManager().isStart()) {
             canalInstance.getMetaManager().start();

sink/pom.xml

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

store/pom.xml

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

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

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