Merge branch 'master' into table_meta_support

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

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

@@ -3,7 +3,6 @@ package com.alibaba.otter.canal.client.impl;
 import java.net.SocketAddress;
 import java.util.concurrent.TimeUnit;
 
-import org.apache.commons.lang.exception.ExceptionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -107,9 +106,10 @@ public class ClusterCanalConnector implements CanalConnector {
                 this.filter = filter;
                 return;
             } catch (Throwable t) {
-                logger.warn("something goes wrong when subscribing from server:{}\n{}",
+                logger.warn(String.format(
-                    currentConnector != null ? currentConnector.getAddress() : "null",
+                        "something goes wrong when subscribing from server: %s",
-                    ExceptionUtils.getFullStackTrace(t));
+                        currentConnector != null ? currentConnector.getAddress() : "null"),
+                        t);
                 times++;
                 restart();
                 logger.info("restart the connector for next round retry.");
@@ -126,9 +126,8 @@ public class ClusterCanalConnector implements CanalConnector {
                 currentConnector.unsubscribe();
                 return;
             } catch (Throwable t) {
-                logger.warn("something goes wrong when unsubscribing from server:{}\n{}",
+                logger.warn(String.format("something goes wrong when unsubscribing from server:%s",
-                    currentConnector != null ? currentConnector.getAddress() : "null",
+                    currentConnector != null ? currentConnector.getAddress() : "null"), t);
-                    ExceptionUtils.getFullStackTrace(t));
                 times++;
                 restart();
                 logger.info("restart the connector for next round retry.");
@@ -144,9 +143,8 @@ public class ClusterCanalConnector implements CanalConnector {
                 Message msg = currentConnector.get(batchSize);
                 return msg;
             } catch (Throwable t) {
-                logger.warn("something goes wrong when getting data from server:{}\n{}",
+                logger.warn(String.format("something goes wrong when getting data from server:%s",
-                    currentConnector != null ? currentConnector.getAddress() : "null",
+                    currentConnector != null ? currentConnector.getAddress() : "null"), t);
-                    ExceptionUtils.getFullStackTrace(t));
                 times++;
                 restart();
                 logger.info("restart the connector for next round retry.");
@@ -162,9 +160,8 @@ public class ClusterCanalConnector implements CanalConnector {
                 Message msg = currentConnector.get(batchSize, timeout, unit);
                 return msg;
             } catch (Throwable t) {
-                logger.warn("something goes wrong when getting data from server:{}\n{}",
+                logger.warn(String.format("something goes wrong when getting data from server:%s",
-                    currentConnector != null ? currentConnector.getAddress() : "null",
+                    currentConnector != null ? currentConnector.getAddress() : "null"), t);
-                    ExceptionUtils.getFullStackTrace(t));
                 times++;
                 restart();
                 logger.info("restart the connector for next round retry.");
@@ -180,9 +177,8 @@ public class ClusterCanalConnector implements CanalConnector {
                 Message msg = currentConnector.getWithoutAck(batchSize);
                 return msg;
             } catch (Throwable t) {
-                logger.warn("something goes wrong when getWithoutAck data from server:{}\n{}",
+                logger.warn(String.format("something goes wrong when getWithoutAck data from server:%s",
-                    currentConnector.getAddress(),
+                    currentConnector.getAddress()), t);
-                    ExceptionUtils.getFullStackTrace(t));
                 times++;
                 restart();
                 logger.info("restart the connector for next round retry.");
@@ -198,9 +194,8 @@ public class ClusterCanalConnector implements CanalConnector {
                 Message msg = currentConnector.getWithoutAck(batchSize, timeout, unit);
                 return msg;
             } catch (Throwable t) {
-                logger.warn("something goes wrong when getWithoutAck data from server:{}\n{}",
+                logger.warn(String.format("something goes wrong when getWithoutAck data from server:%s",
-                    currentConnector.getAddress(),
+                    currentConnector.getAddress()), t);
-                    ExceptionUtils.getFullStackTrace(t));
                 times++;
                 restart();
                 logger.info("restart the connector for next round retry.");
@@ -216,9 +211,8 @@ public class ClusterCanalConnector implements CanalConnector {
                 currentConnector.rollback(batchId);
                 return;
             } catch (Throwable t) {
-                logger.warn("something goes wrong when rollbacking data from server:{}\n{}",
+                logger.warn(String.format("something goes wrong when rollbacking data from server:%s",
-                    currentConnector.getAddress(),
+                    currentConnector.getAddress()), t);
-                    ExceptionUtils.getFullStackTrace(t));
                 times++;
                 restart();
                 logger.info("restart the connector for next round retry.");
@@ -234,9 +228,8 @@ public class ClusterCanalConnector implements CanalConnector {
                 currentConnector.rollback();
                 return;
             } catch (Throwable t) {
-                logger.warn("something goes wrong when rollbacking data from server:{}\n{}",
+                logger.warn(String.format("something goes wrong when rollbacking data from server:%s",
-                    currentConnector.getAddress(),
+                    currentConnector.getAddress()), t);
-                    ExceptionUtils.getFullStackTrace(t));
                 times++;
                 restart();
                 logger.info("restart the connector for next round retry.");
@@ -253,9 +246,8 @@ public class ClusterCanalConnector implements CanalConnector {
                 currentConnector.ack(batchId);
                 return;
             } catch (Throwable t) {
-                logger.warn("something goes wrong when acking data from server:{}\n{}",
+                logger.warn(String.format("something goes wrong when acking data from server:%s",
-                    currentConnector.getAddress(),
+                    currentConnector.getAddress()), t);
-                    ExceptionUtils.getFullStackTrace(t));
                 times++;
                 restart();
                 logger.info("restart the connector for next round retry.");

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

@@ -363,12 +363,12 @@ public class JsonConversion {
                         buf.append('"').append(text).append('"');
                     } else {
                         text = m_data.getFixString((int) m_length, charsetName);
-                        buf.append('"').append(text).append('"');
+                        buf.append('"').append(escapse(text)).append('"');
                     }
 
                     break;
                 case STRING:
-                    buf.append('"').append(m_string_value).append('"');
+                    buf.append('"').append(escapse(m_string_value)).append('"');
                     break;
                 case ERROR:
                     throw new IllegalArgumentException("illegal json data");
@@ -378,6 +378,21 @@ public class JsonConversion {
         }
     }
 
+    private static StringBuilder escapse(String data) {
+        StringBuilder sb = new StringBuilder(data.length());
+        int endIndex = data.length();
+        for (int i = 0; i < endIndex; ++i) {
+            char c = data.charAt(i);
+            if (c == '"') {
+                sb.append('\\');
+            } else if (c == '\\') {
+                sb.append("\\");
+            }
+            sb.append(c);
+        }
+        return sb;
+    }
+
     public static enum Json_enum_type {
         OBJECT, ARRAY, STRING, INT, UINT, DOUBLE, LITERAL_NULL, LITERAL_TRUE, LITERAL_FALSE, OPAQUE, ERROR
     }

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

@@ -949,8 +949,28 @@ public final class RowsLogBuffer {
                 break;
             }
             case LogEvent.MYSQL_TYPE_JSON: {
-                len = buffer.getUint16();
+                switch (meta) {
-                buffer.forward(meta - 2);
+                    case 1: {
+                        len = buffer.getUint8();
+                        break;
+                    }
+                    case 2: {
+                        len = buffer.getUint16();
+                        break;
+                    }
+                    case 3: {
+                        len = buffer.getUint24();
+                        break;
+                    }
+                    case 4: {
+                        len = (int) buffer.getUint32();
+                        break;
+                    }
+                    default:
+                        throw new IllegalArgumentException("!! Unknown JSON packlen = " + meta);
+                }
+                // len = buffer.getUint16();
+                // buffer.forward(meta - 4);
                 int position = buffer.position();
                 Json_Value jsonValue = JsonConversion.parse_value(buffer.getUint8(), buffer, len - 1, charsetName);
                 StringBuilder builder = new StringBuilder();

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

@@ -4,7 +4,6 @@ import java.io.FileInputStream;
 import java.util.Properties;
 
 import org.apache.commons.lang.StringUtils;
-import org.apache.commons.lang.exception.ExceptionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -21,6 +20,10 @@ public class CanalLauncher {
 
     public static void main(String[] args) throws Throwable {
         try {
+            logger.info("## set default uncaught exception handler");
+            setGlobalUncaughtExceptionHandler();
+
+            logger.info("## load canal configurations");
             String conf = System.getProperty("canal.conf", "classpath:canal.properties");
             Properties properties = new Properties();
             if (conf.startsWith(CLASSPATH_URL_PREFIX)) {
@@ -41,8 +44,7 @@ public class CanalLauncher {
                         logger.info("## stop the canal server");
                         controller.stop();
                     } catch (Throwable e) {
-                        logger.warn("##something goes wrong when stopping canal Server:\n{}",
+                        logger.warn("##something goes wrong when stopping canal Server:", e);
-                            ExceptionUtils.getFullStackTrace(e));
                     } finally {
                         logger.info("## canal server is down.");
                     }
@@ -50,9 +52,17 @@ public class CanalLauncher {
 
             });
         } catch (Throwable e) {
-            logger.error("## Something goes wrong when starting up the canal Server:\n{}",
+            logger.error("## Something goes wrong when starting up the canal Server:", e);
-                ExceptionUtils.getFullStackTrace(e));
             System.exit(0);
         }
     }
+
+    private static void setGlobalUncaughtExceptionHandler() {
+        Thread.setDefaultUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
+            @Override
+            public void uncaughtException(Thread t, Throwable e) {
+                logger.error("UnCaughtException", e);
+            }
+        });
+    }
 }

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

@@ -13,7 +13,6 @@ import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.commons.lang.StringUtils;
-import org.apache.commons.lang.exception.ExceptionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.util.Assert;
@@ -186,7 +185,7 @@ public class SpringInstanceConfigMonitor extends AbstractCanalLifeCycle implemen
 
             logger.info("auto notify start {} successful.", destination);
         } catch (Throwable e) {
-            logger.error("scan add found[{}] but start failed", destination, ExceptionUtils.getFullStackTrace(e));
+            logger.error(String.format("scan add found[%s] but start failed", destination), e);
         }
     }
 
@@ -197,7 +196,7 @@ public class SpringInstanceConfigMonitor extends AbstractCanalLifeCycle implemen
             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));
+            logger.error(String.format("scan delete found[%s] but stop failed", destination), e);
             actions.put(destination, action);// 再重新加回去，下一次scan时再执行删除
         }
     }
@@ -209,9 +208,8 @@ public class SpringInstanceConfigMonitor extends AbstractCanalLifeCycle implemen
                 action.reload(destination);
                 logger.info("auto notify reload {} successful.", destination);
             } catch (Throwable e) {
-                logger.error("scan reload found[{}] but reload failed",
+                logger.error(String.format("scan reload found[%s] but reload failed",
-                    destination,
+                    destination), e);
-                    ExceptionUtils.getFullStackTrace(e));
             }
         }
     }

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

@@ -4,7 +4,6 @@ import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-import org.apache.commons.lang.exception.ExceptionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -70,7 +69,7 @@ public class MysqlConnector {
                 negotiate(channel);
             } catch (Exception e) {
                 disconnect();
-                throw new IOException("connect " + this.address + " failure:" + ExceptionUtils.getStackTrace(e));
+                throw new IOException("connect " + this.address + " failure", e);
             }
         } else {
             logger.error("the channel can't be connected twice.");
@@ -90,7 +89,7 @@ public class MysqlConnector {
                 }
                 logger.info("disConnect MysqlConnection to {}...", address);
             } catch (Exception e) {
-                throw new IOException("disconnect " + this.address + " failure:" + ExceptionUtils.getStackTrace(e));
+                throw new IOException("disconnect " + this.address + " failure", e);
             }
 
             // 执行一次quit
@@ -102,7 +101,7 @@ 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));
+                    throw new IOException("KILL DUMP " + connectionId + " failure", e);
                 } finally {
                     if (connector != null) {
                         connector.disconnect();

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

@@ -21,14 +21,16 @@ import java.net.SocketAddress;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
+import com.alibaba.otter.canal.common.utils.BooleanMutex;
+
 /**
  * @author luoyaogui 实现channel的管理（监听连接、读数据、回收） 2016-12-28
  */
 @SuppressWarnings({ "rawtypes", "deprecation" })
 public abstract class SocketChannelPool {
 
-    private static EventLoopGroup              group     = new NioEventLoopGroup();                         // 非阻塞IO线程组
+    private static EventLoopGroup              group     = new NioEventLoopGroup();                        // 非阻塞IO线程组
-    private static Bootstrap                   boot      = new Bootstrap();                                 // 主
+    private static Bootstrap                   boot      = new Bootstrap();                                // 主
     private static Map<Channel, SocketChannel> chManager = new ConcurrentHashMap<Channel, SocketChannel>();
 
     static {
@@ -53,19 +55,20 @@ public abstract class SocketChannelPool {
 
     public static SocketChannel open(SocketAddress address) throws Exception {
         final SocketChannel socket = new SocketChannel();
+        final BooleanMutex mutex = new BooleanMutex(false);
         boot.connect(address).addListener(new ChannelFutureListener() {
 
             @Override
             public void operationComplete(ChannelFuture arg0) throws Exception {
-                if (arg0.isSuccess()) socket.setChannel(arg0.channel());
+                if (arg0.isSuccess()) {
-                synchronized (socket) {
+                    socket.setChannel(arg0.channel());
-                    socket.notify();
                 }
+
+                mutex.set(true);
             }
         });
-        synchronized (socket) {
+        // wait for complete
-            socket.wait();
+        mutex.get();
-        }
         if (null == socket.getChannel()) {
             throw new IOException("can't create socket!");
         }

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

@@ -1,7 +1,5 @@
 package com.alibaba.otter.canal.example;
 
-import org.apache.commons.lang.exception.ExceptionUtils;
-
 import com.alibaba.otter.canal.client.CanalConnector;
 import com.alibaba.otter.canal.client.CanalConnectors;
 
@@ -41,7 +39,7 @@ public class ClusterCanalClientTest extends AbstractCanalClientTest {
                     logger.info("## stop the canal client");
                     clientTest.stop();
                 } catch (Throwable e) {
-                    logger.warn("##something goes wrong when stopping canal:\n{}", ExceptionUtils.getFullStackTrace(e));
+                    logger.warn("##something goes wrong when stopping canal:", e);
                 } finally {
                     logger.info("## canal client is down.");
                 }

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

@@ -2,8 +2,6 @@ package com.alibaba.otter.canal.example;
 
 import java.net.InetSocketAddress;
 
-import org.apache.commons.lang.exception.ExceptionUtils;
-
 import com.alibaba.otter.canal.client.CanalConnector;
 import com.alibaba.otter.canal.client.CanalConnectors;
 import com.alibaba.otter.canal.common.utils.AddressUtils;
@@ -39,7 +37,7 @@ public class SimpleCanalClientTest extends AbstractCanalClientTest {
                     logger.info("## stop the canal client");
                     clientTest.stop();
                 } catch (Throwable e) {
-                    logger.warn("##something goes wrong when stopping canal:\n{}", ExceptionUtils.getFullStackTrace(e));
+                    logger.warn("##something goes wrong when stopping canal:", e);
                 } finally {
                     logger.info("## canal client is down.");
                 }

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

@@ -275,7 +275,7 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
         parseThread.setUncaughtExceptionHandler(handler);
         parseThread.setName(String.format("destination = %s , address = %s , EventParser",
             destination,
-            runningInfo == null ? null : runningInfo.getAddress().toString()));
+            runningInfo == null ? null : runningInfo.getAddress()));
         parseThread.start();
     }
 
@@ -430,7 +430,7 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
                     }
 
                 } catch (Throwable e) {
-                    logger.warn("heartBeat run failed " + ExceptionUtils.getStackTrace(e));
+                    logger.warn("heartBeat run failed ", e);
                 }
             }
 

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

@@ -12,7 +12,6 @@ import java.util.Properties;
 
 import com.mysql.jdbc.Driver;
 import org.apache.commons.lang.StringUtils;
-import org.apache.commons.lang.exception.ExceptionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -235,25 +234,25 @@ public class MysqlConnection implements ErosaConnection {
         try {
             update("set wait_timeout=9999999");
         } catch (Exception e) {
-            logger.warn(ExceptionUtils.getFullStackTrace(e));
+            logger.warn("update wait_timeout failed", e);
         }
         try {
             update("set net_write_timeout=1800");
         } catch (Exception e) {
-            logger.warn(ExceptionUtils.getFullStackTrace(e));
+            logger.warn("update net_write_timeout failed", e);
         }
 
         try {
             update("set net_read_timeout=1800");
         } catch (Exception e) {
-            logger.warn(ExceptionUtils.getFullStackTrace(e));
+            logger.warn("update net_read_timeout failed", e);
         }
 
         try {
             // 设置服务端返回结果时不做编码转化，直接按照数据库的二进制编码进行发送，由客户端自己根据需求进行编码转化
             update("set names 'binary'");
         } catch (Exception e) {
-            logger.warn(ExceptionUtils.getFullStackTrace(e));
+            logger.warn("update names failed", e);
         }
 
         try {
@@ -263,7 +262,7 @@ public class MysqlConnection implements ErosaConnection {
             // 但也不能乱设置，需要和mysql server的checksum配置一致，不然RotateLogEvent会出现乱码
             update("set @master_binlog_checksum= '@@global.binlog_checksum'");
         } catch (Exception e) {
-            logger.warn(ExceptionUtils.getFullStackTrace(e));
+            logger.warn("update master_binlog_checksum failed", e);
         }
 
         try {
@@ -272,7 +271,7 @@ public class MysqlConnection implements ErosaConnection {
             update("set @slave_uuid=uuid()");
         } catch (Exception e) {
             if (!StringUtils.contains(e.getMessage(), "Unknown system variable")) {
-                logger.warn(ExceptionUtils.getFullStackTrace(e));
+                logger.warn("update slave_uuid failed", e);
             }
         }
 
@@ -280,7 +279,7 @@ public class MysqlConnection implements ErosaConnection {
             // mariadb针对特殊的类型，需要设置session变量
             update("SET @mariadb_slave_capability='" + LogEvent.MARIA_SLAVE_CAPABILITY_MINE + "'");
         } catch (Exception e) {
-            logger.warn(ExceptionUtils.getFullStackTrace(e));
+            logger.warn("update mariadb_slave_capability failed", e);
         }
     }
 

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

@@ -10,7 +10,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 
 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;
@@ -243,19 +242,19 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
                     ((HeartBeatCallback) haController).onFailed(e);
                 }
                 reconnect = true;
-                logger.warn("connect failed by " + ExceptionUtils.getStackTrace(e));
+                logger.warn("connect failed by ", e);
             } catch (IOException e) {
                 if (haController != null && haController instanceof HeartBeatCallback) {
                     ((HeartBeatCallback) haController).onFailed(e);
                 }
                 reconnect = true;
-                logger.warn("connect failed by " + ExceptionUtils.getStackTrace(e));
+                logger.warn("connect failed by ", e);
             } catch (Throwable e) {
                 if (haController != null && haController instanceof HeartBeatCallback) {
                     ((HeartBeatCallback) haController).onFailed(e);
                 }
                 reconnect = true;
-                logger.warn("connect failed by " + ExceptionUtils.getStackTrace(e));
+                logger.warn("connect failed by ", e);
             }
 
         }
@@ -555,9 +554,8 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
                     return entryPosition;
                 }
             } catch (Exception e) {
-                logger.warn("the binlogfile:{} doesn't exist, to continue to search the next binlogfile , caused by {}",
+                logger.warn(String.format("the binlogfile:%s doesn't exist, to continue to search the next binlogfile , caused by",
-                    startSearchBinlogFile,
+                    startSearchBinlogFile), e);
-                    ExceptionUtils.getFullStackTrace(e));
                 int binlogSeqNum = Integer.parseInt(startSearchBinlogFile.substring(startSearchBinlogFile.indexOf(".") + 1));
                 if (binlogSeqNum <= 1) {
                     logger.warn("Didn't find the corresponding binlog files");