fixed issue #4491 , upgrade druid 1.2.15

* 测试类报错

* add proxy for SPI

admin/admin-web/src/main/resources/canal-template.properties

@@ -25,7 +25,7 @@ canal.zkServers =
 # flush data to zk
 canal.zookeeper.flush.period = 1000
 canal.withoutNetty = false
-# tcp, kafka, rocketMQ, rabbitMQ
+# tcp, kafka, rocketMQ, rabbitMQ, pulsarMQ
 canal.serverMode = tcp
 # flush meta cursor/parse position to file
 canal.file.data.dir = ${canal.conf.dir}
@@ -96,7 +96,7 @@ canal.instance.tsdb.snapshot.expire = 360
 #################################################
 ######### 		destinations		#############
 #################################################
-canal.destinations = 
+canal.destinations =
 # conf root dir
 canal.conf.dir = ../conf
 # auto scan instance dir add/remove and start/stop instance
@@ -148,8 +148,13 @@ kafka.max.in.flight.requests.per.connection = 1
 kafka.retries = 0
 
 kafka.kerberos.enable = false
-kafka.kerberos.krb5.file = "../conf/kerberos/krb5.conf"
-kafka.kerberos.jaas.file = "../conf/kerberos/jaas.conf"
+kafka.kerberos.krb5.file = ../conf/kerberos/krb5.conf
+kafka.kerberos.jaas.file = ../conf/kerberos/jaas.conf
+
+# sasl demo
+# kafka.sasl.jaas.config = org.apache.kafka.common.security.scram.ScramLoginModule required \\n username=\"alice\" \\npassword="alice-secret\";
+# kafka.sasl.mechanism = SCRAM-SHA-512
+# kafka.security.protocol = SASL_PLAINTEXT
 
 ##################################################
 ######### 		    RocketMQ	     #############
@@ -161,7 +166,7 @@ rocketmq.namespace =
 rocketmq.namesrv.addr = 127.0.0.1:9876
 rocketmq.retry.times.when.send.failed = 0
 rocketmq.vip.channel.enabled = false
-rocketmq.tag = 
+rocketmq.tag =
 
 ##################################################
 ######### 		    RabbitMQ	     #############
@@ -171,4 +176,12 @@ rabbitmq.virtual.host =
 rabbitmq.exchange =
 rabbitmq.username =
 rabbitmq.password =
-rabbitmq.deliveryMode =
+rabbitmq.deliveryMode =
+
+
+##################################################
+######### 		      Pulsar         #############
+##################################################
+pulsarmq.serverUrl =
+pulsarmq.roleToken =
+pulsarmq.topicTenantPrefix =

client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/ProxyOuterAdapter.java

@@ -0,0 +1,87 @@
+package com.alibaba.otter.canal.client.adapter;
+
+import com.alibaba.otter.canal.client.adapter.support.Dml;
+import com.alibaba.otter.canal.client.adapter.support.EtlResult;
+import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+public class ProxyOuterAdapter implements OuterAdapter {
+
+    private OuterAdapter outerAdapter;
+
+    public ProxyOuterAdapter(OuterAdapter outerAdapter) {
+        this.outerAdapter = outerAdapter;
+    }
+
+    private ClassLoader changeCL() {
+        ClassLoader cl = Thread.currentThread().getContextClassLoader();
+        Thread.currentThread().setContextClassLoader(outerAdapter.getClass().getClassLoader());
+        return cl;
+    }
+
+    private void revertCL(ClassLoader cl) {
+        Thread.currentThread().setContextClassLoader(cl);
+    }
+
+    @Override
+    public void init(OuterAdapterConfig configuration, Properties envProperties) {
+        ClassLoader cl = changeCL();
+        try {
+            outerAdapter.init(configuration, envProperties);
+        } finally {
+            revertCL(cl);
+        }
+    }
+
+    @Override
+    public void sync(List<Dml> dmls) {
+        ClassLoader cl = changeCL();
+        try {
+            outerAdapter.sync(dmls);
+        } finally {
+            revertCL(cl);
+        }
+    }
+
+    @Override
+    public void destroy() {
+        ClassLoader cl = changeCL();
+        try {
+            outerAdapter.destroy();
+        } finally {
+            revertCL(cl);
+        }
+    }
+
+    @Override
+    public EtlResult etl(String task, List<String> params) {
+        ClassLoader cl = changeCL();
+        try {
+            return OuterAdapter.super.etl(task, params);
+        } finally {
+            revertCL(cl);
+        }
+    }
+
+    @Override
+    public Map<String, Object> count(String task) {
+        ClassLoader cl = changeCL();
+        try {
+            return OuterAdapter.super.count(task);
+        } finally {
+            revertCL(cl);
+        }
+    }
+
+    @Override
+    public String getDestination(String task) {
+        ClassLoader cl = changeCL();
+        try {
+            return OuterAdapter.super.getDestination(task);
+        } finally {
+            revertCL(cl);
+        }
+    }
+}

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AdapterProcessor.java

@@ -1,5 +1,6 @@
 package com.alibaba.otter.canal.adapter.launcher.loader;
 
+import com.alibaba.otter.canal.connector.core.spi.ProxyCanalMsgConsumer;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Properties;
@@ -26,33 +27,33 @@ import com.alibaba.otter.canal.connector.core.spi.ExtensionLoader;
 
 /**
  * 适配处理器
- * 
+ *
  * @author rewerma 2020-02-01
  * @version 1.0.0
  */
 public class AdapterProcessor {
 
-    private static final Logger             logger                    = LoggerFactory.getLogger(AdapterProcessor.class);
+    private static final Logger logger = LoggerFactory.getLogger(AdapterProcessor.class);
 
-    private static final String             CONNECTOR_SPI_DIR         = "/plugin";
-    private static final String             CONNECTOR_STANDBY_SPI_DIR = "/canal-adapter/plugin";
+    private static final String CONNECTOR_SPI_DIR = "/plugin";
+    private static final String CONNECTOR_STANDBY_SPI_DIR = "/canal-adapter/plugin";
 
-    private CanalMsgConsumer                canalMsgConsumer;
+    private CanalMsgConsumer canalMsgConsumer;
 
-    private String                          canalDestination;                                                           // canal实例
-    private String                          groupId                   = null;                                           // groupId
-    private List<List<OuterAdapter>>        canalOuterAdapters;                                                         // 外部适配器
-    private CanalClientConfig               canalClientConfig;                                                          // 配置
-    private ExecutorService                 groupInnerExecutorService;                                                  // 组内工作线程池
-    private volatile boolean                running                   = false;                                          // 是否运行中
-    private Thread                          thread                    = null;
-    private Thread.UncaughtExceptionHandler handler                   = (t, e) -> logger
+    private String canalDestination;                                                           // canal实例
+    private String groupId = null;                                           // groupId
+    private List<List<OuterAdapter>> canalOuterAdapters;                                                         // 外部适配器
+    private CanalClientConfig canalClientConfig;                                                          // 配置
+    private ExecutorService groupInnerExecutorService;                                                  // 组内工作线程池
+    private volatile boolean running = false;                                          // 是否运行中
+    private Thread thread = null;
+    private Thread.UncaughtExceptionHandler handler = (t, e) -> logger
         .error("parse events has an error", e);
 
-    private SyncSwitch                      syncSwitch;
+    private SyncSwitch syncSwitch;
 
     public AdapterProcessor(CanalClientConfig canalClientConfig, String destination, String groupId,
-                            List<List<OuterAdapter>> canalOuterAdapters){
+        List<List<OuterAdapter>> canalOuterAdapters) {
         this.canalClientConfig = canalClientConfig;
         this.canalDestination = destination;
         this.groupId = groupId;
@@ -63,17 +64,15 @@ public class AdapterProcessor {
 
         // load connector consumer
         ExtensionLoader<CanalMsgConsumer> loader = new ExtensionLoader<>(CanalMsgConsumer.class);
-        canalMsgConsumer = loader
-            .getExtension(canalClientConfig.getMode().toLowerCase(),destination ,CONNECTOR_SPI_DIR, CONNECTOR_STANDBY_SPI_DIR);
+        canalMsgConsumer = new ProxyCanalMsgConsumer(loader
+            .getExtension(canalClientConfig.getMode().toLowerCase(), destination, CONNECTOR_SPI_DIR,
+                CONNECTOR_STANDBY_SPI_DIR));
 
         Properties properties = canalClientConfig.getConsumerProperties();
         properties.put(CanalConstants.CANAL_MQ_FLAT_MESSAGE, canalClientConfig.getFlatMessage());
         properties.put(CanalConstants.CANAL_ALIYUN_ACCESS_KEY, canalClientConfig.getAccessKey());
         properties.put(CanalConstants.CANAL_ALIYUN_SECRET_KEY, canalClientConfig.getSecretKey());
-        ClassLoader cl = Thread.currentThread().getContextClassLoader();
-        Thread.currentThread().setContextClassLoader(canalMsgConsumer.getClass().getClassLoader());
         canalMsgConsumer.init(properties, canalDestination, groupId);
-        Thread.currentThread().setContextClassLoader(cl);
     }
 
     public void start() {
@@ -171,7 +170,7 @@ public class AdapterProcessor {
         }
 
         int retry = canalClientConfig.getRetries() == null
-                    || canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries();
+            || canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries();
         if (retry == -1) {
             // 重试次数-1代表异常时一直阻塞重试
             retry = Integer.MAX_VALUE;

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java

@@ -1,5 +1,6 @@
 package com.alibaba.otter.canal.adapter.launcher.loader;
 
+import com.alibaba.otter.canal.client.adapter.ProxyOuterAdapter;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -87,11 +88,8 @@ public class CanalAdapterLoader {
     private void loadAdapter(OuterAdapterConfig config, List<OuterAdapter> canalOutConnectors) {
         try {
             OuterAdapter adapter;
-            adapter = loader.getExtension(config.getName(), config.getKey());
+            adapter = new ProxyOuterAdapter(loader.getExtension(config.getName(), config.getKey()));
 
-            ClassLoader cl = Thread.currentThread().getContextClassLoader();
-            // 替换ClassLoader
-            Thread.currentThread().setContextClassLoader(adapter.getClass().getClassLoader());
             Environment env = (Environment) SpringContext.getBean(Environment.class);
             Properties evnProperties = null;
             if (env instanceof StandardEnvironment) {
@@ -109,7 +107,6 @@ public class CanalAdapterLoader {
                 }
             }
             adapter.init(config, evnProperties);
-            Thread.currentThread().setContextClassLoader(cl);
             canalOutConnectors.add(adapter);
             logger.info("Load canal adapter: {} succeed", config.getName());
         } catch (Exception e) {

client-adapter/pom.xml

@@ -53,7 +53,7 @@
     <repositories>
         <repository>
             <id>central</id>
-            <url>http://repo1.maven.org/maven2</url>
+            <url>https://repo1.maven.org/maven2</url>
             <releases>
                 <enabled>true</enabled>
             </releases>
@@ -63,7 +63,7 @@
         </repository>
         <repository>
             <id>java.net</id>
-            <url>http://download.java.net/maven/2/</url>
+            <url>https://download.java.net/maven/2/</url>
             <releases>
                 <enabled>true</enabled>
             </releases>
@@ -73,7 +73,7 @@
         </repository>
         <repository>
             <id>aliyun</id>
-            <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
+            <url>https://maven.aliyun.com/nexus/content/groups/public/</url>
             <releases>
                 <enabled>true</enabled>
             </releases>

client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java

@@ -58,6 +58,7 @@ public class RdbEtlService extends AbstractEtlService {
             Map<String, String> columnsMap = new LinkedHashMap<>();
             Map<String, Integer> columnType = new LinkedHashMap<>();
             DruidDataSource dataSource = (DruidDataSource) srcDS;
+            String backtick = SyncUtil.getBacktickByDbType(dataSource.getDbType());
 
             Util.sqlRS(targetDS,
                 "SELECT * FROM " + SyncUtil.getDbTableName(dbMapping, dataSource.getDbType()) + " LIMIT 1 ",
@@ -91,7 +92,7 @@ public class RdbEtlService extends AbstractEtlService {
                         .append(SyncUtil.getDbTableName(dbMapping, dataSource.getDbType()))
                         .append(" (");
                     columnsMap
-                        .forEach((targetColumnName, srcColumnName) -> insertSql.append(targetColumnName).append(","));
+                        .forEach((targetColumnName, srcColumnName) -> insertSql.append(backtick).append(targetColumnName).append(backtick).append(","));
 
                     int len = insertSql.length();
                     insertSql.delete(len - 1, len).append(") VALUES (");
@@ -101,6 +102,7 @@ public class RdbEtlService extends AbstractEtlService {
                     }
                     len = insertSql.length();
                     insertSql.delete(len - 1, len).append(")");
+                    logger.info("executeSqlImport sql:{}",insertSql.toString());
                     try (Connection connTarget = targetDS.getConnection();
                             PreparedStatement pstmt = connTarget.prepareStatement(insertSql.toString())) {
                         connTarget.setAutoCommit(false);
@@ -114,7 +116,7 @@ public class RdbEtlService extends AbstractEtlService {
                             Map<String, Object> pkVal = new LinkedHashMap<>();
                             StringBuilder deleteSql = new StringBuilder(
                                 "DELETE FROM " + SyncUtil.getDbTableName(dbMapping, dataSource.getDbType()) + " WHERE ");
-                            appendCondition(dbMapping, deleteSql, pkVal, rs);
+                            appendCondition(dbMapping, deleteSql, pkVal, rs, backtick);
                             try (PreparedStatement pstmt2 = connTarget.prepareStatement(deleteSql.toString())) {
                                 int k = 1;
                                 for (Object val : pkVal.values()) {
@@ -132,7 +134,6 @@ public class RdbEtlService extends AbstractEtlService {
                                 }
 
                                 Integer type = columnType.get(targetClolumnName.toLowerCase());
-
                                 Object value = rs.getObject(srcColumnName);
                                 if (value != null) {
                                     SyncUtil.setPStmt(type, pstmt, value, i);
@@ -180,7 +181,7 @@ public class RdbEtlService extends AbstractEtlService {
      * 拼接目标表主键where条件
      */
     private static void appendCondition(DbMapping dbMapping, StringBuilder sql, Map<String, Object> values,
-                                        ResultSet rs) throws SQLException {
+                                        ResultSet rs, String backtick) throws SQLException {
         // 拼接主键
         for (Map.Entry<String, String> entry : dbMapping.getTargetPk().entrySet()) {
             String targetColumnName = entry.getKey();
@@ -188,7 +189,7 @@ public class RdbEtlService extends AbstractEtlService {
             if (srcColumnName == null) {
                 srcColumnName = targetColumnName;
             }
-            sql.append(targetColumnName).append("=? AND ");
+            sql.append(backtick).append(targetColumnName).append(backtick).append("=? AND ");
             values.put(targetColumnName, rs.getObject(srcColumnName));
         }
         int len = sql.length();

client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java

@@ -1,5 +1,6 @@
 package com.alibaba.otter.canal.client.kafka;
 
+import com.alibaba.fastjson.JSON;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
@@ -10,7 +11,6 @@ import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.serialization.StringDeserializer;
 
-import com.alibaba.fastjson2.JSON;
 import com.alibaba.otter.canal.client.CanalMQConnector;
 import com.alibaba.otter.canal.client.impl.SimpleCanalConnector;
 import com.alibaba.otter.canal.protocol.FlatMessage;
@@ -43,7 +43,7 @@ public class KafkaCanalConnector implements CanalMQConnector {
     private Map<Integer, Long>               currentOffsets = new ConcurrentHashMap<>();
 
     public KafkaCanalConnector(String servers, String topic, Integer partition, String groupId, Integer batchSize,
-                               boolean flatMessage){
+        boolean flatMessage){
         this.topic = topic;
         this.partition = partition;
         this.flatMessage = flatMessage;
@@ -61,11 +61,11 @@ public class KafkaCanalConnector implements CanalMQConnector {
             batchSize = 100;
         }
         properties.put("max.poll.records", batchSize.toString());
-        properties.put("key.deserializer", StringDeserializer.class.getName());
+        properties.put("key.deserializer", StringDeserializer.class);
         if (!flatMessage) {
-            properties.put("value.deserializer", MessageDeserializer.class.getName());
+            properties.put("value.deserializer", MessageDeserializer.class);
         } else {
-            properties.put("value.deserializer", StringDeserializer.class.getName());
+            properties.put("value.deserializer", StringDeserializer.class);
         }
     }
 
@@ -183,14 +183,13 @@ public class KafkaCanalConnector implements CanalMQConnector {
 
         ConsumerRecords<String, Message> records = kafkaConsumer.poll(unit.toMillis(timeout));
 
-        currentOffsets.clear();
-        for (TopicPartition topicPartition : records.partitions()) {
-            currentOffsets.put(topicPartition.partition(), kafkaConsumer.position(topicPartition));
-        }
-
         if (!records.isEmpty()) {
+            currentOffsets.clear();
             List<Message> messages = new ArrayList<>();
             for (ConsumerRecord<String, Message> record : records) {
+                if (currentOffsets.get(record.partition()) == null) {
+                    currentOffsets.put(record.partition(), record.offset());
+                }
                 messages.add(record.value());
             }
             return messages;
@@ -221,14 +220,13 @@ public class KafkaCanalConnector implements CanalMQConnector {
 
         ConsumerRecords<String, String> records = kafkaConsumer2.poll(unit.toMillis(timeout));
 
-        currentOffsets.clear();
-        for (TopicPartition topicPartition : records.partitions()) {
-            currentOffsets.put(topicPartition.partition(), kafkaConsumer2.position(topicPartition));
-        }
-
         if (!records.isEmpty()) {
+            currentOffsets.clear();
             List<FlatMessage> flatMessages = new ArrayList<>();
             for (ConsumerRecord<String, String> record : records) {
+                if (currentOffsets.get(record.partition()) == null) {
+                    currentOffsets.put(record.partition(), record.offset());
+                }
                 String flatMessageJson = record.value();
                 FlatMessage flatMessage = JSON.parseObject(flatMessageJson, FlatMessage.class);
                 flatMessages.add(flatMessage);
@@ -248,12 +246,14 @@ public class KafkaCanalConnector implements CanalMQConnector {
         // 回滚所有分区
         if (kafkaConsumer != null) {
             for (Map.Entry<Integer, Long> entry : currentOffsets.entrySet()) {
-                kafkaConsumer.seek(new TopicPartition(topic, entry.getKey()), entry.getValue() - 1);
+                kafkaConsumer.seek(new TopicPartition(topic, entry.getKey()), currentOffsets.get(entry.getKey()));
+                kafkaConsumer.commitSync();
             }
         }
         if (kafkaConsumer2 != null) {
             for (Map.Entry<Integer, Long> entry : currentOffsets.entrySet()) {
-                kafkaConsumer2.seek(new TopicPartition(topic, entry.getKey()), entry.getValue() - 1);
+                kafkaConsumer2.seek(new TopicPartition(topic, entry.getKey()), currentOffsets.get(entry.getKey()));
+                kafkaConsumer.commitSync();
             }
         }
     }
@@ -324,3 +324,4 @@ public class KafkaCanalConnector implements CanalMQConnector {
     }
 
 }
+

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

@@ -38,7 +38,7 @@ public class ServerRunningMonitor extends AbstractCanalLifeCycle {
     private ServerRunningData          serverData;
     // 当前实际运行的节点状态信息
     private volatile ServerRunningData activeData;
-    private ScheduledExecutorService   delayExector = Executors.newScheduledThreadPool(1);
+    private ScheduledExecutorService   delayExecutor;
     private int                        delayTime    = 5;
     private ServerRunningListener      listener;
 
@@ -73,7 +73,7 @@ public class ServerRunningMonitor extends AbstractCanalLifeCycle {
                     initRunning();
                 } else {
                     // 否则就是等待delayTime，避免因网络瞬端或者zk异常，导致出现频繁的切换操作
-                    delayExector.schedule(() -> initRunning(), delayTime, TimeUnit.SECONDS);
+                    delayExecutor.schedule(() -> initRunning(), delayTime, TimeUnit.SECONDS);
                 }
             }
 
@@ -90,6 +90,7 @@ public class ServerRunningMonitor extends AbstractCanalLifeCycle {
         try {
             processStart();
             if (zkClient != null) {
+                delayExecutor = Executors.newScheduledThreadPool(1);
                 // 如果需要尽可能释放instance资源，不需要监听running节点，不然即使stop了这台机器，另一台机器立马会start
                 String path = ZookeeperPathUtils.getDestinationServerRunning(destination);
                 zkClient.subscribeDataChanges(path, dataListener);
@@ -122,6 +123,10 @@ public class ServerRunningMonitor extends AbstractCanalLifeCycle {
         if (zkClient != null) {
             String path = ZookeeperPathUtils.getDestinationServerRunning(destination);
             zkClient.unsubscribeDataChanges(path, dataListener);
+            if (delayExecutor != null) {
+                delayExecutor.shutdown();
+                delayExecutor = null;
+            }
 
             releaseRunning(); // 尝试一下release
         } else {

connector/core/src/main/java/com/alibaba/otter/canal/connector/core/filter/AviaterRegexFilter.java

@@ -107,11 +107,11 @@ public class AviaterRegexFilter {
     private List<String> completionPattern(List<String> patterns) {
         List<String> result = new ArrayList<>();
         for (String pattern : patterns) {
-            StringBuffer stringBuffer = new StringBuffer();
-            stringBuffer.append("^");
-            stringBuffer.append(pattern);
-            stringBuffer.append("$");
-            result.add(stringBuffer.toString());
+            StringBuilder stringBuilder = new StringBuilder();
+            stringBuilder.append("^");
+            stringBuilder.append(pattern);
+            stringBuilder.append("$");
+            result.add(stringBuilder.toString());
         }
         return result;
     }

connector/core/src/main/java/com/alibaba/otter/canal/connector/core/producer/MQMessageUtils.java

@@ -374,6 +374,7 @@ public class MQMessageUtils {
             flatMessage.setEs(entry.getHeader().getExecuteTime());
             flatMessage.setTs(System.currentTimeMillis());
             flatMessage.setSql(rowChange.getSql());
+            flatMessage.setGtid(entry.getHeader().getGtid());
 
             if (!rowChange.getIsDdl()) {
                 Map<String, Integer> sqlType = new LinkedHashMap<>();
@@ -432,9 +433,7 @@ public class MQMessageUtils {
                             }
                         }
                         // update操作将记录修改前的值
-                        if (!rowOld.isEmpty()) {
-                            old.add(rowOld);
-                        }
+                        old.add(rowOld);
                     }
                 }
                 if (!sqlType.isEmpty()) {
@@ -526,6 +525,7 @@ public class MQMessageUtils {
                             flatMessageTmp.setEs(flatMessage.getEs());
                             flatMessageTmp.setTs(flatMessage.getTs());
                             flatMessageTmp.setPkNames(flatMessage.getPkNames());
+                            flatMessageTmp.setGtid(flatMessage.getGtid());
                         }
                         List<Map<String, String>> data = flatMessageTmp.getData();
                         if (data == null) {

connector/core/src/main/java/com/alibaba/otter/canal/connector/core/spi/ProxyCanalMQProducer.java

@@ -0,0 +1,66 @@
+package com.alibaba.otter.canal.connector.core.spi;
+
+import com.alibaba.otter.canal.connector.core.config.MQProperties;
+import com.alibaba.otter.canal.connector.core.producer.MQDestination;
+import com.alibaba.otter.canal.connector.core.util.Callback;
+import com.alibaba.otter.canal.protocol.Message;
+import java.util.Properties;
+
+public class ProxyCanalMQProducer implements CanalMQProducer {
+
+    private CanalMQProducer canalMQProducer;
+
+    public ProxyCanalMQProducer(CanalMQProducer canalMQProducer) {
+        this.canalMQProducer = canalMQProducer;
+    }
+
+    private ClassLoader changeCL() {
+        ClassLoader cl = Thread.currentThread().getContextClassLoader();
+        Thread.currentThread().setContextClassLoader(canalMQProducer.getClass().getClassLoader());
+        return cl;
+    }
+
+    private void revertCL(ClassLoader cl) {
+        Thread.currentThread().setContextClassLoader(cl);
+    }
+
+    @Override
+    public void init(Properties properties) {
+        ClassLoader cl = changeCL();
+        try {
+            canalMQProducer.init(properties);
+        } finally {
+            revertCL(cl);
+        }
+    }
+
+    @Override
+    public MQProperties getMqProperties() {
+        ClassLoader cl = changeCL();
+        try {
+            return canalMQProducer.getMqProperties();
+        } finally {
+            revertCL(cl);
+        }
+    }
+
+    @Override
+    public void send(MQDestination canalDestination, Message message, Callback callback) {
+        ClassLoader cl = changeCL();
+        try {
+            canalMQProducer.send(canalDestination, message, callback);
+        } finally {
+            revertCL(cl);
+        }
+    }
+
+    @Override
+    public void stop() {
+        ClassLoader cl = changeCL();
+        try {
+            canalMQProducer.stop();
+        } finally {
+            revertCL(cl);
+        }
+    }
+}

connector/core/src/main/java/com/alibaba/otter/canal/connector/core/spi/ProxyCanalMsgConsumer.java

@@ -0,0 +1,86 @@
+package com.alibaba.otter.canal.connector.core.spi;
+
+import com.alibaba.otter.canal.connector.core.consumer.CommonMessage;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+public class ProxyCanalMsgConsumer implements CanalMsgConsumer {
+
+    private CanalMsgConsumer canalMsgConsumer;
+
+    public ProxyCanalMsgConsumer(CanalMsgConsumer canalMsgConsumer) {
+        this.canalMsgConsumer = canalMsgConsumer;
+    }
+
+    private ClassLoader changeCL() {
+        ClassLoader cl = Thread.currentThread().getContextClassLoader();
+        Thread.currentThread().setContextClassLoader(canalMsgConsumer.getClass().getClassLoader());
+        return cl;
+    }
+
+    private void revertCL(ClassLoader cl) {
+        Thread.currentThread().setContextClassLoader(cl);
+    }
+
+
+    @Override
+    public void init(Properties properties, String topic, String groupId) {
+        ClassLoader cl = changeCL();
+        try {
+            canalMsgConsumer.init(properties, topic, groupId);
+        } finally {
+            revertCL(cl);
+        }
+    }
+
+    @Override
+    public void connect() {
+        ClassLoader cl = changeCL();
+        try {
+            canalMsgConsumer.connect();
+        } finally {
+            revertCL(cl);
+        }
+    }
+
+    @Override
+    public List<CommonMessage> getMessage(Long timeout, TimeUnit unit) {
+        ClassLoader cl = changeCL();
+        try {
+            return canalMsgConsumer.getMessage(timeout, unit);
+        } finally {
+            revertCL(cl);
+        }
+    }
+
+    @Override
+    public void ack() {
+        ClassLoader cl = changeCL();
+        try {
+            canalMsgConsumer.ack();
+        } finally {
+            revertCL(cl);
+        }
+    }
+
+    @Override
+    public void rollback() {
+        ClassLoader cl = changeCL();
+        try {
+            canalMsgConsumer.rollback();
+        } finally {
+            revertCL(cl);
+        }
+    }
+
+    @Override
+    public void disconnect() {
+        ClassLoader cl = changeCL();
+        try {
+            canalMsgConsumer.disconnect();
+        } finally {
+            revertCL(cl);
+        }
+    }
+}

connector/pom.xml

@@ -39,7 +39,7 @@
     <repositories>
         <repository>
             <id>central</id>
-            <url>http://repo1.maven.org/maven2</url>
+            <url>https://repo1.maven.org/maven2</url>
             <releases>
                 <enabled>true</enabled>
             </releases>
@@ -49,7 +49,7 @@
         </repository>
         <repository>
             <id>java.net</id>
-            <url>http://download.java.net/maven/2/</url>
+            <url>https://download.java.net/maven/2/</url>
             <releases>
                 <enabled>true</enabled>
             </releases>
@@ -59,7 +59,7 @@
         </repository>
         <repository>
             <id>aliyun</id>
-            <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
+            <url>https://maven.aliyun.com/nexus/content/groups/public/</url>
             <releases>
                 <enabled>true</enabled>
             </releases>

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

@@ -21,6 +21,7 @@ public final class CharsetConversion {
         protected final String mysqlCharset;
         protected final String mysqlCollation;
         protected final String javaCharset;
+        protected final Charset charset;
 
         Entry(final int id, String mysqlCharset, // NL
               String mysqlCollation, String javaCharset){
@@ -28,6 +29,7 @@ public final class CharsetConversion {
             this.mysqlCharset = mysqlCharset;
             this.mysqlCollation = mysqlCollation;
             this.javaCharset = javaCharset;
+            this.charset = Charset.isSupported(javaCharset) ? Charset.forName(javaCharset) : null;
         }
     }
 
@@ -390,6 +392,23 @@ public final class CharsetConversion {
         }
     }
 
+    public static Charset getNioCharset(final int id) {
+        Entry entry = getEntry(id);
+
+        if (entry != null) {
+            if (entry.charset != null) {
+                return entry.charset;
+            } else {
+                logger.warn("Unknown java charset for: id = " + id + ", name = " + entry.mysqlCharset + ", coll = "
+                            + entry.mysqlCollation);
+                return null;
+            }
+        } else {
+            logger.warn("Unexpect mysql charset: " + id);
+            return null;
+        }
+    }
+
     public static void main(String[] args) {
         for (int i = 0; i < entries.length; i++) {
             Entry entry = entries[i];

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

@@ -1,5 +1,7 @@
 package com.taobao.tddl.dbsync.binlog;
 
+import java.nio.charset.Charset;
+
 import static com.taobao.tddl.dbsync.binlog.event.RowsLogBuffer.appendNumber2;
 import static com.taobao.tddl.dbsync.binlog.event.RowsLogBuffer.appendNumber4;
 import static com.taobao.tddl.dbsync.binlog.event.RowsLogBuffer.usecondsToStr;
@@ -57,23 +59,27 @@ public class JsonConversion {
     public static final int  VALUE_ENTRY_SIZE_LARGE  = (1 + LARGE_OFFSET_SIZE);
 
     public static Json_Value parse_value(int type, LogBuffer buffer, long len, String charsetName) {
+        return parse_value(type, buffer, len, Charset.forName(charsetName));
+    }
+
+    public static Json_Value parse_value(int type, LogBuffer buffer, long len, Charset charset) {
         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, charsetName);
+                return parse_array_or_object(Json_enum_type.OBJECT, buffer, len, false, charset);
             case JSONB_TYPE_LARGE_OBJECT:
-                return parse_array_or_object(Json_enum_type.OBJECT, buffer, len, true, charsetName);
+                return parse_array_or_object(Json_enum_type.OBJECT, buffer, len, true, charset);
             case JSONB_TYPE_SMALL_ARRAY:
-                return parse_array_or_object(Json_enum_type.ARRAY, buffer, len, false, charsetName);
+                return parse_array_or_object(Json_enum_type.ARRAY, buffer, len, false, charset);
             case JSONB_TYPE_LARGE_ARRAY:
-                return parse_array_or_object(Json_enum_type.ARRAY, buffer, len, true, charsetName);
+                return parse_array_or_object(Json_enum_type.ARRAY, buffer, len, true, charset);
             default:
-                return parse_scalar(type, buffer, len, charsetName);
+                return parse_scalar(type, buffer, len, charset);
         }
     }
 
     private static Json_Value parse_array_or_object(Json_enum_type type, LogBuffer buffer, long len, boolean large,
-                                                    String charsetName) {
+                                                    Charset charset) {
         long offset_size = large ? LARGE_OFFSET_SIZE : SMALL_OFFSET_SIZE;
         if (len < 2 * offset_size) {
             throw new IllegalArgumentException("illegal json data");
@@ -100,7 +106,7 @@ public class JsonConversion {
         return large ? buffer.getUint32() : buffer.getUint16();
     }
 
-    private static Json_Value parse_scalar(int type, LogBuffer buffer, long len, String charsetName) {
+    private static Json_Value parse_scalar(int type, LogBuffer buffer, long len, Charset charset) {
         switch (type) {
             case JSONB_TYPE_LITERAL:
                 /* purecov: inspected */
@@ -156,7 +162,7 @@ public class JsonConversion {
                 }
                 return new Json_Value(Json_enum_type.STRING, buffer.rewind()
                     .forward((int) n)
-                    .getFixString((int) str_len, charsetName));
+                    .getFixString((int) str_len, charset));
             case JSONB_TYPE_OPAQUE:
                 /*
                  * There should always be at least one byte, which tells the
@@ -242,7 +248,7 @@ public class JsonConversion {
             this.m_large = large;
         }
 
-        public String key(int i, String charsetName) {
+        public String key(int i, Charset charset) {
             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;
@@ -255,10 +261,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, charsetName);
+            return m_data.rewind().forward((int) key_offset).getFixString((int) key_length, charset);
         }
 
-        public Json_Value element(int i, String charsetName) {
+        public Json_Value element(int i, Charset charset) {
             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;
@@ -271,13 +277,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, charsetName);
+                return parse_scalar(type, m_data, value_entry_size - 1, charset);
             }
             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, charsetName);
+            return parse_value(type, m_data.rewind().forward(value_offset), (int) m_length - value_offset, charset);
         }
 
-        public StringBuilder toJsonString(StringBuilder buf, String charsetName) {
+        public StringBuilder toJsonString(StringBuilder buf, Charset charset) {
             switch (m_type) {
                 case OBJECT:
                     buf.append("{");
@@ -285,9 +291,9 @@ public class JsonConversion {
                         if (i > 0) {
                             buf.append(", ");
                         }
-                        buf.append('"').append(key(i, charsetName)).append('"');
+                        buf.append('"').append(key(i, charset)).append('"');
                         buf.append(": ");
-                        element(i, charsetName).toJsonString(buf, charsetName);
+                        element(i, charset).toJsonString(buf, charset);
                     }
                     buf.append("}");
                     break;
@@ -297,7 +303,7 @@ public class JsonConversion {
                         if (i > 0) {
                             buf.append(", ");
                         }
-                        element(i, charsetName).toJsonString(buf, charsetName);
+                        element(i, charset).toJsonString(buf, charset);
                     }
                     buf.append("]");
                     break;
@@ -398,7 +404,7 @@ public class JsonConversion {
                         }
                         buf.append('"').append(text).append('"');
                     } else {
-                        text = m_data.getFixString((int) m_length, charsetName);
+                        text = m_data.getFixString((int) m_length, charset);
                         buf.append('"').append(escapse(text)).append('"');
                     }
 

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

@@ -1,5 +1,6 @@
 package com.taobao.tddl.dbsync.binlog;
 
+import java.nio.charset.Charset;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -36,6 +37,11 @@ public class JsonDiffConversion {
 
     public static StringBuilder print_json_diff(LogBuffer buffer, long len, String columnName, int columnIndex,
                                                 String charsetName) {
+        return print_json_diff(buffer, len, columnName, columnIndex, Charset.forName(charsetName));
+    }
+
+    public static StringBuilder print_json_diff(LogBuffer buffer, long len, String columnName, int columnIndex,
+                                                Charset charset) {
         int position = buffer.position();
         List<String> operation_names = new ArrayList<>();
         while (buffer.hasRemaining()) {
@@ -111,14 +117,14 @@ public class JsonDiffConversion {
                 Json_Value jsonValue = JsonConversion.parse_value(buffer.getUint8(),
                     buffer,
                     value_length - 1,
-                    charsetName);
+                    charset);
                 buffer.forward((int) value_length - 1);
                 // Read value
                 if (jsonValue.m_type == Json_enum_type.ERROR) {
                     throw new IllegalArgumentException("parsing json value");
                 }
                 StringBuilder jsonBuilder = new StringBuilder();
-                jsonValue.toJsonString(jsonBuilder, charsetName);
+                jsonValue.toJsonString(jsonBuilder, charset);
                 builder.append(jsonBuilder);
             }
 

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

@@ -5,6 +5,8 @@ import java.io.OutputStream;
 import java.io.UnsupportedEncodingException;
 import java.math.BigDecimal;
 import java.math.BigInteger;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
 import java.util.BitSet;
 
@@ -16,6 +18,8 @@ import java.util.BitSet;
  */
 public class LogBuffer {
 
+    static final BigDecimal DECIMAL_ZERO_1_SCALE = BigDecimal.valueOf(0, 1);
+    static final BigDecimal DECIMAL_ONE_1_SCALE = BigDecimal.valueOf(10, 1);
     protected byte[] buffer;
 
     protected int    origin, limit;
@@ -1038,29 +1042,27 @@ public class LogBuffer {
         }
     }
 
-    /* default ANSI charset */
-    public static final String ISO_8859_1 = "ISO-8859-1";
-
     /**
      * Return fix length string from buffer.
      */
     public final String getFixString(final int pos, final int len) {
-        return getFixString(pos, len, ISO_8859_1);
+        return getFixString(pos, len, StandardCharsets.ISO_8859_1);
     }
 
     /**
      * Return next fix length string from buffer.
      */
     public final String getFixString(final int len) {
-        return getFixString(len, ISO_8859_1);
+        return getFixString(len, StandardCharsets.ISO_8859_1);
     }
 
     /**
      * Return fix length string from buffer.
      */
-    public final String getFixString(final int pos, final int len, String charsetName) {
-        if (pos + len > limit || pos < 0) throw new IllegalArgumentException("limit excceed: "
-                                                                             + (pos < 0 ? pos : (pos + len)));
+    public final String getFixString(final int pos, final int len, Charset charset) {
+        if (pos + len > limit || pos < 0) {
+            throw new IllegalArgumentException("limit excceed: " + (pos < 0 ? pos : (pos + len)));
+        }
 
         final int from = origin + pos;
         final int end = from + len;
@@ -1069,19 +1071,16 @@ public class LogBuffer {
         for (; (found < end) && buf[found] != '\0'; found++)
             /* empty loop */;
 
-        try {
-            return new String(buf, from, found - from, charsetName);
-        } catch (UnsupportedEncodingException e) {
-            throw new IllegalArgumentException("Unsupported encoding: " + charsetName, e);
-        }
+        return new String(buf, from, found - from, charset);
     }
 
     /**
      * Return next fix length string from buffer.
      */
-    public final String getFixString(final int len, String charsetName) {
-        if (position + len > origin + limit) throw new IllegalArgumentException("limit excceed: "
-                                                                                + (position + len - origin));
+    public final String getFixString(final int len, Charset charset) {
+        if (position + len > origin + limit) {
+            throw new IllegalArgumentException("limit excceed: " + (position + len - origin));
+        }
 
         final int from = position;
         final int end = from + len;
@@ -1090,97 +1089,141 @@ public class LogBuffer {
         for (; (found < end) && buf[found] != '\0'; found++)
             /* empty loop */;
 
-        try {
-            String string = new String(buf, from, found - from, charsetName);
-            position += len;
-            return string;
-        } catch (UnsupportedEncodingException e) {
-            throw new IllegalArgumentException("Unsupported encoding: " + charsetName, e);
+        String string = new String(buf, from, found - from, charset);
+        position += len;
+        return string;
+    }
+
+    public final String getFixName(final int len, Charset charset) {
+        if (position + len > origin + limit) {
+            throw new IllegalArgumentException("limit excceed: " + (position + len - origin));
         }
+
+        final int from = position;
+        final int end = from + len;
+        byte[] buf = buffer;
+        int found = from;
+        for (; (found < end) && buf[found] != '\0'; found++)
+            /* empty loop */;
+
+        int length = found - from;
+        String string = null;
+        if (length <= 16) {
+            string = NameCache.name(buf, from, length, charset);
+        }
+        if (string == null) {
+            string = new String(buf, from, length, charset);
+        }
+        position += len;
+        return string;
+    }
+
+    public final String getFixName(final int len) {
+        if (position + len > origin + limit) {
+            throw new IllegalArgumentException("limit excceed: " + (position + len - origin));
+        }
+
+        final int from = position;
+        final int end = from + len;
+        byte[] buf = buffer;
+        int found = from;
+        for (; (found < end) && buf[found] != '\0'; found++)
+            /* empty loop */;
+
+        int length = found - from;
+        String string = null;
+        if (length <= 16) {
+            string = NameCache.name(buf, from, length, StandardCharsets.ISO_8859_1);
+        }
+        if (string == null) {
+            string = new String(buf, from, length, StandardCharsets.ISO_8859_1);
+        }
+        position += len;
+        return string;
     }
 
     /**
-     * Return fix-length string from buffer without null-terminate checking. Fix
-     * bug #17 {@link https://github.com/AlibabaTech/canal/issues/17 }
+     * Return fix-length string from buffer without null-terminate checking.
+     * Fix bug #17 {@link https://github.com/AlibabaTech/canal/issues/17 }
      */
-    public final String getFullString(final int pos, final int len, String charsetName) {
-        if (pos + len > limit || pos < 0) throw new IllegalArgumentException("limit excceed: "
-                                                                             + (pos < 0 ? pos : (pos + len)));
-
-        try {
-            return new String(buffer, origin + pos, len, charsetName);
-        } catch (UnsupportedEncodingException e) {
-            throw new IllegalArgumentException("Unsupported encoding: " + charsetName, e);
+    public final String getFullString(final int pos, final int len, Charset charset) {
+        if (pos + len > limit || pos < 0) {
+            throw new IllegalArgumentException("limit excceed: " + (pos < 0 ? pos : (pos + len)));
         }
+
+        return new String(buffer, origin + pos, len, charset);
     }
 
     /**
      * Return next fix-length string from buffer without null-terminate
-     * checking. Fix bug #17 {@link https
-     * ://github.com/AlibabaTech/canal/issues/17 }
+     * checking.
+     * Fix bug #17 {@link https://github.com/AlibabaTech/canal/issues/17 }
      */
-    public final String getFullString(final int len, String charsetName) {
-        if (position + len > origin + limit) throw new IllegalArgumentException("limit excceed: "
-                                                                                + (position + len - origin));
-
-        try {
-            String string = new String(buffer, position, len, charsetName);
-            position += len;
-            return string;
-        } catch (UnsupportedEncodingException e) {
-            throw new IllegalArgumentException("Unsupported encoding: " + charsetName, e);
+    public final String getFullString(final int len, Charset charset) {
+        if (position + len > origin + limit) {
+            throw new IllegalArgumentException("limit excceed: " + (position + len - origin));
         }
+
+        String string = new String(buffer, position, len, charset);
+        position += len;
+        return string;
     }
 
+
     /**
      * Return dynamic length string from buffer.
      */
     public final String getString(final int pos) {
-        return getString(pos, ISO_8859_1);
+        if (pos >= limit || pos < 0) {
+            throw new IllegalArgumentException("limit excceed: " + pos);
+        }
+
+        byte[] buf = buffer;
+        final int len = (0xff & buf[origin + pos]);
+        if (pos + len + 1 > limit) {
+            throw new IllegalArgumentException("limit excceed: " + (pos + len + 1));
+        }
+
+        return new String(buf, origin + pos + 1, len, StandardCharsets.ISO_8859_1);
     }
 
     /**
      * Return next dynamic length string from buffer.
      */
     public final String getString() {
-        return getString(ISO_8859_1);
-    }
-
-    /**
-     * Return dynamic length string from buffer.
-     */
-    public final String getString(final int pos, String charsetName) {
-        if (pos >= limit || pos < 0) throw new IllegalArgumentException("limit excceed: " + pos);
-
-        byte[] buf = buffer;
-        final int len = (0xff & buf[origin + pos]);
-        if (pos + len + 1 > limit) throw new IllegalArgumentException("limit excceed: " + (pos + len + 1));
-
-        try {
-            return new String(buf, origin + pos + 1, len, charsetName);
-        } catch (UnsupportedEncodingException e) {
-            throw new IllegalArgumentException("Unsupported encoding: " + charsetName, e);
+        if (position >= origin + limit) {
+            throw new IllegalArgumentException("limit excceed: " + position);
         }
-    }
-
-    /**
-     * Return next dynamic length string from buffer.
-     */
-    public final String getString(String charsetName) {
-        if (position >= origin + limit) throw new IllegalArgumentException("limit excceed: " + position);
 
         byte[] buf = buffer;
         final int len = (0xff & buf[position]);
-        if (position + len + 1 > origin + limit) throw new IllegalArgumentException("limit excceed: "
-                                                                                    + (position + len + 1 - origin));
-
-        try {
-            String string = new String(buf, position + 1, len, charsetName);
-            position += len + 1;
-            return string;
-        } catch (UnsupportedEncodingException e) {
-            throw new IllegalArgumentException("Unsupported encoding: " + charsetName, e);
+        if (position + len + 1 > origin + limit) {
+            throw new IllegalArgumentException("limit excceed: "
+                                               + (position + len + 1 - origin));
         }
+
+        String string = new String(buf, position + 1, len, StandardCharsets.ISO_8859_1);
+        position += len + 1;
+        return string;
+    }
+
+    public final String getName() {
+        if (position >= origin + limit) {
+            throw new IllegalArgumentException("limit excceed: " + position);
+        }
+
+        byte[] buf = buffer;
+        final int len = (0xff & buf[position]);
+        if (position + len + 1 > origin + limit) {
+            throw new IllegalArgumentException("limit excceed: " + (position + len + 1 - origin));
+        }
+
+        String string = NameCache.name(buf, position + 1, len, StandardCharsets.ISO_8859_1);
+        if (string == null) {
+            string = new String(buf, position + 1, len, StandardCharsets.ISO_8859_1);
+        }
+        position += len + 1;
+        return string;
     }
 
     /**
@@ -1263,6 +1306,15 @@ public class LogBuffer {
             throw new IllegalArgumentException("limit excceed: " + (position + binSize - origin));
         }
 
+        if (precision <= 20 && intg <= 18 && frac <= 9) {
+            // use compact BigDecimal
+            BigDecimal compactDecimal = getBigDecimalCompact(precision, scale, intg, frac, intg0x);
+            if (compactDecimal != null) {
+                position += binSize;
+                return compactDecimal;
+            }
+        }
+
         BigDecimal decimal = getDecimal0(position, intg, frac, // NL
             intg0,
             frac0,
@@ -1272,6 +1324,162 @@ public class LogBuffer {
         return decimal;
     }
 
+    /**
+     * use compact BigDecimal
+     */
+    private BigDecimal getBigDecimalCompact(int precision, int scale, int intg, int frac, int intg0x) {
+        if (precision > 20 || intg > 18 && frac > 9) {
+            return null;
+        }
+
+        BigDecimal compactDecimal = null;
+        long longValue = 0;
+        int small = 0;
+        int pos = position;
+        byte b0 = buffer[position];
+        final int mask = ((b0 & 0x80) == 0x80) ? 0 : -1;
+        b0 ^= 0x80;
+
+        switch (intg0x) {
+            case 0:
+                if (intg == 0) {
+                    longValue = 0;
+                } else {
+                    longValue = (b0 << 24)
+                                | ((0xff & buffer[pos + 1]) << 16)
+                                | ((0xff & buffer[pos + 2]) << 8)
+                                | (0xff & buffer[pos + 3]);
+                    longValue ^= mask;
+                    pos += 4;
+                }
+                break;
+            case 1:
+            case 2:
+                longValue = b0;
+                longValue ^= mask;
+                pos += 1;
+                break;
+            case 3:
+            case 4:
+                longValue = (b0 << 8) | (0xff & buffer[pos + 1]);
+                longValue ^= mask;
+                pos += 2;
+                break;
+            case 5:
+            case 6:
+                longValue = (b0 << 16) | ((0xff & buffer[pos + 1]) << 8) | (0xff & buffer[pos + 2]);
+                longValue ^= mask;
+                pos += 3;
+                break;
+            case 7:
+            case 8:
+                longValue = (b0 << 24)
+                            | ((0xff & buffer[pos + 1]) << 16)
+                            | ((0xff & buffer[pos + 2]) << 8)
+                            | (0xff & buffer[pos + 3]);
+                longValue ^= mask;
+                pos += 4;
+                break;
+            default:
+                break;
+        }
+
+        if (intg > 9) {
+            int value = getInt32BE(buffer, pos);
+            value ^= mask;
+            longValue *= 1_000_000_000;
+            longValue += value;
+            pos += 4;
+        }
+
+        int power = 1;
+        if (pos != position) {
+            b0 = buffer[pos];
+        }
+        switch (frac) {
+            case 0:
+                small = 0;
+                break;
+            case 1:
+            case 2:
+                power = frac == 1 ? 10 : 100;
+                small = b0;
+                small ^= mask;
+                break;
+            case 3:
+            case 4:
+                power = frac == 3 ? 1_000 : 10_000;
+                small = (b0 << 8) | (0xff & buffer[pos + 1]);
+                small ^= mask;
+                break;
+            case 5:
+            case 6:
+                power = frac == 5 ? 100_000 : 1_000_000;
+                small = (b0 << 16) | ((0xff & buffer[pos + 1]) << 8) | (0xff & buffer[pos + 2]);
+                small ^= mask;
+                break;
+            case 7:
+            case 8:
+            case 9:
+                power = frac == 7 ? 10_000_000 : frac == 8 ? 100_000_000 : 1_000_000_000;
+                small = (b0 << 24)
+                        | ((0xff & buffer[pos + 1]) << 16)
+                        | ((0xff & buffer[pos + 2]) << 8)
+                        | (0xff & buffer[pos + 3]);
+                small ^= mask;
+                break;
+            default:
+                break;
+        }
+
+        if (small == 0) {
+            if (frac == 0) {
+                if (mask != 0) {
+                    longValue = -longValue;
+                }
+                compactDecimal = BigDecimal.valueOf(longValue, 0);
+            } else if (longValue == 0) {
+                if (scale < 9) {
+                    compactDecimal = DECIMAL_ZERO_1_SCALE;
+                } else {
+                    compactDecimal = BigDecimal.valueOf(longValue, scale);
+                }
+            } else {
+                if (scale < 9) {
+                    long unscaleValue = longValue * 10;
+                    if (unscaleValue >= longValue) {
+                        if (mask != 0) {
+                            unscaleValue = -unscaleValue;
+                        }
+                        if (unscaleValue == 10) {
+                            compactDecimal = DECIMAL_ONE_1_SCALE;
+                        } else {
+                            compactDecimal = BigDecimal.valueOf(unscaleValue, 1);
+                        }
+                    }
+                } else {
+                    long unscaleValue = longValue * power;
+                    if (unscaleValue >= longValue) {
+                        if (mask != 0) {
+                            unscaleValue = -unscaleValue;
+                        }
+                        compactDecimal = BigDecimal.valueOf(unscaleValue, scale);
+                    }
+                }
+            }
+        } else {
+            long unscaleValue = longValue * power + small;
+            if (unscaleValue >= longValue) {
+                if (mask != 0) {
+                    unscaleValue = -unscaleValue;
+                }
+                compactDecimal = BigDecimal.valueOf(unscaleValue, scale);
+            }
+        }
+
+        return compactDecimal;
+    }
+
     /**
      * Return big decimal from buffer.
      * 
@@ -1445,8 +1653,9 @@ public class LogBuffer {
         }
 
         d_copy[begin] ^= 0x80; /* restore sign */
-        String decimal = String.valueOf(buf, 0, pos);
-        return new BigDecimal(decimal);
+        // String decimal = String.valueOf(buf, 0, pos);
+        // return new BigDecimal(decimal);
+        return new BigDecimal(buf, 0, pos);
     }
 
     /**

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

@@ -18,6 +18,7 @@ public class GtidLogEvent extends LogEvent {
     // / Length of SID in event encoding
     public static final int ENCODED_SID_LENGTH          = 16;
     public static final int LOGICAL_TIMESTAMP_TYPE_CODE = 2;
+    public static final UUID UUID_ZERO                   = new UUID(0, 0);
 
     private boolean         commitFlag;
     private UUID            sid;
@@ -35,11 +36,13 @@ public class GtidLogEvent extends LogEvent {
         buffer.position(commonHeaderLen);
         commitFlag = (buffer.getUint8() != 0); // ENCODED_FLAG_LENGTH
 
-        byte[] bs = buffer.getData(ENCODED_SID_LENGTH);
-        ByteBuffer bb = ByteBuffer.wrap(bs);
-        long high = bb.getLong();
-        long low = bb.getLong();
-        sid = new UUID(high, low);
+        long high = buffer.getBeLong64();
+        long low = buffer.getBeLong64();
+        if (high == 0 && low == 0) {
+            sid = UUID_ZERO;
+        } else {
+            sid = new UUID(high, low);
+        }
 
         gno = buffer.getLong64();
 
@@ -73,11 +76,11 @@ public class GtidLogEvent extends LogEvent {
         return gno;
     }
 
-    public Long getLastCommitted() {
+    public long getLastCommitted() {
         return lastCommitted;
     }
 
-    public Long getSequenceNumber() {
+    public long getSequenceNumber() {
         return sequenceNumber;
     }
 

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

@@ -3,6 +3,8 @@ package com.taobao.tddl.dbsync.binlog.event;
 import com.taobao.tddl.dbsync.binlog.LogBuffer;
 import com.taobao.tddl.dbsync.binlog.LogEvent;
 
+import java.nio.charset.StandardCharsets;
+
 /**
  * <pre>
  * Replication event to ensure to slave that master is alive.
@@ -35,7 +37,7 @@ public class HeartbeatLogEvent extends LogEvent {
             identLen = FN_REFLEN - 1;
         }
 
-        logIdent = buffer.getFullString(commonHeaderLen, identLen, LogBuffer.ISO_8859_1);
+        logIdent = buffer.getFullString(commonHeaderLen, identLen, StandardCharsets.ISO_8859_1);
     }
 
     public int getIdentLen() {

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

@@ -67,7 +67,7 @@ public final class IncidentLogEvent extends LogEvent {
         incident = incidentNumber;
 
         buffer.position(commonHeaderLen + postHeaderLen);
-        message = buffer.getString();
+        message = buffer.getName();
     }
 
     public final int getIncident() {

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

@@ -433,7 +433,7 @@ public class QueryLogEvent extends LogEvent {
     private int             serverCollation           = -1;
     private int             tvSec                     = -1;
     private BigInteger      ddlXid                    = BigInteger.valueOf(-1L);
-    private String          charsetName;
+    private Charset         charset;
 
     private String          timezone;
 
@@ -498,12 +498,12 @@ public class QueryLogEvent extends LogEvent {
 
         /* A 2nd variable part; this is common to all versions */
         final int queryLen = dataLen - dbLen - 1;
-        dbname = buffer.getFixString(dbLen + 1);
+        dbname = buffer.getFixName(dbLen + 1);
         if (clientCharset >= 0) {
-            charsetName = CharsetConversion.getJavaCharset(clientCharset);
+            charset = CharsetConversion.getNioCharset(clientCharset);
 
-            if ((charsetName != null) && (Charset.isSupported(charsetName))) {
-                query = buffer.getFixString(queryLen, charsetName);
+            if (charset != null) {
+                query = buffer.getFixString(queryLen, charset);
             } else {
                 logger.warn("unsupported character set in query log: " + "\n    ID = " + clientCharset + ", Charset = "
                             + CharsetConversion.getCharset(clientCharset) + ", Collation = "
@@ -623,7 +623,7 @@ public class QueryLogEvent extends LogEvent {
                         // is ulonglong
                         break;
                     case Q_CATALOG_NZ_CODE:
-                        catalog = buffer.getString();
+                        catalog = buffer.getName();
                         break;
                     case Q_AUTO_INCREMENT:
                         autoIncrementIncrement = buffer.getUint16();
@@ -639,7 +639,7 @@ public class QueryLogEvent extends LogEvent {
                         serverCollation = buffer.getUint16();
                         break;
                     case Q_TIME_ZONE_CODE:
-                        timezone = buffer.getString();
+                        timezone = buffer.getName();
                         break;
                     case Q_CATALOG_CODE: /* for 5.0.x where 0<=x<=3 masters */
                         final int len = buffer.getUint8();
@@ -663,8 +663,8 @@ public class QueryLogEvent extends LogEvent {
                         buffer.forward(4);
                         break;
                     case Q_INVOKER:
-                        user = buffer.getString();
-                        host = buffer.getString();
+                        user = buffer.getName();
+                        host = buffer.getName();
                         break;
                     case Q_MICROSECONDS:
                         // when.tv_usec= uint3korr(pos);
@@ -685,7 +685,7 @@ public class QueryLogEvent extends LogEvent {
                         String mtsAccessedDbNames[] = new String[mtsAccessedDbs];
                         for (int i = 0; i < mtsAccessedDbs && buffer.position() < end; i++) {
                             int length = end - buffer.position();
-                            mtsAccessedDbNames[i] = buffer.getFixString(length < NAME_LEN ? length : NAME_LEN);
+                            mtsAccessedDbNames[i] = buffer.getFixName(length < NAME_LEN ? length : NAME_LEN);
                         }
                         break;
                     case Q_EXPLICIT_DEFAULTS_FOR_TIMESTAMP:
@@ -829,8 +829,8 @@ public class QueryLogEvent extends LogEvent {
         return autoIncrementOffset;
     }
 
-    public final String getCharsetName() {
-        return charsetName;
+    public final Charset getCharset() {
+        return charset;
     }
 
     public final String getTimezone() {

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

@@ -1,8 +1,11 @@
 package com.taobao.tddl.dbsync.binlog.event;
 
 import java.io.Serializable;
+import java.nio.charset.Charset;
 import java.sql.Timestamp;
 import java.sql.Types;
+import java.time.ZoneId;
+import java.time.zone.ZoneRules;
 import java.util.BitSet;
 
 import org.apache.commons.logging.Log;
@@ -23,7 +26,11 @@ import com.taobao.tddl.dbsync.binlog.LogEvent;
  */
 public final class RowsLogBuffer {
 
-    protected static final Log logger            = LogFactory.getLog(RowsLogBuffer.class);
+    protected static final Log    logger            = LogFactory.getLog(RowsLogBuffer.class);
+    public static final Long[]    longCache         = new Long[1024 * 128];
+    public static final int       longCacheLimit    = longCache.length + 127;
+    public static final Integer[] integerCache      = new Integer[1024 * 128];
+    public static final int       integerCacheLimit = longCache.length + 127;
 
     public static final long   DATETIMEF_INT_OFS = 0x8000000000L;
     public static final long   TIMEF_INT_OFS     = 0x800000L;
@@ -33,7 +40,7 @@ public final class RowsLogBuffer {
     private final LogBuffer    buffer;
     private final int          columnLen;
     private final int          jsonColumnCount;
-    private final String       charsetName;
+    private final Charset      charset;
 
     private final BitSet       nullBits;
     private int                nullBitIndex;
@@ -47,10 +54,10 @@ public final class RowsLogBuffer {
     private int                length;
     private Serializable       value;
 
-    public RowsLogBuffer(LogBuffer buffer, final int columnLen, String charsetName, int jsonColumnCount, boolean partial){
+    public RowsLogBuffer(LogBuffer buffer, final int columnLen, Charset charset, int jsonColumnCount, boolean partial){
         this.buffer = buffer;
         this.columnLen = columnLen;
-        this.charsetName = charsetName;
+        this.charset = charset;
         this.partial = partial;
         this.jsonColumnCount = jsonColumnCount;
         this.nullBits = new BitSet(columnLen);
@@ -309,7 +316,7 @@ public final class RowsLogBuffer {
                 // XXX: How to check signed / unsigned?
                 // value = unsigned ? Long.valueOf(buffer.getUint32()) :
                 // Integer.valueOf(buffer.getInt32());
-                value = Integer.valueOf(buffer.getInt32());
+                value = valueOf(buffer.getInt32());
                 javaType = Types.INTEGER;
                 length = 4;
                 break;
@@ -318,7 +325,7 @@ public final class RowsLogBuffer {
                 // XXX: How to check signed / unsigned?
                 // value = Integer.valueOf(unsigned ? buffer.getUint8() :
                 // buffer.getInt8());
-                value = Integer.valueOf(buffer.getInt8());
+                value = valueOf(buffer.getInt8());
                 javaType = Types.TINYINT; // java.sql.Types.INTEGER;
                 length = 1;
                 break;
@@ -327,7 +334,7 @@ public final class RowsLogBuffer {
                 // XXX: How to check signed / unsigned?
                 // value = Integer.valueOf(unsigned ? buffer.getUint16() :
                 // buffer.getInt16());
-                value = Integer.valueOf((short) buffer.getInt16());
+                value = valueOf((short) buffer.getInt16());
                 javaType = Types.SMALLINT; // java.sql.Types.INTEGER;
                 length = 2;
                 break;
@@ -336,7 +343,7 @@ public final class RowsLogBuffer {
                 // XXX: How to check signed / unsigned?
                 // value = Integer.valueOf(unsigned ? buffer.getUint24() :
                 // buffer.getInt24());
-                value = Integer.valueOf(buffer.getInt24());
+                value = valueOf(buffer.getInt24());
                 javaType = Types.INTEGER;
                 length = 3;
                 break;
@@ -345,7 +352,7 @@ public final class RowsLogBuffer {
                 // XXX: How to check signed / unsigned?
                 // value = unsigned ? buffer.getUlong64()) :
                 // Long.valueOf(buffer.getLong64());
-                value = Long.valueOf(buffer.getLong64());
+                value = valueOf(buffer.getLong64());
                 javaType = Types.BIGINT; // Types.INTEGER;
                 length = 8;
                 break;
@@ -870,7 +877,7 @@ public final class RowsLogBuffer {
                 }
                 // logger.warn("MYSQL_TYPE_ENUM : This enumeration value is "
                 // + "only used internally and cannot exist in a binlog!");
-                value = Integer.valueOf(int32);
+                value = valueOf(int32);
                 javaType = Types.INTEGER;
                 length = len;
                 break;
@@ -1017,7 +1024,7 @@ public final class RowsLogBuffer {
                     javaType = Types.VARBINARY;
                     value = binary;
                 } else {
-                    value = buffer.getFullString(len, charsetName);
+                    value = buffer.getFullString(len, charset);
                     javaType = Types.VARCHAR;
                 }
 
@@ -1039,7 +1046,7 @@ public final class RowsLogBuffer {
                     javaType = Types.BINARY;
                     value = binary;
                 } else {
-                    value = buffer.getFullString(len, charsetName);
+                    value = buffer.getFullString(len, charset);
                     javaType = Types.CHAR; // Types.VARCHAR;
                 }
                 length = len;
@@ -1074,7 +1081,7 @@ public final class RowsLogBuffer {
                         len,
                         columnName,
                         columnIndex,
-                        charsetName);
+                            charset);
                     value = builder.toString();
                     buffer.position(position + len);
                 } else {
@@ -1088,9 +1095,9 @@ public final class RowsLogBuffer {
                         Json_Value jsonValue = JsonConversion.parse_value(buffer.getUint8(),
                             buffer,
                             len - 1,
-                            charsetName);
+                                charset);
                         StringBuilder builder = new StringBuilder();
-                        jsonValue.toJsonString(builder, charsetName);
+                        jsonValue.toJsonString(builder, charset);
                         value = builder.toString();
                         buffer.position(position + len);
                     }
@@ -1211,4 +1218,38 @@ public final class RowsLogBuffer {
             builder.append('0').append(digits[d]);
         }
     }
+
+    public static Long valueOf(long value) {
+        if (value >= -128 & value <= 127) {
+            // if (l >= -128 && l <= 127) {
+            return Long.valueOf(value);
+        }
+        if (value > 127 && value < longCacheLimit) {
+            int cacheIndex = (int) value - 127;
+            Long cacheValue = longCache[cacheIndex];
+            if (cacheValue == null) {
+                cacheValue = new Long(value);
+                longCache[cacheIndex] = cacheValue;
+            }
+            return cacheValue;
+        }
+        return new Long(value);
+    }
+
+    public static Integer valueOf(int value) {
+        if (value >= -128 & value <= 127) {
+            // if (l >= -128 && l <= 127) {
+            return Integer.valueOf(value);
+        }
+        if (value > 127 && value < integerCacheLimit) {
+            int cacheIndex = (int) value - 127;
+            Integer cacheValue = integerCache[cacheIndex];
+            if (cacheValue == null) {
+                cacheValue = new Integer(value);
+                integerCache[cacheIndex] = cacheValue;
+            }
+            return cacheValue;
+        }
+        return new Integer(value);
+    }
 }

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

@@ -1,5 +1,6 @@
 package com.taobao.tddl.dbsync.binlog.event;
 
+import java.nio.charset.Charset;
 import java.util.BitSet;
 
 import com.taobao.tddl.dbsync.binlog.exception.TableIdNotFoundException;
@@ -219,8 +220,8 @@ public abstract class RowsLogEvent extends LogEvent {
         return changeColumns;
     }
 
-    public final RowsLogBuffer getRowsBuf(String charsetName) {
-        return new RowsLogBuffer(rowsBuf, columnLen, charsetName, jsonColumnCount, partial);
+    public final RowsLogBuffer getRowsBuf(Charset charset) {
+        return new RowsLogBuffer(rowsBuf, columnLen, charset, jsonColumnCount, partial);
     }
 
     public final int getFlags(final int flags) {

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

@@ -2,6 +2,8 @@ package com.taobao.tddl.dbsync.binlog.event;
 
 import com.taobao.tddl.dbsync.binlog.LogBuffer;
 
+import java.nio.charset.StandardCharsets;
+
 /**
  * @author jianghang 2013-4-8 上午12:36:29
  * @version 1.0.3
@@ -23,7 +25,7 @@ public class RowsQueryLogEvent extends IgnorableLogEvent {
          */
         int offset = commonHeaderLen + postHeaderLen + 1;
         int len = buffer.limit() - offset;
-        rowsQuery = buffer.getFullString(offset, len, LogBuffer.ISO_8859_1);
+        rowsQuery = buffer.getFullString(offset, len, StandardCharsets.ISO_8859_1);
     }
 
     public String getRowsQuery() {

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

@@ -36,7 +36,7 @@ public class StartLogEventV3 extends LogEvent {
 
         buffer.position(descriptionEvent.commonHeaderLen);
         binlogVersion = buffer.getUint16(); // ST_BINLOG_VER_OFFSET
-        serverVersion = buffer.getFixString(ST_SERVER_VER_LEN); // ST_SERVER_VER_OFFSET
+        serverVersion = buffer.getFixName(ST_SERVER_VER_LEN); // ST_SERVER_VER_OFFSET
     }
 
     public StartLogEventV3(){

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

@@ -423,10 +423,10 @@ public final class TableMapLogEvent extends LogEvent {
 
         /* Read the variable part of the event */
         buffer.position(commonHeaderLen + postHeaderLen);
-        dbname = buffer.getString();
+        dbname = buffer.getName();
         buffer.forward(1); /* termination null */
         // fixed issue #2714
-        tblname = buffer.getString();
+        tblname = buffer.getName();
         buffer.forward(1); /* termination null */
 
         // Read column information from buffer

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

@@ -2,6 +2,7 @@ package com.taobao.tddl.dbsync.binlog.event;
 
 import java.io.IOException;
 import java.io.Serializable;
+import java.nio.charset.Charset;
 
 import com.taobao.tddl.dbsync.binlog.CharsetConversion;
 import com.taobao.tddl.dbsync.binlog.LogBuffer;
@@ -100,8 +101,8 @@ public final class UserVarLogEvent extends LogEvent {
                     value = buffer.getDecimal(precision, scale); // bin2decimal
                     break;
                 case STRING_RESULT:
-                    String charsetName = CharsetConversion.getJavaCharset(charsetNumber);
-                    value = buffer.getFixString(valueLen, charsetName);
+                    Charset charset = CharsetConversion.getNioCharset(charsetNumber);
+                    value = buffer.getFixString(valueLen, charset);
                     break;
                 case ROW_RESULT:
                     // this seems to be banned in MySQL altogether

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

@@ -5,6 +5,8 @@ import com.taobao.tddl.dbsync.binlog.event.FormatDescriptionLogEvent;
 import com.taobao.tddl.dbsync.binlog.event.IgnorableLogEvent;
 import com.taobao.tddl.dbsync.binlog.event.LogHeader;
 
+import java.nio.charset.StandardCharsets;
+
 /**
  * mariadb的ANNOTATE_ROWS_EVENT类型
  * 
@@ -23,7 +25,7 @@ public class AnnotateRowsEvent extends IgnorableLogEvent {
 
         int offset = commonHeaderLen + postHeaderLen;
         int len = buffer.limit() - offset;
-        rowsQuery = buffer.getFullString(offset, len, LogBuffer.ISO_8859_1);
+        rowsQuery = buffer.getFullString(offset, len, StandardCharsets.ISO_8859_1);
     }
 
     public String getRowsQuery() {

dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/BaseLogFetcherTest.java

@@ -52,7 +52,7 @@ public class BaseLogFetcherTest {
                 event.getHeader().getLogPos() - event.getHeader().getEventLen(),
                 event.getTable().getDbName(),
                 event.getTable().getTableName()));
-            RowsLogBuffer buffer = event.getRowsBuf(charset.name());
+            RowsLogBuffer buffer = event.getRowsBuf(charset);
             BitSet columns = event.getColumns();
             BitSet changeColumns = event.getChangeColumns();
             while (buffer.nextOneRow(columns)) {

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

@@ -558,6 +558,11 @@ public class CanalController {
         }
 
         ZkClientx.clearClients();
+
+        // 需要释放 CanalServerWithEmbedded 否则主线程退出后，进程无法自动完整退出...
+        if (embededCanalServer != null) {
+            embededCanalServer.stop();
+        }
     }
 
     private void initCid(String path) {

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

@@ -1,5 +1,6 @@
 package com.alibaba.otter.canal.deployer;
 
+import com.alibaba.otter.canal.connector.core.spi.ProxyCanalMQProducer;
 import java.util.Properties;
 
 import com.alibaba.otter.canal.connector.core.config.MQProperties;
@@ -21,21 +22,21 @@ import com.alibaba.otter.canal.server.CanalMQStarter;
  */
 public class CanalStarter {
 
-    private static final Logger logger                    = LoggerFactory.getLogger(CanalStarter.class);
+    private static final Logger logger = LoggerFactory.getLogger(CanalStarter.class);
 
-    private static final String CONNECTOR_SPI_DIR         = "/plugin";
+    private static final String CONNECTOR_SPI_DIR = "/plugin";
     private static final String CONNECTOR_STANDBY_SPI_DIR = "/canal/plugin";
 
-    private CanalController     controller                = null;
-    private CanalMQProducer     canalMQProducer           = null;
-    private Thread              shutdownThread            = null;
-    private CanalMQStarter      canalMQStarter            = null;
+    private CanalController controller = null;
+    private CanalMQProducer canalMQProducer = null;
+    private Thread shutdownThread = null;
+    private CanalMQStarter canalMQStarter = null;
     private volatile Properties properties;
-    private volatile boolean    running                   = false;
+    private volatile boolean running = false;
 
     private CanalAdminWithNetty canalAdmin;
 
-    public CanalStarter(Properties properties){
+    public CanalStarter(Properties properties) {
         this.properties = properties;
     }
 
@@ -67,10 +68,8 @@ public class CanalStarter {
             canalMQProducer = loader
                 .getExtension(serverMode.toLowerCase(), CONNECTOR_SPI_DIR, CONNECTOR_STANDBY_SPI_DIR);
             if (canalMQProducer != null) {
-                ClassLoader cl = Thread.currentThread().getContextClassLoader();
-                Thread.currentThread().setContextClassLoader(canalMQProducer.getClass().getClassLoader());
+                canalMQProducer =  new ProxyCanalMQProducer(canalMQProducer);
                 canalMQProducer.init(properties);
-                Thread.currentThread().setContextClassLoader(cl);
             }
         }
 

filter/src/main/java/com/alibaba/otter/canal/filter/aviater/AviaterRegexFilter.java

@@ -109,11 +109,11 @@ public class AviaterRegexFilter implements CanalEventFilter<String> {
     private List<String> completionPattern(List<String> patterns) {
         List<String> result = new ArrayList<>();
         for (String pattern : patterns) {
-            StringBuffer stringBuffer = new StringBuffer();
-            stringBuffer.append("^");
-            stringBuffer.append(pattern);
-            stringBuffer.append("$");
-            result.add(stringBuffer.toString());
+            StringBuilder stringBuilder = new StringBuilder();
+            stringBuilder.append("^");
+            stringBuilder.append(pattern);
+            stringBuilder.append("$");
+            result.add(stringBuilder.toString());
         }
         return result;
     }

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

@@ -171,7 +171,7 @@ public class FileMixedMetaManager extends MemoryMetaManager implements CanalMeta
                 return null;
             }
 
-            String json = FileUtils.readFileToString(dataFile, charset.name());
+            String json = FileUtils.readFileToString(dataFile, charset);
             return JsonUtils.unmarshalFromString(json, FileMetaInstanceData.class);
         } catch (IOException e) {
             throw new CanalMetaManagerException(e);
@@ -209,7 +209,11 @@ public class FileMixedMetaManager extends MemoryMetaManager implements CanalMeta
 
                 data.setClientDatas(clientDatas);
             }
-
+            //fixed issue https://github.com/alibaba/canal/issues/4312
+            //客户端数据为空时不覆盖文件内容 （适合单客户端）
+            if(data.getClientDatas().isEmpty()){
+                return;
+            }
             String json = JsonUtils.marshalToString(data);
             try {
                 FileUtils.writeStringToFile(dataFile, json);

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

@@ -175,7 +175,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
         builder.setKey("gtid");
         builder.setValue(logEvent.getGtidStr());
 
-        if (logEvent.getLastCommitted() != null) {
+        if (logEvent.getLastCommitted() != -1) {
             builder.setKey("lastCommitted");
             builder.setValue(String.valueOf(logEvent.getLastCommitted()));
             builder.setKey("sequenceNumber");
@@ -392,7 +392,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
         // mysql5.6支持，需要设置binlog-rows-query-log-events=1，可详细打印原始DML语句
         String queryString = null;
         try {
-            queryString = new String(event.getRowsQuery().getBytes(ISO_8859_1), charset.name());
+            queryString = new String(event.getRowsQuery().getBytes(ISO_8859_1), charset);
             String tableName = null;
             if (useDruidDdlFilter) {
                 List<DdlResult> results = DruidDdlParser.parse(queryString, null);
@@ -414,7 +414,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
         // mariaDb支持，需要设置binlog_annotate_row_events=true，可详细打印原始DML语句
         String queryString = null;
         try {
-            queryString = new String(event.getRowsQuery().getBytes(ISO_8859_1), charset.name());
+            queryString = new String(event.getRowsQuery().getBytes(ISO_8859_1), charset);
             return buildQueryEntry(queryString, event.getHeader());
         } catch (UnsupportedEncodingException e) {
             throw new CanalParseException(e);
@@ -508,10 +508,10 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
 
     public void parseTableMapEvent(TableMapLogEvent event) {
         try {
-            String charsetDbName = new String(event.getDbName().getBytes(ISO_8859_1), charset.name());
+            String charsetDbName = new String(event.getDbName().getBytes(ISO_8859_1), charset);
             event.setDbname(charsetDbName);
 
-            String charsetTbName = new String(event.getTableName().getBytes(ISO_8859_1), charset.name());
+            String charsetTbName = new String(event.getTableName().getBytes(ISO_8859_1), charset);
             event.setTblname(charsetTbName);
         } catch (UnsupportedEncodingException e) {
             throw new CanalParseException(e);
@@ -550,7 +550,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
             rowChangeBuider.setIsDdl(false);
 
             rowChangeBuider.setEventType(eventType);
-            RowsLogBuffer buffer = event.getRowsBuf(charset.name());
+            RowsLogBuffer buffer = event.getRowsBuf(charset);
             BitSet columns = event.getColumns();
             BitSet changeColumns = event.getChangeColumns();
 

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

@@ -31,12 +31,12 @@ import com.google.common.cache.LoadingCache;
  */
 public class TableMetaCache {
 
-    public static final String              COLUMN_NAME    = "Field";
-    public static final String              COLUMN_TYPE    = "Type";
-    public static final String              IS_NULLABLE    = "Null";
-    public static final String              COLUMN_KEY     = "Key";
-    public static final String              COLUMN_DEFAULT = "Default";
-    public static final String              EXTRA          = "Extra";
+    public static final String              COLUMN_NAME    = "field";
+    public static final String              COLUMN_TYPE    = "type";
+    public static final String              IS_NULLABLE    = "null";
+    public static final String              COLUMN_KEY     = "key";
+    public static final String              COLUMN_DEFAULT = "default";
+    public static final String              EXTRA          = "extra";
     private MysqlConnection                 connection;
     private boolean                         isOnRDS        = false;
     private boolean                         isOnPolarX     = false;
@@ -125,7 +125,7 @@ public class TableMetaCache {
         Map<String, Integer> nameMaps = new HashMap<>(6, 1f);
         int index = 0;
         for (FieldPacket fieldPacket : packet.getFieldDescriptors()) {
-            nameMaps.put(fieldPacket.getName(), index++);
+            nameMaps.put(StringUtils.lowerCase(fieldPacket.getName()), index++);
         }
 
         int size = packet.getFieldDescriptors().size();

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

@@ -100,7 +100,7 @@ public class DdlResult {
     @Override
     public String toString() {
         DdlResult ddlResult = this;
-        StringBuffer sb = new StringBuffer();
+        StringBuilder sb = new StringBuilder();
         do {
             sb.append(String.format("DdlResult [schemaName=%s , tableName=%s , oriSchemaName=%s , oriTableName=%s , type=%s ];",
                 ddlResult.schemaName,

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

@@ -146,7 +146,7 @@ public class MemoryTableMeta implements TableMetaTSDB {
     public Map<String, String> snapshot() {
         Map<String, String> schemaDdls = new HashMap<>();
         for (Schema schema : repository.getSchemas()) {
-            StringBuffer data = new StringBuffer(4 * 1024);
+            StringBuilder data = new StringBuilder(4 * 1024);
             for (String table : schema.showTables()) {
                 SchemaObject schemaObject = schema.findTable(table);
                 schemaObject.getStatement().output(data);

parse/src/main/java/com/alibaba/otter/canal/parse/index/FileMixedLogPositionManager.java

@@ -182,7 +182,7 @@ public class FileMixedLogPositionManager extends AbstractLogPositionManager {
                 return null;
             }
 
-            String json = FileUtils.readFileToString(dataFile, charset.name());
+            String json = FileUtils.readFileToString(dataFile, charset);
             return JsonUtils.unmarshalFromString(json, LogPosition.class);
         } catch (IOException e) {
             throw new CanalMetaManagerException(e);

parse/src/test/java/com/alibaba/otter/canal/parse/DirectLogFetcherTest.java

@@ -302,7 +302,7 @@ public class DirectLogFetcherTest {
                 event.getHeader().getLogPos() - event.getHeader().getEventLen(),
                 event.getTable().getDbName(),
                 event.getTable().getTableName()));
-            RowsLogBuffer buffer = event.getRowsBuf(charset.name());
+            RowsLogBuffer buffer = event.getRowsBuf(charset);
             BitSet columns = event.getColumns();
             BitSet changeColumns = event.getChangeColumns();
             while (buffer.nextOneRow(columns)) {

parse/src/test/java/com/alibaba/otter/canal/parse/MysqlBinlogParsePerformanceTest.java

@@ -134,7 +134,7 @@ public class MysqlBinlogParsePerformanceTest {
 
     public static void parseRowsEvent(RowsLogEvent event, AtomicLong sum) {
         try {
-            RowsLogBuffer buffer = event.getRowsBuf(charset.name());
+            RowsLogBuffer buffer = event.getRowsBuf(charset);
             BitSet columns = event.getColumns();
             BitSet changeColumns = event.getChangeColumns();
             while (buffer.nextOneRow(columns)) {

pom.xml

@@ -58,7 +58,7 @@
         </repository>
         <repository>
             <id>java.net</id>
-            <url>http://download.java.net/maven/2/</url>
+            <url>https://download.java.net/maven/2/</url>
             <releases>
                 <enabled>true</enabled>
             </releases>
@@ -222,7 +222,7 @@
             <dependency>
                 <groupId>com.alibaba.fastjson2</groupId>
                 <artifactId>fastjson2</artifactId>
-                <version>2.0.7</version>
+                <version>2.0.15</version>
             </dependency>
             <dependency>
                 <groupId>com.google.guava</groupId>
@@ -262,7 +262,7 @@
             <dependency>
                 <groupId>com.alibaba</groupId>
                 <artifactId>druid</artifactId>
-                <version>1.2.12</version>
+                <version>1.2.15</version>
             </dependency>
             <dependency>
                 <groupId>com.lmax</groupId>

protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java

@@ -27,6 +27,7 @@ public class FlatMessage implements Serializable {
     private Map<String, String>       mysqlType;
     private List<Map<String, String>> data;
     private List<Map<String, String>> old;
+    private String                    gtid;
 
     public FlatMessage() {
     }
@@ -146,10 +147,18 @@ public class FlatMessage implements Serializable {
         this.es = es;
     }
 
+    public String getGtid() {
+        return gtid;
+    }
+
+    public void setGtid(String gtid) {
+        this.gtid = gtid;
+    }
+
     @Override
     public String toString() {
         return "FlatMessage [id=" + id + ", database=" + database + ", table=" + table + ", isDdl=" + isDdl + ", type="
                + type + ", es=" + es + ", ts=" + ts + ", sql=" + sql + ", sqlType=" + sqlType + ", mysqlType="
-               + mysqlType + ", data=" + data + ", old=" + old + "]";
+               + mysqlType + ", data=" + data + ", old=" + old + ", gtid=" + gtid +"]";
     }
 }