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

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

README.md

@@ -9,7 +9,7 @@
 
 <h1>背景</h1>
 <p style="font-size: 14px;">   早期，阿里巴巴B2B公司因为存在杭州和美国双机房部署，存在跨机房同步的业务需求。不过早期的数据库同步业务，主要是基于trigger的方式获取增量变更，不过从2010年开始，阿里系公司开始逐步的尝试基于数据库的日志解析，获取增量变更进行同步，由此衍生出了增量订阅&amp;消费的业务，从此开启了一段新纪元。</p>
-<p style="font-size: 14px;">   ps. 目前内部版本已经支持mysql和oracle部分版本的日志解析，当前的canal开源版本支持5.6及以下的版本(阿里内部mysql 5.6.10, mysql 5.5.18和5.1.40/48)</p>
+<p style="font-size: 14px;">   ps. 目前内部版本已经支持mysql和oracle部分版本的日志解析，当前的canal开源版本支持5.7及以下的版本(阿里内部mysql 5.7.13, 5.6.10, mysql 5.5.18和5.1.40/48)</p>
 <p style="font-size: 14px;"> </p>
 <p style="font-size: 14px;">基于日志增量订阅&amp;消费支持的业务：</p>
 <ol style="font-size: 14px;">
@@ -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.22</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.22</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>
+			<groupId>junit</groupId>
-		  <artifactId>junit</artifactId>
+			<artifactId>junit</artifactId>
 		</dependency>
-  </dependencies>
+	</dependencies>
 </project>

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

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

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

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

dbsync/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.0.22</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(" :");
+                        buf.append(": ");
-                        element(i).toJsonString(buf);
+                        element(i, charsetName).toJsonString(buf, charsetName);
                     }
                     buf.append("}");
                     break;
@@ -292,7 +293,7 @@ public class JsonConversion {
                         if (i > 0) {
                             buf.append(", ");
                         }
-                        element(i).toJsonString(buf);
+                        element(i, charsetName).toJsonString(buf, charsetName);
                     }
                     buf.append("]");
                     break;
@@ -361,7 +362,7 @@ public class JsonConversion {
                         }
                         buf.append('"').append(text).append('"');
                     } else {
-                        text = m_data.getFixString((int) m_length);
+                        text = m_data.getFixString((int) m_length, charsetName);
                         buf.append('"').append(text).append('"');
                     }
 

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

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

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

@@ -141,6 +141,11 @@ public final class FormatDescriptionLogEvent extends StartLogEventV3 {
         }
     }
 
+    public FormatDescriptionLogEvent(final int binlogVersion, int binlogChecksum){
+        this(binlogVersion);
+        this.header.checksumAlg = binlogChecksum;
+    }
+
     public FormatDescriptionLogEvent(final int binlogVersion){
         this.binlogVersion = 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/*
+                                                         + 1 + (MAX_DBS_IN_EVENT_MTS * (1 + NAME_LEN)) + 3 + 1 + 32 * 3
-                                                                                                                             * type
+                                                         + 1 + 60/*
-                                                                                                                             * ,
+                                                                  * type ,
-                                                                                                                             * user_len
+                                                                  * user_len ,
-                                                                                                                             * ,
+                                                                  * user ,
-                                                                                                                             * user
+                                                                  * host_len ,
-                                                                                                                             * ,
+                                                                  * host
-                                                                                                                             * 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_THREAD_ID_OFFSET                = 0;
-    public static final int Q_EXEC_TIME_OFFSET          = 4;
+    public static final int Q_EXEC_TIME_OFFSET                = 4;
-    public static final int Q_DB_LEN_OFFSET             = 8;
+    public static final int Q_DB_LEN_OFFSET                   = 8;
-    public static final int Q_ERR_CODE_OFFSET           = 9;
+    public static final int Q_ERR_CODE_OFFSET                 = 9;
-    public static final int Q_STATUS_VARS_LEN_OFFSET    = 11;
+    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_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_FLAGS2_CODE                     = 0;
-    public static final int Q_SQL_MODE_CODE             = 1;
+    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_CATALOG_CODE                    = 2;
-    public static final int Q_AUTO_INCREMENT            = 3;
+    public static final int Q_AUTO_INCREMENT                  = 3;
-    public static final int Q_CHARSET_CODE              = 4;
+    public static final int Q_CHARSET_CODE                    = 4;
-    public static final int Q_TIME_ZONE_CODE            = 5;
+    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,6 +90,7 @@ public final class RotateLogEvent extends LogEvent {
         int filenameLen = buffer.limit() - filenameOffset;
         if (filenameLen > FN_REFLEN - 1) filenameLen = FN_REFLEN - 1;
         buffer.position(filenameOffset);
+
         filename = buffer.getFixString(filenameLen);
     }
 

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

@@ -952,16 +952,15 @@ 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];
                 // buffer.fillBytes(binary, 0, len);
                 // value = binary;
-                javaType = Types.VARBINARY;
+                javaType = Types.VARCHAR;
                 length = len;
                 break;
             }

deployer/pom.xml

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

@@ -5,6 +5,7 @@ import java.util.Properties;
 
 import org.I0Itec.zkclient.IZkStateListener;
 import org.I0Itec.zkclient.exception.ZkNoNodeException;
+import org.I0Itec.zkclient.exception.ZkNodeExistsException;
 import org.apache.commons.lang.BooleanUtils;
 import org.apache.commons.lang.StringUtils;
 import org.apache.zookeeper.Watcher.Event.KeeperState;
@@ -172,6 +173,8 @@ public class CanalController {
                 if (zkclientx != null) {
                     runningMonitor.setZkClient(zkclientx);
                 }
+                // 触发创建一下cid节点
+                runningMonitor.init();
                 return runningMonitor;
             }
         }));
@@ -189,10 +192,10 @@ public class CanalController {
                         instanceConfigs.put(destination, config);
                     }
 
-                    if (!config.getLazy() && !embededCanalServer.isStart(destination)) {
+                    if (!embededCanalServer.isStart(destination)) {
                         // HA机制启动
                         ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination);
-                        if (!runningMonitor.isStart()) {
+                        if (!config.getLazy() && !runningMonitor.isStart()) {
                             runningMonitor.start();
                         }
                     }
@@ -389,10 +392,10 @@ public class CanalController {
             final String destination = entry.getKey();
             InstanceConfig config = entry.getValue();
             // 创建destination的工作节点
-            if (!config.getLazy() && !embededCanalServer.isStart(destination)) {
+            if (!embededCanalServer.isStart(destination)) {
                 // HA机制启动
                 ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination);
-                if (!runningMonitor.isStart()) {
+                if (!config.getLazy() && !runningMonitor.isStart()) {
                     runningMonitor.start();
                 }
             }
@@ -448,6 +451,9 @@ public class CanalController {
                 String parentDir = path.substring(0, path.lastIndexOf('/'));
                 zkclientx.createPersistent(parentDir, true);
                 zkclientx.createEphemeral(path);
+            } catch (ZkNodeExistsException e) {
+                // ignore
+                // 因为第一次启动时创建了cid,但在stop/start的时可能会关闭和新建,允许出现NodeExists问题s
             }
 
         }

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

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

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

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

driver/pom.xml

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

@@ -103,7 +103,8 @@ public class MysqlConnector {
                     MysqlUpdateExecutor executor = new MysqlUpdateExecutor(connector);
                     executor.update("KILL CONNECTION " + connectionId);
                 } catch (Exception e) {
-                    throw new IOException("KILL DUMP " + connectionId + " failure:" + ExceptionUtils.getStackTrace(e));
+                    // 忽略具体异常
+                    logger.info("KILL DUMP " + connectionId + " failure:" + ExceptionUtils.getStackTrace(e));
                 } finally {
                     if (connector != null) {
                         connector.disconnect();

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.22</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.22</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.22</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.
+ * Created with Intellig IDEA. Author: yinxiu Date: 2016-01-07 Time: 22:26
- * Author: yinxiu
- * Date: 2016-01-07
- * Time: 22:26
  */
 public class AbstractCanalInstance extends AbstractCanalLifeCycle implements CanalInstance {
 
-    private static final Logger logger = LoggerFactory.getLogger(AbstractCanalInstance.class);
+    private static final Logger                      logger = LoggerFactory.getLogger(AbstractCanalInstance.class);
 
-    protected Long                                   canalId;                               // 和manager交互唯一标示
+    protected Long                                   canalId;                                                      // 和manager交互唯一标示
-    protected String                                 destination;                           // 队列名字
+    protected String                                 destination;                                                  // 队列名字
-    protected CanalEventStore<Event>                 eventStore;                            // 有序队列
+    protected CanalEventStore<Event>                 eventStore;                                                   // 有序队列
 
-    protected CanalEventParser                       eventParser;                           // 解析对应的数据信息
+    protected CanalEventParser                       eventParser;                                                  // 解析对应的数据信息
-    protected CanalEventSink<List<CanalEntry.Entry>> eventSink;                             // 链接parse和store的桥接器
+    protected CanalEventSink<List<CanalEntry.Entry>> eventSink;                                                    // 链接parse和store的桥接器
-    protected CanalMetaManager                       metaManager;                           // 消费信息管理器
+    protected CanalMetaManager                       metaManager;                                                  // 消费信息管理器
-    protected CanalAlarmHandler                      alarmHandler;                          // alarm报警机制
+    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.22</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.22</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.22</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.22</version>
+		<version>1.0.24</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>

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

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

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.22</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/AbstractEventParser.java

@@ -22,7 +22,6 @@ import com.alibaba.otter.canal.parse.CanalEventParser;
 import com.alibaba.otter.canal.parse.exception.CanalParseException;
 import com.alibaba.otter.canal.parse.exception.TableIdNotFoundException;
 import com.alibaba.otter.canal.parse.inbound.EventTransactionBuffer.TransactionFlushCallback;
-import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser;
 import com.alibaba.otter.canal.parse.index.CanalLogPositionManager;
 import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
 import com.alibaba.otter.canal.protocol.CanalEntry;
@@ -325,7 +324,6 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
             startTs = System.currentTimeMillis();
         }
         CanalEntry.Entry event = binlogParser.parse(bod);
-
         if (enabled) {
             this.parsingInterval = System.currentTimeMillis() - startTs;
         }
@@ -388,7 +386,11 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
             String name = String.format("destination = %s , address = %s , HeartBeatTimeTask",
                 destination,
                 runningInfo == null ? null : runningInfo.getAddress().toString());
-            synchronized (MysqlEventParser.class) {
+            synchronized (AbstractEventParser.class) {
+                // synchronized (MysqlEventParser.class) {
+                // why use MysqlEventParser.class, u know, MysqlEventParser is
+                // the child class 4 AbstractEventParser,
+                // do this is ...
                 if (timer == null) {
                     timer = new Timer(name, true);
                 }

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

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

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

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

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

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

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

@@ -6,6 +6,7 @@ import java.nio.ByteBuffer;
 import java.nio.charset.Charset;
 import java.util.List;
 
+import org.apache.commons.lang.StringUtils;
 import org.apache.commons.lang.exception.ExceptionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -24,6 +25,8 @@ import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.DirectLogFetcher;
 import com.taobao.tddl.dbsync.binlog.LogContext;
 import com.taobao.tddl.dbsync.binlog.LogDecoder;
 import com.taobao.tddl.dbsync.binlog.LogEvent;
+import com.taobao.tddl.dbsync.binlog.LogPosition;
+import com.taobao.tddl.dbsync.binlog.event.FormatDescriptionLogEvent;
 
 public class MysqlConnection implements ErosaConnection {
 
@@ -34,6 +37,7 @@ public class MysqlConnection implements ErosaConnection {
     private Charset             charset = Charset.forName("UTF-8");
     private BinlogFormat        binlogFormat;
     private BinlogImage         binlogImage;
+    private int                 binlogChecksum;
 
     public MysqlConnection(){
     }
@@ -79,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());
@@ -89,6 +93,8 @@ public class MysqlConnection implements ErosaConnection {
         decoder.handle(LogEvent.QUERY_EVENT);
         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);
@@ -105,11 +111,14 @@ public class MysqlConnection implements ErosaConnection {
 
     public void dump(String binlogfilename, Long binlogPosition, SinkFunction func) throws IOException {
         updateSettings();
+        loadBinlogChecksum();
         sendBinlogDump(binlogfilename, binlogPosition);
         DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize());
         fetcher.start(connector.getChannel());
         LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT);
         LogContext context = new LogContext();
+        context.setLogPosition(new LogPosition(binlogfilename));
+        context.setFormatDescription(new FormatDescriptionLogEvent(4, binlogChecksum));
         while (fetcher.fetch()) {
             LogEvent event = null;
             event = decoder.decode(fetcher, context);
@@ -196,9 +205,22 @@ public class MysqlConnection implements ErosaConnection {
             // 如果不设置会出现错误： Slave can not handle replication events with the
             // checksum that master is configured to log
             // 但也不能乱设置，需要和mysql server的checksum配置一致，不然RotateLogEvent会出现乱码
-            update("set @master_binlog_checksum= '@@global.binlog_checksum'");
+            // '@@global.binlog_checksum'需要去掉单引号,在mysql 5.6.29下导致master退出
+            update("set @master_binlog_checksum= @@global.binlog_checksum");
         } catch (Exception e) {
-            logger.warn(ExceptionUtils.getFullStackTrace(e));
+            if (!StringUtils.contains(e.getMessage(), "Unknown system variable")) {
+                logger.warn(ExceptionUtils.getFullStackTrace(e));
+            }
+        }
+
+        try {
+            // 参考: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 {
@@ -256,6 +278,28 @@ public class MysqlConnection implements ErosaConnection {
         }
     }
 
+    /**
+     * 获取主库checksum信息
+     * https://dev.mysql.com/doc/refman/5.6/en/replication-options
+     * -binary-log.html#option_mysqld_binlog-checksum
+     */
+    private void loadBinlogChecksum() {
+        ResultSetPacket rs = null;
+        try {
+            rs = query("select @master_binlog_checksum");
+        } catch (IOException e) {
+            throw new CanalParseException(e);
+        }
+
+        List<String> columnValues = rs.getFieldValues();
+        if (columnValues != null && columnValues.size() >= 1 && columnValues.get(0) != null
+            && columnValues.get(0).toUpperCase().equals("CRC32")) {
+            binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_CRC32;
+        } else {
+            binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_OFF;
+        }
+    }
+
     public static enum BinlogFormat {
 
         STATEMENT("STATEMENT"), ROW("ROW"), MIXED("MIXED");

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

@@ -13,6 +13,7 @@ import org.apache.commons.lang.StringUtils;
 import org.apache.commons.lang.exception.ExceptionUtils;
 import org.springframework.util.CollectionUtils;
 
+import com.alibaba.otter.canal.common.utils.JsonUtils;
 import com.alibaba.otter.canal.parse.CanalEventParser;
 import com.alibaba.otter.canal.parse.CanalHASwitchable;
 import com.alibaba.otter.canal.parse.driver.mysql.packets.server.FieldPacket;
@@ -93,6 +94,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
                 for (BinlogFormat supportFormat : supportBinlogFormats) {
                     if (supportFormat != null && format == supportFormat) {
                         found = true;
+                        break;
                     }
                 }
 
@@ -107,6 +109,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
                 for (BinlogImage supportImage : supportBinlogImages) {
                     if (supportImage != null && image == supportImage) {
                         found = true;
+                        break;
                     }
                 }
 
@@ -177,10 +180,9 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
     }
 
     protected void stopHeartBeat() {
+        TimerTask heartBeatTimerTask = this.heartBeatTimerTask;
         super.stopHeartBeat();
-
+        if (heartBeatTimerTask != null && heartBeatTimerTask instanceof MysqlDetectingTimeTask) {
-        if (heartBeatTimerTask != null) {
-
             MysqlConnection mysqlConnection = ((MysqlDetectingTimeTask) heartBeatTimerTask).getMysqlConnection();
             try {
                 mysqlConnection.disconnect();
@@ -395,7 +397,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
                     // binlog定位位点失败,可能有两个原因:
                     // 1. binlog位点被删除
                     // 2.vip模式的mysql,发生了主备切换,判断一下serverId是否变化,针对这种模式可以发起一次基于时间戳查找合适的binlog位点
-                    boolean case2 = (standbyInfo.getAddress() == null)
+                    boolean case2 = (standbyInfo == null || standbyInfo.getAddress() == null)
                                     && logPosition.getPostion().getServerId() != null
                                     && !logPosition.getPostion().getServerId().equals(findServerId(mysqlConnection));
                     if (case2) {
@@ -410,7 +412,8 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
                     }
                 }
                 // 其余情况
-                logger.warn("prepare to find start position just last position");
+                logger.warn("prepare to find start position just last position\n {}",
+                    JsonUtils.marshalToString(logPosition));
                 return logPosition.getPostion();
             } else {
                 // 针对切换的情况，考虑回退时间

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

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

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

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

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

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

parse/src/test/java/com/alibaba/otter/canal/parse/inbound/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/MysqlEventParserTest.java

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

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.22</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,10 +199,15 @@
                 <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>
-                <version>2.4.1</version>
+                <version>2.6.1</version>
             </dependency>
             <!-- log -->
             <dependency>
@@ -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.22</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.22</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,6 +131,8 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
      */
     @Override
     public void subscribe(ClientIdentity clientIdentity) throws CanalServerException {
+        checkStart(clientIdentity.getDestination());
+
         CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
         if (!canalInstance.getMetaManager().isStart()) {
             canalInstance.getMetaManager().start();

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

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

sink/pom.xml

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