fix bug : full ddl sqls should not assign to extra field but data field

admin/pom.xml

@@ -36,7 +36,7 @@
             <dependency>
                 <groupId>org.springframework.boot</groupId>
                 <artifactId>spring-boot-dependencies</artifactId>
-                <version>2.5.15</version>
+                <version>2.5.4</version>
                 <type>pom</type>
                 <scope>import</scope>
             </dependency>
@@ -48,12 +48,7 @@
             <dependency>
                 <groupId>io.ebean</groupId>
                 <artifactId>ebean</artifactId>
-                <version>11.45.1</version>
+                <version>11.41.1</version>
-            </dependency>
-            <dependency>
-                <groupId>org.yaml</groupId>
-                <artifactId>snakeyaml</artifactId>
-                <version>2.0</version>
             </dependency>
         </dependencies>
     </dependencyManagement>

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

@@ -25,9 +25,6 @@ import org.springframework.util.PropertyPlaceholderHelper;
 public class YamlUtils {
 
     public static <T> T resourceYmlToObj(String resource, String prefix, Class<T> clazz) {
-        if (!StringUtils.startsWithIgnoreCase(resource, "classpath:")) {
-            resource = "classpath:" + resource;
-        }
         ClassPathResource classPathResource = new ClassPathResource(resource);
 
         String content;

client-adapter/launcher/pom.xml

@@ -17,7 +17,7 @@
             <dependency>
                 <groupId>org.springframework.boot</groupId>
                 <artifactId>spring-boot-dependencies</artifactId>
-                <version>2.5.15</version>
+                <version>2.5.4</version>
                 <type>pom</type>
                 <scope>import</scope>
             </dependency>

client-adapter/launcher/src/main/bin/restart.sh

@@ -1,7 +1,5 @@
 #!/bin/bash
 
-args=$@
-
 case $(uname) in
 Linux)
   bin_abs_path=$(readlink -f $(dirname $0))
@@ -11,5 +9,5 @@ Linux)
   ;;
 esac
 
-sh "$bin_abs_path"/stop.sh $args
+sh "$bin_abs_path"/stop.sh
-sh "$bin_abs_path"/startup.sh $args
+sh "$bin_abs_path"/startup.sh

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

@@ -52,7 +52,7 @@ public class AdapterProcessor {
     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,11 +63,8 @@ public class AdapterProcessor {
 
         // load connector consumer
         ExtensionLoader<CanalMsgConsumer> loader = new ExtensionLoader<>(CanalMsgConsumer.class);
-        // see https://github.com/alibaba/canal/pull/5175
+        canalMsgConsumer = new ProxyCanalMsgConsumer(loader
-        String key = destination + "_" + groupId;
+            .getExtension(canalClientConfig.getMode().toLowerCase(), destination, CONNECTOR_SPI_DIR,
-        canalMsgConsumer = new ProxyCanalMsgConsumer(loader.getExtension(canalClientConfig.getMode().toLowerCase(),
-            key,
-            CONNECTOR_SPI_DIR,
                 CONNECTOR_STANDBY_SPI_DIR));
 
         Properties properties = canalClientConfig.getConsumerProperties();

client-adapter/pom.xml

@@ -117,7 +117,7 @@
             <dependency>
                 <groupId>org.springframework.boot</groupId>
                 <artifactId>spring-boot</artifactId>
-                <version>2.5.15</version>
+                <version>2.5.4</version>
             </dependency>
             <dependency>
                 <groupId>com.h2database</groupId>
@@ -238,7 +238,7 @@
             <dependency>
                 <groupId>org.yaml</groupId>
                 <artifactId>snakeyaml</artifactId>
-                <version>2.0</version>
+                <version>1.29</version>
             </dependency>
             <!-- test -->
             <dependency>

common/pom.xml

@@ -90,6 +90,7 @@
 		<dependency>
 			<groupId>com.alibaba</groupId>
 			<artifactId>druid</artifactId>
+			<optional>true</optional>
 		</dependency>
 		<dependency>
 			<groupId>org.springframework</groupId>

connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQConstants.java

@@ -62,17 +62,8 @@ public class PulsarMQConstants {
     public static final String PULSARMQ_ADMIN_SERVER_URL          = ROOT + "." + "adminServerUrl";
 
     /**
-     * Pulsar 监听器名字
+     * Pulsar admin服务器地址
      */
     public static final String PULSARMQ_LISTENER_NAME             = ROOT + "." + "listenerName";
 
-    /**
-     * Pulsar 开启chunking
-     */
-    public static final String PULSARMQ_ENABLE_CHUNKING           = ROOT + "." + "enableChunking";
-
-    /**
-     * Pulsar 压缩算法
-     */
-    public static final String PULSARMQ_COMPRESSION_TYPE          = ROOT + "." + "compressionType";
 }

connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQProducerConfig.java

@@ -42,16 +42,6 @@ public class PulsarMQProducerConfig extends MQProperties {
      */
     private String listenerName;
 
-    /**
-     * enableChunking
-     */
-    private boolean enableChunking;
-
-    /**
-     * compressionType
-     */
-    private String  compressionType;
-
     public String getServerUrl() {
         return serverUrl;
     }
@@ -91,20 +81,4 @@ public class PulsarMQProducerConfig extends MQProperties {
     public void setListenerName(String listenerName) {
         this.listenerName = listenerName;
     }
-
-    public void setEnableChunking(boolean enableChunking) {
-        this.enableChunking = enableChunking;
-    }
-
-    public boolean getEnableChunking() {
-        return this.enableChunking;
-    }
-
-    public void setCompressionType(String compressionType) {
-        this.compressionType = compressionType;
-    }
-
-    public String getCompressionType() {
-        return this.compressionType;
-    }
 }

connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/producer/CanalPulsarMQProducer.java

@@ -137,16 +137,6 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
             tmpProperties.setListenerName(listenerName);
         }
 
-        String enableChunkingStr = PropertiesUtils.getProperty(properties, PulsarMQConstants.PULSARMQ_ENABLE_CHUNKING);
-        if (!StringUtils.isEmpty(enableChunkingStr)) {
-            tmpProperties.setEnableChunking(Boolean.parseBoolean(enableChunkingStr));
-        }
-
-        String compressionType = PropertiesUtils.getProperty(properties, PulsarMQConstants.PULSARMQ_COMPRESSION_TYPE);
-        if (!StringUtils.isEmpty(compressionType)) {
-            tmpProperties.setCompressionType(compressionType);
-        }
-
         if (logger.isDebugEnabled()) {
             logger.debug("Load pulsar properties ==> {}", JSON.toJSON(this.mqProperties));
         }
@@ -418,34 +408,11 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
                     }
 
                     // 创建指定topic的生产者
-                    ProducerBuilder producerBuilder = client.newProducer();
+                    producer = client.newProducer()
-                    if (pulsarMQProperties.getEnableChunking()) {
+                        .topic(fullTopic)
-                        producerBuilder.enableChunking(true);
-                        producerBuilder.enableBatching(false);
-                    }
-
-                    if (!StringUtils.isEmpty(pulsarMQProperties.getCompressionType())) {
-                        switch (pulsarMQProperties.getCompressionType().toLowerCase()) {
-                            case "lz4":
-                                producerBuilder.compressionType(CompressionType.LZ4);
-                                break;
-                            case "zlib":
-                                producerBuilder.compressionType(CompressionType.ZLIB);
-                                break;
-                            case "zstd":
-                                producerBuilder.compressionType(CompressionType.ZSTD);
-                                break;
-                            case "snappy":
-                                producerBuilder.compressionType(CompressionType.SNAPPY);
-                                break;
-                        }
-                    }
-
-                    producer = producerBuilder.topic(fullTopic)
                         // 指定路由器
                         .messageRouter(new MessageRouterImpl(topic))
                         .create();
-
                     // 放入缓存
                     PRODUCERS.put(topic, producer);
                 }

connector/rocketmq-connector/src/main/java/com/alibaba/otter/canal/connector/rocketmq/producer/CanalRocketMQProducer.java

@@ -1,29 +1,5 @@
 package com.alibaba.otter.canal.connector.rocketmq.producer;
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
-
-import org.apache.commons.lang.StringUtils;
-import org.apache.rocketmq.acl.common.AclClientRPCHook;
-import org.apache.rocketmq.acl.common.SessionCredentials;
-import org.apache.rocketmq.client.AccessChannel;
-import org.apache.rocketmq.client.exception.MQClientException;
-import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl;
-import org.apache.rocketmq.client.impl.producer.TopicPublishInfo;
-import org.apache.rocketmq.client.producer.DefaultMQProducer;
-import org.apache.rocketmq.client.producer.SendResult;
-import org.apache.rocketmq.common.message.Message;
-import org.apache.rocketmq.common.message.MessageQueue;
-import org.apache.rocketmq.remoting.RPCHook;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import com.alibaba.fastjson2.JSON;
 import com.alibaba.fastjson2.JSONWriter;
 import com.alibaba.otter.canal.common.CanalException;
@@ -40,6 +16,29 @@ import com.alibaba.otter.canal.connector.core.util.CanalMessageSerializerUtil;
 import com.alibaba.otter.canal.connector.rocketmq.config.RocketMQConstants;
 import com.alibaba.otter.canal.connector.rocketmq.config.RocketMQProducerConfig;
 import com.alibaba.otter.canal.protocol.FlatMessage;
+import org.apache.commons.lang.StringUtils;
+import org.apache.rocketmq.acl.common.AclClientRPCHook;
+import org.apache.rocketmq.acl.common.SessionCredentials;
+import org.apache.rocketmq.client.AccessChannel;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl;
+import org.apache.rocketmq.client.impl.producer.TopicPublishInfo;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.remoting.RPCHook;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
 /**
  * RocketMQ Producer SPI 实现
@@ -47,18 +46,15 @@ import com.alibaba.otter.canal.protocol.FlatMessage;
  * @author rewerma 2020-01-27
  * @version 1.0.0
  */
-@SPI("rocketmq")
+@SPI("rocketmq") public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQProducer {
-public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQProducer {
 
     private static final Logger logger = LoggerFactory.getLogger(CanalRocketMQProducer.class);
 
     private              DefaultMQProducer  defaultMQProducer;
     private static final String             CLOUD_ACCESS_CHANNEL = "cloud";
-    private static final String  NAMESPACE_SEPARATOR  = "%";
     protected            ThreadPoolExecutor sendPartitionExecutor;
 
-    @Override
+    @Override public void init(Properties properties) {
-    public void init(Properties properties) {
         RocketMQProducerConfig rocketMQProperties = new RocketMQProducerConfig();
         this.mqProperties = rocketMQProperties;
         super.init(properties);
@@ -147,14 +143,15 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
         }
     }
 
-    @Override
+    @Override public void send(MQDestination destination, com.alibaba.otter.canal.protocol.Message message,
-    public void send(MQDestination destination, com.alibaba.otter.canal.protocol.Message message, Callback callback) {
+                               Callback callback) {
         ExecutorTemplate template = new ExecutorTemplate(sendExecutor);
         try {
             if (!StringUtils.isEmpty(destination.getDynamicTopic())) {
                 // 动态topic
-                Map<String, com.alibaba.otter.canal.protocol.Message> messageMap = MQMessageUtils
+                Map<String, com.alibaba.otter.canal.protocol.Message> messageMap = MQMessageUtils.messageTopics(message,
-                    .messageTopics(message, destination.getTopic(), destination.getDynamicTopic());
+                        destination.getTopic(),
+                        destination.getDynamicTopic());
 
                 for (Map.Entry<String, com.alibaba.otter.canal.protocol.Message> entry : messageMap.entrySet()) {
                     String topicName = entry.getKey().replace('.', '_');
@@ -266,9 +263,7 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
                             List<Message> messages = flatMessagePart.stream()
                                     .map(flatMessage -> new Message(topicName,
                                             ((RocketMQProducerConfig) this.mqProperties).getTag(),
-                                    JSON.toJSONBytes(flatMessage,
+                                            JSON.toJSONBytes(flatMessage, JSONWriter.Feature.WriteNulls, JSONWriter.Feature.LargeObject)))
-                                        JSONWriter.Feature.WriteNulls,
-                                        JSONWriter.Feature.LargeObject)))
                                     .collect(Collectors.toList());
                             // 批量发送
                             sendMessage(messages, index);
@@ -283,7 +278,7 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
                 List<Message> messages = flatMessages.stream()
                         .map(flatMessage -> new Message(topicName,
                                 ((RocketMQProducerConfig) this.mqProperties).getTag(),
-                        JSON.toJSONBytes(flatMessage, JSONWriter.Feature.WriteNulls, JSONWriter.Feature.LargeObject)))
+                                JSON.toJSONBytes(flatMessage, JSONWriter.Feature.WriteNulls,JSONWriter.Feature.LargeObject)))
                         .collect(Collectors.toList());
                 // 批量发送
                 sendMessage(messages, partition);
@@ -309,19 +304,14 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
         }
     }
 
-    @SuppressWarnings("deprecation")
+    @SuppressWarnings("deprecation") private void sendMessage(List<Message> messages, int partition) {
-    private void sendMessage(List<Message> messages, int partition) {
         if (messages.isEmpty()) {
             return;
         }
 
         // 获取一下messageQueue
         DefaultMQProducerImpl innerProducer = this.defaultMQProducer.getDefaultMQProducerImpl();
-        String topic = messages.get(0).getTopic();
+        TopicPublishInfo topicInfo = innerProducer.getTopicPublishInfoTable().get(messages.get(0).getTopic());
-        if (StringUtils.isNotBlank(this.defaultMQProducer.getNamespace())) {
-            topic = this.defaultMQProducer.getNamespace() + NAMESPACE_SEPARATOR + topic;
-        }
-        TopicPublishInfo topicInfo = innerProducer.getTopicPublishInfoTable().get(topic);
         if (topicInfo == null) {
             for (Message message : messages) {
                 sendMessage(message, partition);
@@ -356,8 +346,7 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
         }
     }
 
-    @Override
+    @Override public void stop() {
-    public void stop() {
         logger.info("## Stop RocketMQ producer##");
         this.defaultMQProducer.shutdown();
         if (sendPartitionExecutor != null) {

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

@@ -819,7 +819,7 @@ public class QueryLogEvent extends LogEvent {
             case Q_MASTER_DATA_WRITTEN_CODE:
                 return "Q_MASTER_DATA_WRITTEN_CODE";
             case Q_INVOKER:
-                return "Q_INVOKER";
+                return "case Q_INVOKER";
             case Q_MICROSECONDS:
                 return "Q_MICROSECONDS";
             case Q_UPDATED_DB_NAMES:

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

@@ -5,11 +5,6 @@
 # enable gtid use true/false
 canal.instance.gtidon=false
 
-# rds oss binlog
-canal.instance.rds.accesskey=
-canal.instance.rds.secretkey=
-canal.instance.rds.instanceId=
-
 # position info
 canal.instance.master.address=127.0.0.1:3306
 canal.instance.master.journal.name=
@@ -17,15 +12,10 @@ canal.instance.master.position=
 canal.instance.master.timestamp=
 canal.instance.master.gtid=
 
-# ssl
+# rds oss binlog
-#canal.instance.master.sslMode=DISABLED
+canal.instance.rds.accesskey=
-#canal.instance.master.tlsVersions=
+canal.instance.rds.secretkey=
-#canal.instance.master.trustCertificateKeyStoreType=
+canal.instance.rds.instanceId=
-#canal.instance.master.trustCertificateKeyStoreUrl=
-#canal.instance.master.trustCertificateKeyStorePassword=
-#canal.instance.master.clientCertificateKeyStoreType=
-#canal.instance.master.clientCertificateKeyStoreUrl=
-#canal.instance.master.clientCertificateKeyStorePassword=
 
 # table meta tsdb info
 canal.instance.tsdb.enable=true

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

@@ -111,6 +111,7 @@
 		<property name="defaultConnectionTimeoutInSeconds" value="${canal.instance.network.soTimeout:30}" />
 
 		<!-- 解析编码 -->
+		<!-- property name="connectionCharsetNumber" value="${canal.instance.connectionCharsetNumber:33}" /-->
 		<property name="connectionCharset" value="${canal.instance.connectionCharset:UTF-8}" />
 
 		<!-- 解析位点记录 -->
@@ -139,18 +140,6 @@
 				<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
 				<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
 				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
-				<property name="sslInfo">
-					<bean class="com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo">
-						<property name="sslMode" value="${canal.instance.master.sslMode:DISABLED}"/>
-						<property name="tlsVersions" value="${canal.instance.master.tlsVersions:}"/>
-						<property name="trustCertificateKeyStoreType" value="${canal.instance.master.trustCertificateKeyStoreType:}"/>
-						<property name="trustCertificateKeyStoreUrl" value="${canal.instance.master.trustCertificateKeyStoreUrl:}"/>
-						<property name="trustCertificateKeyStorePassword" value="${canal.instance.master.trustCertificateKeyStorePassword:}"/>
-						<property name="clientCertificateKeyStoreType" value="${canal.instance.master.clientCertificateKeyStoreType:}"/>
-						<property name="clientCertificateKeyStoreUrl" value="${canal.instance.master.clientCertificateKeyStoreUrl:}"/>
-						<property name="clientCertificateKeyStorePassword" value="${canal.instance.master.clientCertificateKeyStorePassword:}"/>
-					</bean>
-				</property>
 			</bean>
 		</property>
 		<property name="standbyInfo">
@@ -161,18 +150,6 @@
 				<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
 				<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
 				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
-				<property name="sslInfo">
-					<bean class="com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo">
-						<property name="sslMode" value="${canal.instance.standby.sslMode:DISABLED}"/>
-						<property name="tlsVersions" value="${canal.instance.standby.tlsVersions:}"/>
-						<property name="trustCertificateKeyStoreType" value="${canal.instance.standby.trustCertificateKeyStoreType:}"/>
-						<property name="trustCertificateKeyStoreUrl" value="${canal.instance.standby.trustCertificateKeyStoreUrl:}"/>
-						<property name="trustCertificateKeyStorePassword" value="${canal.instance.standby.trustCertificateKeyStorePassword:}"/>
-						<property name="clientCertificateKeyStoreType" value="${canal.instance.standby.clientCertificateKeyStoreType:}"/>
-						<property name="clientCertificateKeyStoreUrl" value="${canal.instance.standby.clientCertificateKeyStoreUrl:}"/>
-						<property name="clientCertificateKeyStorePassword" value="${canal.instance.standby.clientCertificateKeyStorePassword:}"/>
-					</bean>
-				</property>
 			</bean>
 		</property>
 

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

@@ -97,6 +97,7 @@
 		<property name="defaultConnectionTimeoutInSeconds" value="${canal.instance.network.soTimeout:30}" />
 
 		<!-- 解析编码 -->
+		<!-- property name="connectionCharsetNumber" value="${canal.instance.connectionCharsetNumber:33}" /-->
 		<property name="connectionCharset" value="${canal.instance.connectionCharset:UTF-8}" />
 
 		<!-- 解析位点记录 -->
@@ -125,18 +126,6 @@
 				<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
 				<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
 				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
-				<property name="sslInfo">
-					<bean class="com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo">
-						<property name="sslMode" value="${canal.instance.master.sslMode:DISABLED}"/>
-						<property name="tlsVersions" value="${canal.instance.master.tlsVersions:}"/>
-						<property name="trustCertificateKeyStoreType" value="${canal.instance.master.trustCertificateKeyStoreType:}"/>
-						<property name="trustCertificateKeyStoreUrl" value="${canal.instance.master.trustCertificateKeyStoreUrl:}"/>
-						<property name="trustCertificateKeyStorePassword" value="${canal.instance.master.trustCertificateKeyStorePassword:}"/>
-						<property name="clientCertificateKeyStoreType" value="${canal.instance.master.clientCertificateKeyStoreType:}"/>
-						<property name="clientCertificateKeyStoreUrl" value="${canal.instance.master.clientCertificateKeyStoreUrl:}"/>
-						<property name="clientCertificateKeyStorePassword" value="${canal.instance.master.clientCertificateKeyStorePassword:}"/>
-					</bean>
-				</property>
 			</bean>
 		</property>
 		<property name="standbyInfo">
@@ -147,18 +136,6 @@
 				<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
 				<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
 				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
-				<property name="sslInfo">
-					<bean class="com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo">
-						<property name="sslMode" value="${canal.instance.standby.sslMode:DISABLED}"/>
-						<property name="tlsVersions" value="${canal.instance.standby.tlsVersions:}"/>
-						<property name="trustCertificateKeyStoreType" value="${canal.instance.standby.trustCertificateKeyStoreType:}"/>
-						<property name="trustCertificateKeyStoreUrl" value="${canal.instance.standby.trustCertificateKeyStoreUrl:}"/>
-						<property name="trustCertificateKeyStorePassword" value="${canal.instance.standby.trustCertificateKeyStorePassword:}"/>
-						<property name="clientCertificateKeyStoreType" value="${canal.instance.standby.clientCertificateKeyStoreType:}"/>
-						<property name="clientCertificateKeyStoreUrl" value="${canal.instance.standby.clientCertificateKeyStoreUrl:}"/>
-						<property name="clientCertificateKeyStorePassword" value="${canal.instance.standby.clientCertificateKeyStorePassword:}"/>
-					</bean>
-				</property>
 			</bean>
 		</property>
 

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

@@ -103,6 +103,7 @@
 		<property name="defaultConnectionTimeoutInSeconds" value="${canal.instance.network.soTimeout:30}" />
 
 		<!-- 解析编码 -->
+		<!-- property name="connectionCharsetNumber" value="${canal.instance.connectionCharsetNumber:33}" /-->
 		<property name="connectionCharset" value="${canal.instance.connectionCharset:UTF-8}" />
 
 		<!-- 解析位点记录 -->
@@ -122,18 +123,6 @@
 				<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
 				<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
 				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
-				<property name="sslInfo">
-					<bean class="com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo">
-						<property name="sslMode" value="${canal.instance.master.sslMode:DISABLED}"/>
-						<property name="tlsVersions" value="${canal.instance.master.tlsVersions:}"/>
-						<property name="trustCertificateKeyStoreType" value="${canal.instance.master.trustCertificateKeyStoreType:}"/>
-						<property name="trustCertificateKeyStoreUrl" value="${canal.instance.master.trustCertificateKeyStoreUrl:}"/>
-						<property name="trustCertificateKeyStorePassword" value="${canal.instance.master.trustCertificateKeyStorePassword:}"/>
-						<property name="clientCertificateKeyStoreType" value="${canal.instance.master.clientCertificateKeyStoreType:}"/>
-						<property name="clientCertificateKeyStoreUrl" value="${canal.instance.master.clientCertificateKeyStoreUrl:}"/>
-						<property name="clientCertificateKeyStorePassword" value="${canal.instance.master.clientCertificateKeyStorePassword:}"/>
-					</bean>
-				</property>
 			</bean>
 		</property>
 		<property name="standbyInfo">
@@ -144,18 +133,6 @@
 				<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
 				<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
 				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
-				<property name="sslInfo">
-					<bean class="com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo">
-						<property name="sslMode" value="${canal.instance.standby.sslMode:DISABLED}"/>
-						<property name="tlsVersions" value="${canal.instance.standby.tlsVersions:}"/>
-						<property name="trustCertificateKeyStoreType" value="${canal.instance.standby.trustCertificateKeyStoreType:}"/>
-						<property name="trustCertificateKeyStoreUrl" value="${canal.instance.standby.trustCertificateKeyStoreUrl:}"/>
-						<property name="trustCertificateKeyStorePassword" value="${canal.instance.standby.trustCertificateKeyStorePassword:}"/>
-						<property name="clientCertificateKeyStoreType" value="${canal.instance.standby.clientCertificateKeyStoreType:}"/>
-						<property name="clientCertificateKeyStoreUrl" value="${canal.instance.standby.clientCertificateKeyStoreUrl:}"/>
-						<property name="clientCertificateKeyStorePassword" value="${canal.instance.standby.clientCertificateKeyStorePassword:}"/>
-					</bean>
-				</property>
 			</bean>
 		</property>
 
@@ -234,6 +211,7 @@
 		<property name="defaultConnectionTimeoutInSeconds" value="${canal.instance.network.soTimeout:30}" />
 
 		<!-- 解析编码 -->
+		<!-- property name="connectionCharsetNumber" value="${canal.instance.connectionCharsetNumber:33}" /-->
 		<property name="connectionCharset" value="${canal.instance.connectionCharset:UTF-8}" />
 
 		<!-- 解析位点记录 -->
@@ -253,18 +231,6 @@
 				<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
 				<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
 				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
-				<property name="sslInfo">
-					<bean class="com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo">
-						<property name="sslMode" value="${canal.instance.master.sslMode:DISABLED}"/>
-						<property name="tlsVersions" value="${canal.instance.master.tlsVersions:}"/>
-						<property name="trustCertificateKeyStoreType" value="${canal.instance.master.trustCertificateKeyStoreType:}"/>
-						<property name="trustCertificateKeyStoreUrl" value="${canal.instance.master.trustCertificateKeyStoreUrl:}"/>
-						<property name="trustCertificateKeyStorePassword" value="${canal.instance.master.trustCertificateKeyStorePassword:}"/>
-						<property name="clientCertificateKeyStoreType" value="${canal.instance.master.clientCertificateKeyStoreType:}"/>
-						<property name="clientCertificateKeyStoreUrl" value="${canal.instance.master.clientCertificateKeyStoreUrl:}"/>
-						<property name="clientCertificateKeyStorePassword" value="${canal.instance.master.clientCertificateKeyStorePassword:}"/>
-					</bean>
-				</property>
 			</bean>
 		</property>
 		<property name="standbyInfo">
@@ -275,18 +241,6 @@
 				<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
 				<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
 				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
-				<property name="sslInfo">
-					<bean class="com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo">
-						<property name="sslMode" value="${canal.instance.standby.sslMode:DISABLED}"/>
-						<property name="tlsVersions" value="${canal.instance.standby.tlsVersions:}"/>
-						<property name="trustCertificateKeyStoreType" value="${canal.instance.standby.trustCertificateKeyStoreType:}"/>
-						<property name="trustCertificateKeyStoreUrl" value="${canal.instance.standby.trustCertificateKeyStoreUrl:}"/>
-						<property name="trustCertificateKeyStorePassword" value="${canal.instance.standby.trustCertificateKeyStorePassword:}"/>
-						<property name="clientCertificateKeyStoreType" value="${canal.instance.standby.clientCertificateKeyStoreType:}"/>
-						<property name="clientCertificateKeyStoreUrl" value="${canal.instance.standby.clientCertificateKeyStoreUrl:}"/>
-						<property name="clientCertificateKeyStorePassword" value="${canal.instance.standby.clientCertificateKeyStorePassword:}"/>
-					</bean>
-				</property>
 			</bean>
 		</property>
 

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

@@ -94,6 +94,7 @@
 		<property name="defaultConnectionTimeoutInSeconds" value="${canal.instance.network.soTimeout:30}" />
 
 		<!-- 解析编码 -->
+		<!-- property name="connectionCharsetNumber" value="${canal.instance.connectionCharsetNumber:33}" /-->
 		<property name="connectionCharset" value="${canal.instance.connectionCharset:UTF-8}" />
 
 		<!-- 解析位点记录 -->
@@ -113,18 +114,6 @@
 				<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
 				<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
 				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
-				<property name="sslInfo">
-					<bean class="com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo">
-						<property name="sslMode" value="${canal.instance.master.sslMode:DISABLED}"/>
-						<property name="tlsVersions" value="${canal.instance.master.tlsVersions:}"/>
-						<property name="trustCertificateKeyStoreType" value="${canal.instance.master.trustCertificateKeyStoreType:}"/>
-						<property name="trustCertificateKeyStoreUrl" value="${canal.instance.master.trustCertificateKeyStoreUrl:}"/>
-						<property name="trustCertificateKeyStorePassword" value="${canal.instance.master.trustCertificateKeyStorePassword:}"/>
-						<property name="clientCertificateKeyStoreType" value="${canal.instance.master.clientCertificateKeyStoreType:}"/>
-						<property name="clientCertificateKeyStoreUrl" value="${canal.instance.master.clientCertificateKeyStoreUrl:}"/>
-						<property name="clientCertificateKeyStorePassword" value="${canal.instance.master.clientCertificateKeyStorePassword:}"/>
-					</bean>
-				</property>
 			</bean>
 		</property>
 		<property name="standbyInfo">
@@ -135,18 +124,6 @@
 				<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
 				<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
 				<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
-				<property name="sslInfo">
-					<bean class="com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo">
-						<property name="sslMode" value="${canal.instance.standby.sslMode:DISABLED}"/>
-						<property name="tlsVersions" value="${canal.instance.standby.tlsVersions:}"/>
-						<property name="trustCertificateKeyStoreType" value="${canal.instance.standby.trustCertificateKeyStoreType:}"/>
-						<property name="trustCertificateKeyStoreUrl" value="${canal.instance.standby.trustCertificateKeyStoreUrl:}"/>
-						<property name="trustCertificateKeyStorePassword" value="${canal.instance.standby.trustCertificateKeyStorePassword:}"/>
-						<property name="clientCertificateKeyStoreType" value="${canal.instance.standby.clientCertificateKeyStoreType:}"/>
-						<property name="clientCertificateKeyStoreUrl" value="${canal.instance.standby.clientCertificateKeyStoreUrl:}"/>
-						<property name="clientCertificateKeyStorePassword" value="${canal.instance.standby.clientCertificateKeyStorePassword:}"/>
-					</bean>
-				</property>
 			</bean>
 		</property>
 

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

@@ -1,7 +1,5 @@
 package com.alibaba.otter.canal.parse.driver.mysql;
 
-import static com.alibaba.otter.canal.parse.driver.mysql.packets.Capability.CLIENT_SSL;
-
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.security.DigestException;
@@ -12,12 +10,12 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import com.alibaba.otter.canal.parse.driver.mysql.packets.HeaderPacket;
-import com.alibaba.otter.canal.parse.driver.mysql.packets.client.*;
+import com.alibaba.otter.canal.parse.driver.mysql.packets.client.AuthSwitchResponsePacket;
+import com.alibaba.otter.canal.parse.driver.mysql.packets.client.ClientAuthenticationPacket;
+import com.alibaba.otter.canal.parse.driver.mysql.packets.client.QuitCommandPacket;
 import com.alibaba.otter.canal.parse.driver.mysql.packets.server.*;
 import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannel;
 import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannelPool;
-import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo;
-import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslMode;
 import com.alibaba.otter.canal.parse.driver.mysql.utils.MSC;
 import com.alibaba.otter.canal.parse.driver.mysql.utils.MySQLPasswordEncrypter;
 import com.alibaba.otter.canal.parse.driver.mysql.utils.PacketManager;
@@ -30,17 +28,18 @@ import com.alibaba.otter.canal.parse.driver.mysql.utils.PacketManager;
  */
 public class MysqlConnector {
 
-    public static final int     timeout           = 5 * 1000;                                     // 5s
     private static final Logger logger            = LoggerFactory.getLogger(MysqlConnector.class);
     private InetSocketAddress   address;
     private String              username;
     private String              password;
-    private SslInfo             sslInfo;
+
+    private byte                charsetNumber     = 33;
     private String              defaultSchema;
     private int                 soTimeout         = 30 * 1000;
     private int                 connTimeout       = 5 * 1000;
     private int                 receiveBufferSize = 16 * 1024;
     private int                 sendBufferSize    = 16 * 1024;
+
     private SocketChannel       channel;
     private volatile boolean    dumping           = false;
     // mysql connectionId
@@ -49,6 +48,8 @@ public class MysqlConnector {
     // serverVersion
     private String              serverVersion;
 
+    public static final int     timeout           = 5 * 1000;                                     // 5s
+
     public MysqlConnector(){
     }
 
@@ -61,25 +62,20 @@ public class MysqlConnector {
         this.password = password;
     }
 
-    public MysqlConnector(InetSocketAddress address, String username, String password, String defaultSchema){
+    public MysqlConnector(InetSocketAddress address, String username, String password, byte charsetNumber,
+                          String defaultSchema){
         this(address, username, password);
 
+        this.charsetNumber = charsetNumber;
         this.defaultSchema = defaultSchema;
     }
 
-    public MysqlConnector(InetSocketAddress address, String username, String password, String defaultSchema,
-                          SslInfo sslInfo){
-        this(address, username, password, defaultSchema);
-        this.sslInfo = sslInfo;
-    }
-
     public void connect() throws IOException {
         if (connected.compareAndSet(false, true)) {
             try {
                 channel = SocketChannelPool.open(address);
                 logger.info("connect MysqlConnection to {}...", address);
                 negotiate(channel);
-                printSslStatus();
             } catch (Exception e) {
                 disconnect();
                 throw new IOException("connect " + this.address + " failure", e);
@@ -89,28 +85,6 @@ public class MysqlConnector {
         }
     }
 
-    private void printSslStatus() {
-        try {
-            MysqlQueryExecutor executor = new MysqlQueryExecutor(this);
-            ResultSetPacket result = executor.query("SHOW STATUS LIKE 'Ssl_version'");
-            String sslVersion = "";
-            if (result.getFieldValues() != null && result.getFieldValues().size() >= 2) {
-                sslVersion = result.getFieldValues().get(1);
-            }
-            result = executor.query("SHOW STATUS LIKE 'Ssl_cipher'");
-            String sslCipher = "";
-            if (result.getFieldValues() != null && result.getFieldValues().size() >= 2) {
-                sslCipher = result.getFieldValues().get(1);
-            }
-            logger.info("connect MysqlConnection in sslMode {}, Ssl_version:{}, Ssl_cipher:{}",
-                (sslInfo != null ? sslInfo.getSslMode() : SslMode.DISABLED),
-                sslVersion,
-                sslCipher);
-        } catch (Exception e) {
-            logger.warn("Can't show SSL status, server may not standard MySQL server", e);
-        }
-    }
-
     public void reconnect() throws IOException {
         disconnect();
         connect();
@@ -157,6 +131,7 @@ public class MysqlConnector {
 
     public MysqlConnector fork() {
         MysqlConnector connector = new MysqlConnector();
+        connector.setCharsetNumber(getCharsetNumber());
         connector.setDefaultSchema(getDefaultSchema());
         connector.setAddress(getAddress());
         connector.setPassword(password);
@@ -165,7 +140,6 @@ public class MysqlConnector {
         connector.setSendBufferSize(getSendBufferSize());
         connector.setSoTimeout(getSoTimeout());
         connector.setConnTimeout(connTimeout);
-        connector.setSslInfo(getSslInfo());
         return connector;
     }
 
@@ -196,23 +170,6 @@ public class MysqlConnector {
         }
         HandshakeInitializationPacket handshakePacket = new HandshakeInitializationPacket();
         handshakePacket.fromBytes(body);
-        byte serverCharsetNumber = (handshakePacket.serverCharsetNumber != 0) ? handshakePacket.serverCharsetNumber : 33;
-        SslMode sslMode = sslInfo != null ? sslInfo.getSslMode() : SslMode.DISABLED;
-        if (sslMode != SslMode.DISABLED) {
-            boolean serverSupportSsl = (handshakePacket.serverCapabilities & CLIENT_SSL) > 0;
-            if (!serverSupportSsl) {
-                throw new IOException("MySQL Server does not support SSL: " + address + " serverCapabilities: "
-                                      + handshakePacket.serverCapabilities);
-            }
-            byte[] sslPacket = new SslRequestCommandPacket(serverCharsetNumber).toBytes();
-            HeaderPacket sslHeader = new HeaderPacket();
-            sslHeader.setPacketBodyLength(sslPacket.length);
-            sslHeader.setPacketSequenceNumber((byte) (header.getPacketSequenceNumber() + 1));
-            header.setPacketSequenceNumber((byte) (header.getPacketSequenceNumber() + 1));
-            PacketManager.writePkg(channel, sslHeader.toBytes(), sslPacket);
-            channel = SocketChannelPool.connectSsl(channel, sslInfo);
-            this.channel = channel;
-        }
         if (handshakePacket.protocolVersion != MSC.DEFAULT_PROTOCOL_VERSION) {
             // HandshakeV9
             auth323(channel, (byte) (header.getPacketSequenceNumber() + 1), handshakePacket.seed);
@@ -222,27 +179,15 @@ public class MysqlConnector {
         connectionId = handshakePacket.threadId; // 记录一下connection
         serverVersion = handshakePacket.serverVersion; // 记录serverVersion
         logger.info("handshake initialization packet received, prepare the client authentication packet to send");
-        // 某些老协议的 server 默认不返回 auth plugin，需要使用默认的 mysql_native_password
+        ClientAuthenticationPacket clientAuth = new ClientAuthenticationPacket();
-        String authPluginName = (handshakePacket.authPluginName != null
+        clientAuth.setCharsetNumber(charsetNumber);
-                                 && handshakePacket.authPluginName.length > 0) ? new String(
-                                     handshakePacket.authPluginName) : "mysql_native_password";
-        logger.info("auth plugin: {}", authPluginName);
-        boolean isSha2Password = false;
-        ClientAuthenticationPacket clientAuth;
-        if ("caching_sha2_password".equals(authPluginName)) {
-            clientAuth = new ClientAuthenticationSHA2Packet();
-            isSha2Password = true;
-        } else {
-            clientAuth = new ClientAuthenticationPacket();
-        }
-        clientAuth.setCharsetNumber(serverCharsetNumber);
 
         clientAuth.setUsername(username);
         clientAuth.setPassword(password);
         clientAuth.setServerCapabilities(handshakePacket.serverCapabilities);
         clientAuth.setDatabaseName(defaultSchema);
         clientAuth.setScrumbleBuff(joinAndCreateScrumbleBuff(handshakePacket));
-        clientAuth.setAuthPluginName(authPluginName.getBytes());
+        clientAuth.setAuthPluginName("mysql_native_password".getBytes());
 
         byte[] clientAuthPkgBody = clientAuth.toBytes();
         HeaderPacket h = new HeaderPacket();
@@ -260,12 +205,6 @@ public class MysqlConnector {
         assert body != null;
         byte marker = body[0];
         if (marker == -2 || marker == 1) {
-            if (isSha2Password && body[1] == 3) {
-                // sha2 auth ok
-                logger.info("caching_sha2_password auth success.");
-                header = PacketManager.readHeader(channel, 4);
-                body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout);
-            } else {
             byte[] authData = null;
             String pluginName = null;
             if (marker == 1) {
@@ -285,7 +224,7 @@ public class MysqlConnector {
                 encryptedPassword = getPassword().getBytes();
                 header = authSwitchAfterAuth(encryptedPassword, header);
                 body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout);
-                } else if (pluginName == null || "mysql_native_password".equals(pluginName)) {
+            } else if ("mysql_native_password".equals(pluginName)) {
                 try {
                     encryptedPassword = MySQLPasswordEncrypter.scramble411(getPassword().getBytes(), authData);
                 } catch (NoSuchAlgorithmException e) {
@@ -296,8 +235,7 @@ public class MysqlConnector {
             } else if ("caching_sha2_password".equals(pluginName)) {
                 byte[] scramble = authData;
                 try {
-                        encryptedPassword = MySQLPasswordEncrypter.scrambleCachingSha2(getPassword().getBytes(),
+                    encryptedPassword = MySQLPasswordEncrypter.scrambleCachingSha2(getPassword().getBytes(), scramble);
-                            scramble);
                 } catch (DigestException e) {
                     throw new RuntimeException("can't encrypt password that will be sent to MySQL server.", e);
                 }
@@ -317,7 +255,7 @@ public class MysqlConnector {
                 body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout);
             }
         }
-        }
+        assert body != null;
         if (body[0] < 0) {
             if (body[0] == -1) {
                 ErrorPacket err = new ErrorPacket();
@@ -442,6 +380,14 @@ public class MysqlConnector {
         this.username = username;
     }
 
+    public byte getCharsetNumber() {
+        return charsetNumber;
+    }
+
+    public void setCharsetNumber(byte charsetNumber) {
+        this.charsetNumber = charsetNumber;
+    }
+
     public String getDefaultSchema() {
         return defaultSchema;
     }
@@ -482,6 +428,10 @@ public class MysqlConnector {
         this.channel = channel;
     }
 
+    public void setPassword(String password) {
+        this.password = password;
+    }
+
     public long getConnectionId() {
         return connectionId;
     }
@@ -510,19 +460,8 @@ public class MysqlConnector {
         return password;
     }
 
-    public void setPassword(String password) {
-        this.password = password;
-    }
-
     public String getServerVersion() {
         return serverVersion;
     }
 
-    public SslInfo getSslInfo() {
-        return sslInfo;
-    }
-
-    public void setSslInfo(SslInfo sslInfo) {
-        this.sslInfo = sslInfo;
-    }
 }

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

@@ -1,65 +0,0 @@
-package com.alibaba.otter.canal.parse.driver.mysql.packets.client;
-
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-
-import com.alibaba.otter.canal.parse.driver.mysql.packets.Capability;
-import com.alibaba.otter.canal.parse.driver.mysql.utils.ByteHelper;
-import com.alibaba.otter.canal.parse.driver.mysql.utils.MSC;
-import com.alibaba.otter.canal.parse.driver.mysql.utils.MySQLPasswordEncrypter;
-
-import org.apache.commons.lang.StringUtils;
-
-public class ClientAuthenticationSHA2Packet extends ClientAuthenticationPacket {
-
-    @Override
-    public byte[] toBytes() throws IOException {
-        ByteArrayOutputStream out = new ByteArrayOutputStream();
-        // 1. write client_flags
-        int clientCapability = 0;
-        clientCapability |= Capability.CLIENT_LONG_PASSWORD;
-        clientCapability |= Capability.CLIENT_LONG_FLAG;
-        clientCapability |= Capability.CLIENT_PROTOCOL_41;
-        clientCapability |= Capability.CLIENT_TRANSACTIONS;
-        clientCapability |= Capability.CLIENT_SECURE_CONNECTION;
-        clientCapability |= Capability.CLIENT_MULTI_STATEMENTS;
-        clientCapability |= Capability.CLIENT_PLUGIN_AUTH;
-        clientCapability |= Capability.CLIENT_PLUGIN_AUTH_LENENC_CLIENT_DATA;
-        if (getDatabaseName() != null) {
-            clientCapability |= Capability.CLIENT_CONNECT_WITH_DB;
-        }
-        ByteHelper.writeUnsignedIntLittleEndian(clientCapability, out);
-
-        // 2. write max_packet_size
-        ByteHelper.writeUnsignedIntLittleEndian(MSC.MAX_PACKET_LENGTH, out);
-        // 3. write charset_number
-        out.write(getCharsetNumber());
-        // 4. write (filler) always 0x00...
-        out.write(new byte[23]);
-        // 5. write (Null-Terminated String) user
-        ByteHelper.writeNullTerminatedString(getUsername(), out);
-        // 6. write (Length Coded Binary) scramble_buff (1 + x bytes)
-        if (StringUtils.isEmpty(getPassword())) {
-            out.write(0x00);
-        } else {
-            try {
-                byte[] encryptedPassword = MySQLPasswordEncrypter.scrambleCachingSha2(getPassword().getBytes(),
-                    getScrumbleBuff());
-                ByteHelper.writeBinaryCodedLengthBytes(encryptedPassword, out);
-            } catch (Exception e) {
-                throw new IOException("can't encrypt password that will be sent to MySQL server.", e);
-            }
-        }
-        // 7 . (Null-Terminated String) databasename (optional)
-        if (getDatabaseName() != null) {
-            ByteHelper.writeNullTerminatedString(getDatabaseName(), out);
-        }
-        // 8 . (Null-Terminated String) auth plugin name (optional)
-        if (getAuthPluginName() != null) {
-            ByteHelper.writeNullTerminated(getAuthPluginName(), out);
-        }
-        // end write
-        return out.toByteArray();
-    }
-
-}

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

@@ -1,41 +0,0 @@
-package com.alibaba.otter.canal.parse.driver.mysql.packets.client;
-
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-
-import com.alibaba.otter.canal.parse.driver.mysql.packets.Capability;
-import com.alibaba.otter.canal.parse.driver.mysql.packets.IPacket;
-import com.alibaba.otter.canal.parse.driver.mysql.utils.ByteHelper;
-
-/**
- * @author 枭博
- * @date 2024/05/14
- */
-public class SslRequestCommandPacket implements IPacket {
-
-    private final int serverCharsetNumber;
-
-    public SslRequestCommandPacket(int serverCharsetNumber){
-        this.serverCharsetNumber = serverCharsetNumber;
-    }
-
-    @Override
-    public void fromBytes(byte[] data) throws IOException {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public byte[] toBytes() throws IOException {
-        ByteArrayOutputStream out = new ByteArrayOutputStream();
-        int clientCapabilities = Capability.CLIENT_LONG_FLAG | Capability.CLIENT_PROTOCOL_41
-                                 | Capability.CLIENT_SECURE_CONNECTION | Capability.CLIENT_PLUGIN_AUTH
-                                 | Capability.CLIENT_SSL;
-        ByteHelper.writeUnsignedIntLittleEndian(clientCapabilities, out);
-        ByteHelper.writeUnsignedIntLittleEndian(0, out);
-        out.write(serverCharsetNumber);
-        for (int i = 0; i < 23; i++) {
-            out.write(0);
-        }
-        return out.toByteArray();
-    }
-}

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

@@ -10,8 +10,6 @@ import java.net.SocketException;
 import java.net.SocketTimeoutException;
 import java.nio.channels.ClosedByInterruptException;
 
-import javax.net.ssl.SSLSocket;
-
 /**
  * 使用BIO进行dump
  *
@@ -24,13 +22,11 @@ public class BioSocketChannel implements SocketChannel {
     private Socket       socket;
     private InputStream  input;
     private OutputStream output;
-    private final boolean ssl;
 
     BioSocketChannel(Socket socket) throws IOException{
         this.socket = socket;
         this.input = new BufferedInputStream(socket.getInputStream(), 16384);
         this.output = socket.getOutputStream();
-        this.ssl = (socket instanceof SSLSocket);
     }
 
     public void write(byte[]... buf) throws IOException {
@@ -92,9 +88,9 @@ public class BioSocketChannel implements SocketChannel {
             }
         }
         if (remain > 0 && accTimeout >= timeout) {
-            throw new SocketTimeoutException(
+            throw new SocketTimeoutException("Timeout occurred, failed to read total " + readSize + " bytes in "
-                "Timeout occurred, failed to read total " + readSize + " bytes in " + timeout
+                                             + timeout + " milliseconds, actual read only " + (readSize - remain)
-                                             + " milliseconds, actual read only " + (readSize - remain) + " bytes");
+                                             + " bytes");
         }
         return data;
     }
@@ -138,14 +134,6 @@ public class BioSocketChannel implements SocketChannel {
         return false;
     }
 
-    public boolean isSsl() {
-        return ssl;
-    }
-
-    public Socket getSocket() {
-        return socket;
-    }
-
     public SocketAddress getRemoteSocketAddress() {
         Socket socket = this.socket;
         if (socket != null) {
@@ -167,6 +155,16 @@ public class BioSocketChannel implements SocketChannel {
     public void close() {
         Socket socket = this.socket;
         if (socket != null) {
+            try {
+                socket.shutdownInput();
+            } catch (IOException e) {
+                // Ignore, could not do anymore
+            }
+            try {
+                socket.shutdownOutput();
+            } catch (IOException e) {
+                // Ignore, could not do anymore
+            }
             try {
                 socket.close();
             } catch (IOException e) {

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

@@ -1,28 +1,7 @@
 package com.alibaba.otter.canal.parse.driver.mysql.socket;
 
-import java.io.IOException;
-import java.io.InputStream;
 import java.net.Socket;
 import java.net.SocketAddress;
-import java.net.URL;
-import java.security.InvalidAlgorithmParameterException;
-import java.security.KeyStore;
-import java.security.cert.*;
-import java.util.*;
-import java.util.stream.Collectors;
-
-import javax.naming.InvalidNameException;
-import javax.naming.ldap.LdapName;
-import javax.naming.ldap.Rdn;
-import javax.net.ssl.*;
-import javax.security.auth.x500.X500Principal;
-
-import org.apache.commons.lang.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo;
-import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslMode;
 
 /**
  * @author luoyaogui 实现channel的管理（监听连接、读数据、回收） 2016-12-28
@@ -30,295 +9,14 @@ import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslMode;
  */
 public abstract class BioSocketChannelPool {
 
-    private static final Logger logger = LoggerFactory.getLogger(BioSocketChannelPool.class);
-
     public static BioSocketChannel open(SocketAddress address) throws Exception {
-        Socket socket = createSocket(address);
+        Socket socket = new Socket();
-        return new BioSocketChannel(socket);
-    }
-
-    public static BioSocketChannel openSsl(Socket socket, SslInfo sslInfo) throws Exception {
-        SslMode sslMode = sslInfo.getSslMode();
-        switch (sslMode) {
-            case REQUIRED:
-            case PREFERRED:
-            case VERIFY_CA:
-            case VERIFY_IDENTITY:
-                SSLSocket sslSocket = createSslSocket(socket, sslInfo);
-                return new BioSocketChannel(sslSocket);
-            default:
-                throw new UnsupportedOperationException("Unsupported ssl mode: " + sslMode);
-        }
-    }
-
-    private static Socket createSocket(SocketAddress address) throws IOException {
-        Socket socket;
-        socket = new Socket();
         socket.setSoTimeout(BioSocketChannel.SO_TIMEOUT);
         socket.setTcpNoDelay(true);
         socket.setKeepAlive(true);
         socket.setReuseAddress(true);
         socket.connect(address, BioSocketChannel.DEFAULT_CONNECT_TIMEOUT);
-        return socket;
+        return new BioSocketChannel(socket);
-    }
-
-    /**
-     * from JDBC driver com.mysql.cj.protocol.ExportControlled#performTlsHandshake
-     * com.mysql.cj.protocol.ExportControlled#getSSLContext
-     *
-     * @param socket
-     * @param sslInfo
-     * @return
-     * @throws Exception
-     */
-    private static SSLSocket createSslSocket(Socket socket, SslInfo sslInfo) throws Exception {
-        SslMode sslMode = sslInfo.getSslMode();
-        boolean verifyServerCert = sslMode == SslMode.VERIFY_CA || sslMode == SslMode.VERIFY_IDENTITY;
-
-        String clientCertificateKeyStoreUrl = sslInfo.getClientCertificateKeyStoreUrl();
-        String clientCertificateKeyStoreType = sslInfo.getClientCertificateKeyStoreType() != null ? sslInfo
-            .getClientCertificateKeyStoreType() : "JKS";
-        String clientCertificateKeyStorePassword = sslInfo.getClientCertificateKeyStorePassword();
-        String trustCertificateKeyStoreUrl = sslInfo.getTrustCertificateKeyStoreUrl();
-        String trustCertificateKeyStoreType = sslInfo.getTrustCertificateKeyStoreType() != null ? sslInfo
-            .getTrustCertificateKeyStoreType() : "JKS";
-        String trustCertificateKeyStorePassword = sslInfo.getTrustCertificateKeyStorePassword();
-        boolean fallbackToDefaultTrustStore = true;
-        String hostName = sslMode == SslMode.VERIFY_IDENTITY ? socket.getInetAddress().getHostName() : null;
-
-        SSLContext sslContext = getSSLContext(clientCertificateKeyStoreUrl,
-            clientCertificateKeyStoreType,
-            clientCertificateKeyStorePassword,
-            trustCertificateKeyStoreUrl,
-            trustCertificateKeyStoreType,
-            trustCertificateKeyStorePassword,
-            fallbackToDefaultTrustStore,
-            verifyServerCert,
-            hostName);
-        SSLSocketFactory socketFactory = sslContext.getSocketFactory();
-
-        SSLSocket sslSocket = (SSLSocket) socketFactory
-            .createSocket(socket, socket.getInetAddress().getHostName(), socket.getPort(), true);
-
-        String[] protocolArr = null;
-        if (StringUtils.isNotEmpty(sslInfo.getTlsVersions())) {
-            protocolArr = StringUtils.split(sslInfo.getTlsVersions(), ",");
-        }
-        if (protocolArr == null || protocolArr.length == 0) {
-            protocolArr = new String[] { "TLSv1.2", "TLSv1.3" };
-        }
-        logger.info("SSL protocol: {}", StringUtils.join(protocolArr, ","));
-        sslSocket.setEnabledProtocols(protocolArr);
-        sslSocket.setSoTimeout(BioSocketChannel.SO_TIMEOUT);
-        sslSocket.startHandshake();
-        logger.info("SSL socket handshake success.");
-        return sslSocket;
-    }
-
-    private static SSLContext getSSLContext(String clientCertificateKeyStoreUrl, String clientCertificateKeyStoreType,
-                                            String clientCertificateKeyStorePassword,
-                                            String trustCertificateKeyStoreUrl, String trustCertificateKeyStoreType,
-                                            String trustCertificateKeyStorePassword,
-                                            boolean fallbackToDefaultTrustStore, boolean verifyServerCert,
-                                            String hostName) throws Exception {
-
-        KeyManager[] kms = null;
-        List<TrustManager> tms = new ArrayList<>();
-
-        TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
-        KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
-
-        if (StringUtils.isNotEmpty(clientCertificateKeyStoreUrl)) {
-            InputStream ksIS = null;
-            try {
-                if (StringUtils.isNotEmpty(clientCertificateKeyStoreType)) {
-                    KeyStore clientKeyStore = KeyStore.getInstance(clientCertificateKeyStoreType);
-                    URL ksURL = new URL(clientCertificateKeyStoreUrl);
-                    char[] password = (clientCertificateKeyStorePassword == null) ? new char[0] : clientCertificateKeyStorePassword
-                        .toCharArray();
-                    ksIS = ksURL.openStream();
-                    clientKeyStore.load(ksIS, password);
-                    kmf.init(clientKeyStore, password);
-                    kms = kmf.getKeyManagers();
-                }
-            } finally {
-                if (ksIS != null) {
-                    try {
-                        ksIS.close();
-                    } catch (IOException e) {
-                        // can't close input stream, but keystore can be properly initialized so we
-                        // shouldn't throw this exception
-                    }
-                }
-            }
-        }
-
-        InputStream trustStoreIS = null;
-        try {
-            String trustStoreType = "";
-            char[] trustStorePassword = null;
-            KeyStore trustKeyStore = null;
-
-            if (StringUtils.isNotEmpty(trustCertificateKeyStoreUrl)
-                && StringUtils.isNotEmpty(trustCertificateKeyStoreType)) {
-                trustStoreType = trustCertificateKeyStoreType;
-                trustStorePassword = (trustCertificateKeyStorePassword == null) ? new char[0] : trustCertificateKeyStorePassword
-                    .toCharArray();
-                trustStoreIS = new URL(trustCertificateKeyStoreUrl).openStream();
-
-                trustKeyStore = KeyStore.getInstance(trustStoreType);
-                trustKeyStore.load(trustStoreIS, trustStorePassword);
-            }
-
-            if (trustKeyStore != null || fallbackToDefaultTrustStore) {
-                tmf.init(trustKeyStore);
-                // (trustKeyStore == null) initializes the TrustManagerFactory with the default
-                // truststore.
-
-                // building the customized list of TrustManagers from original one if it's
-                // available
-                TrustManager[] origTms = tmf.getTrustManagers();
-
-                for (TrustManager tm : origTms) {
-                    // wrap X509TrustManager or put original if non-X509 TrustManager
-                    tms.add(tm instanceof X509TrustManager ? new X509TrustManagerWrapper((X509TrustManager) tm,
-                        verifyServerCert,
-                        hostName) : tm);
-                }
-            }
-
-        } finally {
-            if (trustStoreIS != null) {
-                try {
-                    trustStoreIS.close();
-                } catch (IOException e) {
-                    // can't close input stream, but keystore can be properly initialized so we
-                    // shouldn't throw this exception
-                }
-            }
-        }
-
-        // if original TrustManagers are not available then putting one
-        // X509TrustManagerWrapper which take care only
-        // about expiration check
-        if (tms.size() == 0) {
-            tms.add(new X509TrustManagerWrapper(verifyServerCert, hostName));
-        }
-
-        SSLContext sslContext = SSLContext.getInstance("TLS");
-        sslContext.init(kms, tms.toArray(new TrustManager[tms.size()]), null);
-        return sslContext;
-    }
-
-    public static class X509TrustManagerWrapper implements X509TrustManager {
-
-        private X509TrustManager   origTm           = null;
-        private boolean            verifyServerCert = false;
-        private String             hostName         = null;
-        private CertificateFactory certFactory      = null;
-        private PKIXParameters     validatorParams  = null;
-        private CertPathValidator  validator        = null;
-
-        public X509TrustManagerWrapper(X509TrustManager tm, boolean verifyServerCertificate,
-                                       String hostName) throws CertificateException{
-            this.origTm = tm;
-            this.verifyServerCert = verifyServerCertificate;
-            this.hostName = hostName;
-
-            if (verifyServerCertificate) {
-                try {
-                    Set<TrustAnchor> anch = Arrays.stream(tm.getAcceptedIssuers())
-                        .map(c -> new TrustAnchor(c, null))
-                        .collect(Collectors.toSet());
-                    this.validatorParams = new PKIXParameters(anch);
-                    this.validatorParams.setRevocationEnabled(false);
-                    this.validator = CertPathValidator.getInstance("PKIX");
-                    this.certFactory = CertificateFactory.getInstance("X.509");
-                } catch (Exception e) {
-                    throw new CertificateException(e);
-                }
-            }
-
-        }
-
-        public X509TrustManagerWrapper(boolean verifyServerCertificate, String hostName){
-            this.verifyServerCert = verifyServerCertificate;
-            this.hostName = hostName;
-        }
-
-        public X509Certificate[] getAcceptedIssuers() {
-            return this.origTm != null ? this.origTm.getAcceptedIssuers() : new X509Certificate[0];
-        }
-
-        public void checkServerTrusted(X509Certificate[] chain, String authType) throws CertificateException {
-            for (int i = 0; i < chain.length; i++) {
-                chain[i].checkValidity();
-            }
-
-            if (this.validatorParams != null) {
-                X509CertSelector certSelect = new X509CertSelector();
-                certSelect.setSerialNumber(chain[0].getSerialNumber());
-
-                try {
-                    CertPath certPath = this.certFactory.generateCertPath(Arrays.asList(chain));
-                    // Validate against truststore
-                    CertPathValidatorResult result = this.validator.validate(certPath, this.validatorParams);
-                    // Check expiration for the CA used to validate this path
-                    ((PKIXCertPathValidatorResult) result).getTrustAnchor().getTrustedCert().checkValidity();
-
-                } catch (InvalidAlgorithmParameterException e) {
-                    throw new CertificateException(e);
-                } catch (CertPathValidatorException e) {
-                    throw new CertificateException(e);
-                }
-            }
-
-            if (this.verifyServerCert) {
-                if (this.origTm != null) {
-                    this.origTm.checkServerTrusted(chain, authType);
-                } else {
-                    throw new CertificateException(
-                        "Can't verify server certificate because no trust manager is found.");
-                }
-
-                // verify server certificate identity
-                if (this.hostName != null) {
-                    logger.info("verify hostName: {}", this.hostName);
-                    Set<String> expectHostNames = new HashSet<>();
-                    for (X509Certificate certificate : chain) {
-                        String dn = certificate.getSubjectX500Principal().getName(X500Principal.RFC2253);
-                        String cn = null;
-                        try {
-                            LdapName ldapDN = new LdapName(dn);
-                            for (Rdn rdn : ldapDN.getRdns()) {
-                                if (rdn.getType().equalsIgnoreCase("CN")) {
-                                    cn = rdn.getValue().toString();
-                                    break;
-                                }
-                            }
-                        } catch (InvalidNameException e) {
-                            throw new CertificateException(
-                                "Failed to retrieve the Common Name (CN) from the server certificate.");
-                        }
-                        expectHostNames.add(cn);
-                    }
-
-                    if (!expectHostNames.contains(this.hostName)) {
-                        throw new CertificateException(
-                            "Server certificate identity check failed. The certificate Common Name "
-                                                       + expectHostNames.stream()
-                                                           .map(h -> "'" + h + "'")
-                                                           .collect(Collectors.joining(", "))
-                                                       + " does not match with '" + this.hostName + "'.");
-                    }
-
-                }
-            }
-        }
-
-        public void checkClientTrusted(X509Certificate[] chain, String authType) throws CertificateException {
-            this.origTm.checkClientTrusted(chain, authType);
-        }
     }
 
 }

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

@@ -1,14 +1,8 @@
 package com.alibaba.otter.canal.parse.driver.mysql.socket;
 
-import java.io.IOException;
 import java.net.SocketAddress;
 
 import org.apache.commons.lang.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo;
-import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslMode;
 
 /**
  * @author agapple 2018年3月12日 下午10:46:22
@@ -16,8 +10,6 @@ import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslMode;
  */
 public abstract class SocketChannelPool {
 
-    private static final Logger logger = LoggerFactory.getLogger(SocketChannelPool.class);
-
     public static SocketChannel open(SocketAddress address) throws Exception {
         String type = chooseSocketChannel();
         if ("netty".equalsIgnoreCase(type)) {
@@ -25,32 +17,7 @@ public abstract class SocketChannelPool {
         } else {
             return BioSocketChannelPool.open(address);
         }
-    }
 
-    public static SocketChannel connectSsl(SocketChannel channel, SslInfo sslInfo) throws IOException {
-        SslMode sslMode = sslInfo.getSslMode();
-        String type = chooseSocketChannel();
-        if ("netty".equalsIgnoreCase(type)) {
-            throw new UnsupportedOperationException("canal socketChannel netty not support ssl mode: " + sslMode);
-        } else {
-            SocketAddress remoteSocketAddress = channel.getRemoteSocketAddress();
-            try {
-                return BioSocketChannelPool.openSsl(((BioSocketChannel) channel).getSocket(), sslInfo);
-            } catch (Exception e) {
-                if (sslMode == SslMode.PREFERRED) {
-                    // still use non ssl channel
-                    logger.info("{} still use non SSL channel due to SSL connect failed.", remoteSocketAddress, e);
-                    return channel;
-                }
-                IOException ioe;
-                if (e instanceof IOException) {
-                    ioe = (IOException) e;
-                } else {
-                    ioe = new IOException(e);
-                }
-                throw ioe;
-            }
-        }
     }
 
     private static String chooseSocketChannel() {

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

@@ -1,98 +0,0 @@
-package com.alibaba.otter.canal.parse.driver.mysql.ssl;
-
-/**
- * @author 枭博
- * @date 2024/05/14
- */
-public class SslInfo {
-
-    private SslMode sslMode = SslMode.DISABLED;
-    private String  tlsVersions;                       // 和 enabledTLSProtocols 同含义，TLSv1.2,TLSv1.3
-    private String  trustCertificateKeyStoreType;      // trustStore 证书类型，支持 JKS (默认) 和 PKCS12
-    private String  trustCertificateKeyStoreUrl;       // trustStore 证书
-    private String  trustCertificateKeyStorePassword;  // trustStore 证书密码
-    private String  clientCertificateKeyStoreType;     // client 证书类型，支持 JKS (默认) 和 PKCS12
-    private String  clientCertificateKeyStoreUrl;      // client 证书
-    private String  clientCertificateKeyStorePassword; // client 证书密码
-
-    public SslInfo(SslMode sslMode, String tlsVersions, String trustCertificateKeyStoreType,
-                   String trustCertificateKeyStoreUrl, String trustCertificateKeyStorePassword,
-                   String clientCertificateKeyStoreType, String clientCertificateKeyStoreUrl,
-                   String clientCertificateKeyStorePassword) {
-        this.sslMode = sslMode;
-        this.tlsVersions = tlsVersions;
-        this.trustCertificateKeyStoreType = trustCertificateKeyStoreType;
-        this.trustCertificateKeyStoreUrl = trustCertificateKeyStoreUrl;
-        this.trustCertificateKeyStorePassword = trustCertificateKeyStorePassword;
-        this.clientCertificateKeyStoreType = clientCertificateKeyStoreType;
-        this.clientCertificateKeyStoreUrl = clientCertificateKeyStoreUrl;
-        this.clientCertificateKeyStorePassword = clientCertificateKeyStorePassword;
-    }
-
-    public SslInfo() {
-    }
-
-    public SslMode getSslMode() {
-        return sslMode;
-    }
-
-    public void setSslMode(SslMode sslMode) {
-        this.sslMode = sslMode;
-    }
-
-    public String getTlsVersions() {
-        return tlsVersions;
-    }
-
-    public void setTlsVersions(String tlsVersions) {
-        this.tlsVersions = tlsVersions;
-    }
-
-    public String getTrustCertificateKeyStoreType() {
-        return trustCertificateKeyStoreType;
-    }
-
-    public void setTrustCertificateKeyStoreType(String trustCertificateKeyStoreType) {
-        this.trustCertificateKeyStoreType = trustCertificateKeyStoreType;
-    }
-
-    public String getTrustCertificateKeyStoreUrl() {
-        return trustCertificateKeyStoreUrl;
-    }
-
-    public void setTrustCertificateKeyStoreUrl(String trustCertificateKeyStoreUrl) {
-        this.trustCertificateKeyStoreUrl = trustCertificateKeyStoreUrl;
-    }
-
-    public String getTrustCertificateKeyStorePassword() {
-        return trustCertificateKeyStorePassword;
-    }
-
-    public void setTrustCertificateKeyStorePassword(String trustCertificateKeyStorePassword) {
-        this.trustCertificateKeyStorePassword = trustCertificateKeyStorePassword;
-    }
-
-    public String getClientCertificateKeyStoreType() {
-        return clientCertificateKeyStoreType;
-    }
-
-    public void setClientCertificateKeyStoreType(String clientCertificateKeyStoreType) {
-        this.clientCertificateKeyStoreType = clientCertificateKeyStoreType;
-    }
-
-    public String getClientCertificateKeyStoreUrl() {
-        return clientCertificateKeyStoreUrl;
-    }
-
-    public void setClientCertificateKeyStoreUrl(String clientCertificateKeyStoreUrl) {
-        this.clientCertificateKeyStoreUrl = clientCertificateKeyStoreUrl;
-    }
-
-    public String getClientCertificateKeyStorePassword() {
-        return clientCertificateKeyStorePassword;
-    }
-
-    public void setClientCertificateKeyStorePassword(String clientCertificateKeyStorePassword) {
-        this.clientCertificateKeyStorePassword = clientCertificateKeyStorePassword;
-    }
-}

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

@@ -1,26 +0,0 @@
-package com.alibaba.otter.canal.parse.driver.mysql.ssl;
-
-public enum SslMode {
-
-    /**
-     * 关闭SSL
-     */
-    DISABLED,
-    /**
-     * 尝试SSL传输
-     */
-    PREFERRED,
-    /**
-     * 要求SSL传输，不校验证书
-     */
-    REQUIRED,
-    /**
-     * 要求SSL传输，校验证书，不校验证书里的域名
-     */
-    VERIFY_CA,
-    /**
-     * 要求SSL传输，校验证书，校验证书里的域名
-     */
-    VERIFY_IDENTITY
-
-}

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

@@ -276,7 +276,8 @@ public class CanalInstanceWithManager extends AbstractCanalInstance {
             mysqlEventParser.setDestination(destination);
             // 编码参数
             mysqlEventParser.setConnectionCharset(parameters.getConnectionCharset());
-            // 网络相关参数1
+            mysqlEventParser.setConnectionCharsetNumber(parameters.getConnectionCharsetNumber());
+            // 网络相关参数
             mysqlEventParser.setDefaultConnectionTimeoutInSeconds(parameters.getDefaultConnectionTimeoutInSeconds());
             mysqlEventParser.setSendBufferSize(parameters.getSendBufferSize());
             mysqlEventParser.setReceiveBufferSize(parameters.getReceiveBufferSize());
@@ -290,15 +291,13 @@ public class CanalInstanceWithManager extends AbstractCanalInstance {
                 mysqlEventParser.setMasterInfo(new AuthenticationInfo(dbAddresses.get(0),
                     parameters.getDbUsername(),
                     parameters.getDbPassword(),
-                    parameters.getDefaultDatabaseName(),
+                    parameters.getDefaultDatabaseName()));
-                    parameters.getSslInfo()));
 
                 if (dbAddresses.size() > 1) {
                     mysqlEventParser.setStandbyInfo(new AuthenticationInfo(dbAddresses.get(1),
                         parameters.getDbUsername(),
                         parameters.getDbPassword(),
-                        parameters.getDefaultDatabaseName(),
+                        parameters.getDefaultDatabaseName()));
-                        parameters.getSslInfo()));
                 }
             }
 
@@ -370,6 +369,7 @@ public class CanalInstanceWithManager extends AbstractCanalInstance {
             localBinlogEventParser.setDestination(destination);
             localBinlogEventParser.setBufferSize(parameters.getReceiveBufferSize());
             localBinlogEventParser.setConnectionCharset(parameters.getConnectionCharset());
+            localBinlogEventParser.setConnectionCharsetNumber(parameters.getConnectionCharsetNumber());
             localBinlogEventParser.setDirectory(parameters.getLocalBinlogDirectory());
             localBinlogEventParser.setProfilingEnabled(false);
             localBinlogEventParser.setDetectingEnable(parameters.getDetectingEnable());
@@ -381,8 +381,7 @@ public class CanalInstanceWithManager extends AbstractCanalInstance {
                 localBinlogEventParser.setMasterInfo(new AuthenticationInfo(dbAddresses.get(0),
                     parameters.getDbUsername(),
                     parameters.getDbPassword(),
-                    parameters.getDefaultDatabaseName(),
+                    parameters.getDefaultDatabaseName()));
-                    parameters.getSslInfo()));
             }
 
             eventParser = localBinlogEventParser;

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

@@ -2,7 +2,6 @@ package com.alibaba.otter.canal.instance.manager;
 
 import java.util.Properties;
 
-import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.BeanFactory;
@@ -70,9 +69,6 @@ public class PlainCanalInstanceGenerator implements CanalInstanceGenerator {
     // ================ setter / getter ================
 
     private BeanFactory getBeanFactory(String springXml) {
-        if (!StringUtils.startsWithIgnoreCase(springXml, "classpath:")) {
-            springXml = "classpath:" + springXml;
-        }
         ApplicationContext applicationContext = new ClassPathXmlApplicationContext(springXml);
         return applicationContext;
     }

instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/model/CanalParameter.java

@@ -9,8 +9,6 @@ import org.apache.commons.lang.StringUtils;
 import org.apache.commons.lang.builder.ToStringBuilder;
 
 import com.alibaba.otter.canal.common.utils.CanalToStringStyle;
-import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo;
-import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslMode;
 
 /**
  * canal运行相关参数
@@ -58,6 +56,8 @@ public class CanalParameter implements Serializable {
     private Integer                  defaultConnectionTimeoutInSeconds  = 30;                        // sotimeout
     private Integer                  receiveBufferSize                  = 64 * 1024;
     private Integer                  sendBufferSize                     = 64 * 1024;
+    // 编码信息
+    private Byte                     connectionCharsetNumber            = (byte) 33;
     private String                   connectionCharset                  = "UTF-8";
 
     // 数据库信息
@@ -66,21 +66,6 @@ public class CanalParameter implements Serializable {
     private String                   dbUsername;                                                     // 数据库用户
     private String                   dbPassword;                                                     // 数据库密码
 
-    private String                   sslMode                            = SslMode.DISABLED.name();
-    private String                   tlsVersions;                                                    // 和
-                                                                                                     // enabledTLSProtocols
-                                                                                                     // 同含义，TLSv1.2,TLSv1.3
-    private String                   trustCertificateKeyStoreType;                                   // trustStore
-                                                                                                     // 证书类型，支持 JKS (默认)
-                                                                                                     // 和 PKCS12
-    private String                   trustCertificateKeyStoreUrl;                                    // trustStore 证书路径
-    private String                   trustCertificateKeyStorePassword;                               // trustStore 证书密码
-    private String                   clientCertificateKeyStoreType;                                  // client 证书类型，支持
-                                                                                                     // JKS (默认) 和
-                                                                                                     // PKCS12
-    private String                   clientCertificateKeyStoreUrl;                                   // client 证书路径
-    private String                   clientCertificateKeyStorePassword;                              // client 证书密码
-
     // binlog链接信息
     private IndexMode                indexMode;
     private List<String>             positions;                                                      // 数据库positions信息
@@ -123,39 +108,10 @@ public class CanalParameter implements Serializable {
     private InetSocketAddress        masterAddress;                                                  // 主库信息
     private String                   masterUsername;                                                 // 帐号
     private String                   masterPassword;                                                 // 密码
-    private String                   masterSslMode                      = SslMode.DISABLED.name();
-    private String                   masterTlsVersions;                                              // 和
-                                                                                                     // enabledTLSProtocols
-                                                                                                     // 同含义，TLSv1.2,TLSv1.3
-    private String                   masterTrustCertificateKeyStoreType;                             // trustStore
-                                                                                                     // 证书类型，支持 JKS (默认)
-                                                                                                     // 和 PKCS12
-    private String                   masterTrustCertificateKeyStoreUrl;                              // trustStore 证书路径
-    private String                   masterTrustCertificateKeyStorePassword;                         // trustStore 证书密码
-    private String                   masterClientCertificateKeyStoreType;                            // client 证书类型，支持
-                                                                                                     // JKS (默认) 和
-                                                                                                     // PKCS12
-    private String                   masterClientCertificateKeyStoreUrl;                             // client 证书路径
-    private String                   masterClientCertificateKeyStorePassword;                        // client 证书密码
 
     private InetSocketAddress        standbyAddress;                                                 // 备库信息
     private String                   standbyUsername;                                                // 帐号
     private String                   standbyPassword;
-    private String                   standbySslMode                     = SslMode.DISABLED.name();
-    private String                   standbyTlsVersions;                                             // 和
-                                                                                                     // enabledTLSProtocols
-                                                                                                     // 同含义，TLSv1.2,TLSv1.3
-    private String                   standbyTrustCertificateKeyStoreType;                            // trustStore
-                                                                                                     // 证书类型，支持 JKS (默认)
-                                                                                                     // 和 PKCS12
-    private String                   standbyTrustCertificateKeyStoreUrl;                             // trustStore 证书路径
-    private String                   standbyTrustCertificateKeyStorePassword;                        // trustStore 证书密码
-    private String                   standbyClientCertificateKeyStoreType;                           // client 证书类型，支持
-                                                                                                     // JKS (默认) 和
-                                                                                                     // PKCS12
-    private String                   standbyClientCertificateKeyStoreUrl;                            // client 证书路径
-    private String                   standbyClientCertificateKeyStorePassword;                       // client 证书密码
-
     private String                   masterLogfileName                  = null;                      // master起始位置
     private Long                     masterLogfileOffest                = null;
     private Long                     masterTimestamp                    = null;
@@ -164,11 +120,11 @@ public class CanalParameter implements Serializable {
     private Long                     standbyTimestamp                   = null;
     private Boolean                  parallel                           = Boolean.FALSE;
 
-    // 自定义alarmHandler类全路径
+    //自定义alarmHandler类全路径
     private String                   alarmHandlerClass                  = null;
-    // 自定义alarmHandler插件文件夹路径
+    //自定义alarmHandler插件文件夹路径
     private String                   alarmHandlerPluginDir              = null;
-    // 是否支持多流消费
+    //是否支持多流消费
     private Boolean                  multiStreamEnable                  = Boolean.FALSE;
 
     public static enum RunMode {
@@ -227,7 +183,6 @@ public class CanalParameter implements Serializable {
     }
 
     public static enum StorageMode {
-
         /** 内存存储模式 */
         MEMORY,
         /** 文件存储模式 */
@@ -250,7 +205,6 @@ public class CanalParameter implements Serializable {
     }
 
     public static enum StorageScavengeMode {
-
         /** 在存储满的时候触发 */
         ON_FULL,
         /** 在每次有ack请求时触发 */
@@ -278,7 +232,6 @@ public class CanalParameter implements Serializable {
     }
 
     public static enum SourcingType {
-
         /** mysql DB */
         MYSQL,
         /** localBinLog */
@@ -306,7 +259,6 @@ public class CanalParameter implements Serializable {
     }
 
     public static enum MetaMode {
-
         /** 内存存储模式 */
         MEMORY,
         /** 文件存储模式 */
@@ -334,7 +286,6 @@ public class CanalParameter implements Serializable {
     }
 
     public static enum IndexMode {
-
         /** 内存存储模式 */
         MEMORY,
         /** 文件存储模式 */
@@ -368,7 +319,6 @@ public class CanalParameter implements Serializable {
     }
 
     public static enum BatchMode {
-
         /** 对象数量 */
         ITEMSIZE,
 
@@ -583,6 +533,14 @@ public class CanalParameter implements Serializable {
         this.sendBufferSize = sendBufferSize;
     }
 
+    public Byte getConnectionCharsetNumber() {
+        return connectionCharsetNumber;
+    }
+
+    public void setConnectionCharsetNumber(Byte connectionCharsetNumber) {
+        this.connectionCharsetNumber = connectionCharsetNumber;
+    }
+
     public String getConnectionCharset() {
         return connectionCharset;
     }
@@ -773,134 +731,6 @@ public class CanalParameter implements Serializable {
         this.dbPassword = dbPassword;
     }
 
-    public SslInfo getSslInfo() {
-        if (dbUsername == null) {
-            if (masterUsername != null) {
-                return new SslInfo(SslMode.valueOf(masterSslMode),
-                    masterTlsVersions,
-                    masterTrustCertificateKeyStoreType,
-                    masterTrustCertificateKeyStoreUrl,
-                    masterTrustCertificateKeyStorePassword,
-                    masterClientCertificateKeyStoreType,
-                    masterClientCertificateKeyStoreUrl,
-                    masterClientCertificateKeyStorePassword);
-            } else {
-                return new SslInfo(SslMode.valueOf(standbySslMode),
-                    standbyTlsVersions,
-                    standbyTrustCertificateKeyStoreType,
-                    standbyTrustCertificateKeyStoreUrl,
-                    standbyTrustCertificateKeyStorePassword,
-                    standbyClientCertificateKeyStoreType,
-                    standbyClientCertificateKeyStoreUrl,
-                    standbyClientCertificateKeyStorePassword);
-            }
-        }
-        return new SslInfo(SslMode.valueOf(sslMode),
-            tlsVersions,
-            trustCertificateKeyStoreType,
-            trustCertificateKeyStoreUrl,
-            trustCertificateKeyStorePassword,
-            clientCertificateKeyStoreType,
-            clientCertificateKeyStoreUrl,
-            clientCertificateKeyStorePassword);
-    }
-
-    public void setSslMode(String sslMode) {
-        this.sslMode = sslMode;
-    }
-
-    public void setTlsVersions(String tlsVersions) {
-        this.tlsVersions = tlsVersions;
-    }
-
-    public void setTrustCertificateKeyStoreType(String trustCertificateKeyStoreType) {
-        this.trustCertificateKeyStoreType = trustCertificateKeyStoreType;
-    }
-
-    public void setTrustCertificateKeyStoreUrl(String trustCertificateKeyStoreUrl) {
-        this.trustCertificateKeyStoreUrl = trustCertificateKeyStoreUrl;
-    }
-
-    public void setTrustCertificateKeyStorePassword(String trustCertificateKeyStorePassword) {
-        this.trustCertificateKeyStorePassword = trustCertificateKeyStorePassword;
-    }
-
-    public void setClientCertificateKeyStoreType(String clientCertificateKeyStoreType) {
-        this.clientCertificateKeyStoreType = clientCertificateKeyStoreType;
-    }
-
-    public void setClientCertificateKeyStoreUrl(String clientCertificateKeyStoreUrl) {
-        this.clientCertificateKeyStoreUrl = clientCertificateKeyStoreUrl;
-    }
-
-    public void setClientCertificateKeyStorePassword(String clientCertificateKeyStorePassword) {
-        this.clientCertificateKeyStorePassword = clientCertificateKeyStorePassword;
-    }
-
-    public void setMasterSslMode(String masterSslMode) {
-        this.masterSslMode = masterSslMode;
-    }
-
-    public void setMasterTlsVersions(String masterTlsVersions) {
-        this.masterTlsVersions = masterTlsVersions;
-    }
-
-    public void setMasterTrustCertificateKeyStoreType(String masterTrustCertificateKeyStoreType) {
-        this.masterTrustCertificateKeyStoreType = masterTrustCertificateKeyStoreType;
-    }
-
-    public void setMasterTrustCertificateKeyStoreUrl(String masterTrustCertificateKeyStoreUrl) {
-        this.masterTrustCertificateKeyStoreUrl = masterTrustCertificateKeyStoreUrl;
-    }
-
-    public void setMasterTrustCertificateKeyStorePassword(String masterTrustCertificateKeyStorePassword) {
-        this.masterTrustCertificateKeyStorePassword = masterTrustCertificateKeyStorePassword;
-    }
-
-    public void setMasterClientCertificateKeyStoreType(String masterClientCertificateKeyStoreType) {
-        this.masterClientCertificateKeyStoreType = masterClientCertificateKeyStoreType;
-    }
-
-    public void setMasterClientCertificateKeyStoreUrl(String masterClientCertificateKeyStoreUrl) {
-        this.masterClientCertificateKeyStoreUrl = masterClientCertificateKeyStoreUrl;
-    }
-
-    public void setMasterClientCertificateKeyStorePassword(String masterClientCertificateKeyStorePassword) {
-        this.masterClientCertificateKeyStorePassword = masterClientCertificateKeyStorePassword;
-    }
-
-    public void setStandbySslMode(String standbySslMode) {
-        this.standbySslMode = standbySslMode;
-    }
-
-    public void setStandbyTlsVersions(String standbyTlsVersions) {
-        this.standbyTlsVersions = standbyTlsVersions;
-    }
-
-    public void setStandbyTrustCertificateKeyStoreType(String standbyTrustCertificateKeyStoreType) {
-        this.standbyTrustCertificateKeyStoreType = standbyTrustCertificateKeyStoreType;
-    }
-
-    public void setStandbyTrustCertificateKeyStoreUrl(String standbyTrustCertificateKeyStoreUrl) {
-        this.standbyTrustCertificateKeyStoreUrl = standbyTrustCertificateKeyStoreUrl;
-    }
-
-    public void setStandbyTrustCertificateKeyStorePassword(String standbyTrustCertificateKeyStorePassword) {
-        this.standbyTrustCertificateKeyStorePassword = standbyTrustCertificateKeyStorePassword;
-    }
-
-    public void setStandbyClientCertificateKeyStoreType(String standbyClientCertificateKeyStoreType) {
-        this.standbyClientCertificateKeyStoreType = standbyClientCertificateKeyStoreType;
-    }
-
-    public void setStandbyClientCertificateKeyStoreUrl(String standbyClientCertificateKeyStoreUrl) {
-        this.standbyClientCertificateKeyStoreUrl = standbyClientCertificateKeyStoreUrl;
-    }
-
-    public void setStandbyClientCertificateKeyStorePassword(String standbyClientCertificateKeyStorePassword) {
-        this.standbyClientCertificateKeyStorePassword = standbyClientCertificateKeyStorePassword;
-    }
-
     public List<String> getPositions() {
         if (positions == null) {
             positions = new ArrayList<>();

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

@@ -1,6 +1,5 @@
 package com.alibaba.otter.canal.instance.spring;
 
-import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.BeanFactory;
@@ -45,9 +44,6 @@ public class SpringCanalInstanceGenerator implements CanalInstanceGenerator {
     }
 
     private BeanFactory getBeanFactory(String springXml) {
-        if (!StringUtils.startsWithIgnoreCase(springXml, "classpath:")) {
-            springXml = "classpath:" + springXml;
-        }
         ApplicationContext applicationContext = new ClassPathXmlApplicationContext(springXml);
         return applicationContext;
     }

instance/spring/src/test/java/com/alibaba/otter/canal/instance/spring/integrated/DefaultSpringInstanceTest.java

@@ -24,7 +24,7 @@ public class DefaultSpringInstanceTest {
     @Before
     public void start() {
         System.setProperty("canal.instance.destination", "retl");
-        context = new ClassPathXmlApplicationContext(new String[] { "classpath:spring/default-instance.xml" });
+        context = new ClassPathXmlApplicationContext(new String[] { "spring/default-instance.xml" });
     }
 
     @After

instance/spring/src/test/java/com/alibaba/otter/canal/instance/spring/integrated/GroupSpringInstanceTest.java

@@ -24,7 +24,7 @@ public class GroupSpringInstanceTest {
     @Before
     public void start() {
         System.setProperty("canal.instance.destination", "retl");
-        context = new ClassPathXmlApplicationContext(new String[] { "classpath:spring/group-instance.xml" });
+        context = new ClassPathXmlApplicationContext(new String[] { "spring/group-instance.xml" });
     }
 
     @After

instance/spring/src/test/java/com/alibaba/otter/canal/instance/spring/integrated/MemorySpringInstanceTest.java

@@ -24,7 +24,7 @@ public class MemorySpringInstanceTest {
     @Before
     public void start() {
         System.setProperty("canal.instance.destination", "retl");
-        context = new ClassPathXmlApplicationContext(new String[] { "classpath:spring/memory-instance.xml" });
+        context = new ClassPathXmlApplicationContext(new String[] { "spring/memory-instance.xml" });
     }
 
     @After

instance/spring/src/test/resources/spring/default-instance.xml

@@ -114,6 +114,7 @@
 		<property name="defaultConnectionTimeoutInSeconds" value="${canal.instance.network.soTimeout:30}" />
 		
 		<!-- 解析编码 -->
+		<!-- property name="connectionCharsetNumber" value="${canal.instance.connectionCharsetNumber:33}" /-->
 		<property name="connectionCharset" value="${canal.instance.connectionCharset:UTF-8}" />
 	
 		<!-- 解析位点记录 -->

instance/spring/src/test/resources/spring/file-instance.xml

@@ -100,6 +100,7 @@
 		<property name="defaultConnectionTimeoutInSeconds" value="${canal.instance.network.soTimeout:30}" />
 		
 		<!-- 解析编码 -->
+		<!-- property name="connectionCharsetNumber" value="${canal.instance.connectionCharsetNumber:33}" /-->
 		<property name="connectionCharset" value="${canal.instance.connectionCharset:UTF-8}" />
 
 		<!-- 解析位点记录 -->

instance/spring/src/test/resources/spring/group-instance.xml

@@ -106,6 +106,7 @@
 		<property name="defaultConnectionTimeoutInSeconds" value="${canal.instance.network.soTimeout:30}" />
 		
 		<!-- 解析编码 -->
+		<!-- property name="connectionCharsetNumber" value="${canal.instance.connectionCharsetNumber:33}" /-->
 		<property name="connectionCharset" value="${canal.instance.connectionCharset:UTF-8}" />
 	
 		<!-- 解析位点记录 -->
@@ -185,6 +186,7 @@
 		<property name="defaultConnectionTimeoutInSeconds" value="${canal.instance.network.soTimeout:30}" />
 		
 		<!-- 解析编码 -->
+		<!-- property name="connectionCharsetNumber" value="${canal.instance.connectionCharsetNumber:33}" /-->
 		<property name="connectionCharset" value="${canal.instance.connectionCharset:UTF-8}" />
 	
 		<!-- 解析位点记录 -->

instance/spring/src/test/resources/spring/memory-instance.xml

@@ -97,6 +97,7 @@
 		<property name="defaultConnectionTimeoutInSeconds" value="${canal.instance.network.soTimeout:30}" />
 		
 		<!-- 解析编码 -->
+		<!-- property name="connectionCharsetNumber" value="${canal.instance.connectionCharsetNumber:33}" /-->
 		<property name="connectionCharset" value="${canal.instance.connectionCharset:UTF-8}" />
 	
 		<!-- 解析位点记录 -->

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

@@ -33,6 +33,7 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
     protected TableMetaTSDB        tableMetaTSDB;
 
     // 编码信息
+    protected byte                 connectionCharsetNumber   = (byte) 33;
     protected Charset              connectionCharset         = Charset.forName("UTF-8");
     protected boolean              filterQueryDcl            = false;
     protected boolean              filterQueryDml            = false;
@@ -204,6 +205,10 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
 
     // ============================ setter / getter =========================
 
+    public void setConnectionCharsetNumber(byte connectionCharsetNumber) {
+        this.connectionCharsetNumber = connectionCharsetNumber;
+    }
+
     public void setConnectionCharsetStd(Charset connectionCharset) {
         this.connectionCharset = connectionCharset;
     }

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

@@ -116,6 +116,7 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements
         MysqlConnection connection = new MysqlConnection(runningInfo.getAddress(),
             runningInfo.getUsername(),
             runningInfo.getPassword(),
+            connectionCharsetNumber,
             runningInfo.getDefaultDatabaseName());
         connection.getConnector().setReceiveBufferSize(64 * 1024);
         connection.getConnector().setSendBufferSize(64 * 1024);

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

@@ -27,7 +27,6 @@ import com.alibaba.otter.canal.parse.driver.mysql.packets.client.RegisterSlaveCo
 import com.alibaba.otter.canal.parse.driver.mysql.packets.client.SemiAckCommandPacket;
 import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ErrorPacket;
 import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ResultSetPacket;
-import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo;
 import com.alibaba.otter.canal.parse.driver.mysql.utils.PacketManager;
 import com.alibaba.otter.canal.parse.exception.CanalParseException;
 import com.alibaba.otter.canal.parse.inbound.ErosaConnection;
@@ -74,27 +73,14 @@ public class MysqlConnection implements ErosaConnection {
         connector.setConnTimeout(connTimeout);
     }
 
-    public MysqlConnection(InetSocketAddress address, String username, String password, String defaultSchema){
+    public MysqlConnection(InetSocketAddress address, String username, String password, byte charsetNumber,
+                           String defaultSchema){
         authInfo = new AuthenticationInfo();
         authInfo.setAddress(address);
         authInfo.setUsername(username);
         authInfo.setPassword(password);
         authInfo.setDefaultDatabaseName(defaultSchema);
-        connector = new MysqlConnector(address, username, password, defaultSchema);
+        connector = new MysqlConnector(address, username, password, charsetNumber, defaultSchema);
-        // 将connection里面的参数透传下
-        connector.setSoTimeout(soTimeout);
-        connector.setConnTimeout(connTimeout);
-    }
-
-    public MysqlConnection(InetSocketAddress address, String username, String password, String defaultSchema,
-                           SslInfo sslInfo){
-        authInfo = new AuthenticationInfo();
-        authInfo.setAddress(address);
-        authInfo.setUsername(username);
-        authInfo.setPassword(password);
-        authInfo.setDefaultDatabaseName(defaultSchema);
-        authInfo.setSslInfo(sslInfo);
-        connector = new MysqlConnector(address, username, password, defaultSchema, sslInfo);
         // 将connection里面的参数透传下
         connector.setSoTimeout(soTimeout);
         connector.setConnTimeout(connTimeout);
@@ -603,32 +589,6 @@ public class MysqlConnection implements ErosaConnection {
         }
     }
 
-    /**
-     * MySQL 8.4版本开始部分命令出现变化
-     * https://dev.mysql.com/doc/relnotes/mysql/8.4/en/news-8-4-0.html#mysqld-8-4-0-deprecation-removal
-     * 
-     * @param major
-     * @param minor
-     * @return
-     */
-    public boolean atLeast(int major, int minor) {
-        if (isMariaDB()) {
-            return false;
-        }
-        String version = connector.getServerVersion();
-        if (StringUtils.isNotEmpty(version)) {
-            String[] parts = version.split("\\.");
-            int majorVer = Integer.parseInt(parts[0]);
-            int minorVer = Integer.parseInt(parts[1]);
-            return (majorVer > major) || (majorVer == major && minorVer >= minor);
-        }
-        return false;
-    }
-
-    public boolean atLeastMySQL84() {
-        return atLeast(8, 4);
-    }
-
     private void accumulateReceivedBytes(long x) {
         if (receivedBinlogBytes != null) {
             receivedBinlogBytes.addAndGet(x);
@@ -653,7 +613,7 @@ public class MysqlConnection implements ErosaConnection {
 
         private String value;
 
-        private BinlogFormat(String value) {
+        private BinlogFormat(String value){
             this.value = value;
         }
 
@@ -693,7 +653,7 @@ public class MysqlConnection implements ErosaConnection {
 
         private String value;
 
-        private BinlogImage(String value) {
+        private BinlogImage(String value){
             this.value = value;
         }
 

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

@@ -305,8 +305,8 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
         MysqlConnection connection = new MysqlConnection(runningInfo.getAddress(),
             runningInfo.getUsername(),
             runningInfo.getPassword(),
-            runningInfo.getDefaultDatabaseName(),
+            connectionCharsetNumber,
-            runningInfo.getSslInfo());
+            runningInfo.getDefaultDatabaseName());
         connection.getConnector().setReceiveBufferSize(receiveBufferSize);
         connection.getConnector().setSendBufferSize(sendBufferSize);
         connection.getConnector().setSoTimeout(defaultConnectionTimeoutInSeconds * 1000);
@@ -668,20 +668,13 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
      * 查询当前的binlog位置
      */
     private EntryPosition findEndPosition(MysqlConnection mysqlConnection) {
-        String showSql = "show master status";
         try {
-            if (mysqlConnection.atLeastMySQL84()) {
+            String showSql = multiStreamEnable ? "show master status with " + destination : "show master status";
-                // 8.4新语法
-                showSql = "show binary log status";
-            } else if (multiStreamEnable) {
-                // 兼容polardb-x的多流binlog
-                showSql = "show master status with " + destination;
-            }
             ResultSetPacket packet = mysqlConnection.query(showSql);
             List<String> fields = packet.getFieldValues();
             if (CollectionUtils.isEmpty(fields)) {
                 throw new CanalParseException(
-                    "command : '" + showSql + "' has an error! pls check. you need (at least one of) the SUPER,REPLICATION CLIENT privilege(s) for this operation");
+                        "command : 'show master status' has an error! pls check. you need (at least one of) the SUPER,REPLICATION CLIENT privilege(s) for this operation");
             }
             EntryPosition endPosition = new EntryPosition(fields.get(0), Long.valueOf(fields.get(1)));
             if (isGTIDMode() && fields.size() > 4) {
@@ -697,7 +690,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
             }
             return endPosition;
         } catch (IOException e) {
-            throw new CanalParseException("command : '" + showSql + "' has an error!", e);
+            throw new CanalParseException("command : 'show master status' has an error!", e);
         }
     }
 
@@ -728,12 +721,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
     @SuppressWarnings("unused")
     private SlaveEntryPosition findSlavePosition(MysqlConnection mysqlConnection) {
         try {
-            String showSql = "show slave stauts";
+            ResultSetPacket packet = mysqlConnection.query("show slave status");
-            if (mysqlConnection.atLeastMySQL84()) {
-                // 兼容mysql 8.4
-                showSql = "show replica status";
-            }
-            ResultSetPacket packet = mysqlConnection.query(showSql);
             List<FieldPacket> names = packet.getFieldDescriptors();
             List<String> fields = packet.getFieldValues();
             if (CollectionUtils.isEmpty(fields)) {

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

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

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

@@ -4,18 +4,39 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 
-import org.apache.commons.lang.StringUtils;
-
-import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
 import com.alibaba.polardbx.druid.sql.SQLUtils;
 import com.alibaba.polardbx.druid.sql.ast.SQLExpr;
 import com.alibaba.polardbx.druid.sql.ast.SQLStatement;
 import com.alibaba.polardbx.druid.sql.ast.expr.SQLIdentifierExpr;
 import com.alibaba.polardbx.druid.sql.ast.expr.SQLPropertyExpr;
-import com.alibaba.polardbx.druid.sql.ast.statement.*;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableAddConstraint;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableAddIndex;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableDropConstraint;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableDropIndex;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableDropKey;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableItem;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableRename;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableStatement;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLConstraint;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLCreateDatabaseStatement;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLCreateIndexStatement;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLCreateTableStatement;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLDeleteStatement;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLDropDatabaseStatement;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLDropIndexStatement;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLDropTableStatement;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLExprTableSource;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLInsertStatement;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLTableSource;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLTruncateStatement;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLUnique;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLUpdateStatement;
 import com.alibaba.polardbx.druid.sql.dialect.mysql.ast.statement.MySqlRenameTableStatement;
 import com.alibaba.polardbx.druid.sql.parser.ParserException;
 import com.alibaba.polardbx.druid.util.JdbcConstants;
+import org.apache.commons.lang.StringUtils;
+
+import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
 
 /**
  * @author agapple 2017年7月27日 下午4:05:34

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

@@ -51,6 +51,7 @@ public class RdsBinlogEventParserProxy extends MysqlEventParser {
             rdsLocalBinlogEventParser.setDestination(destination);
             rdsLocalBinlogEventParser.setAlarmHandler(this.getAlarmHandler());
             rdsLocalBinlogEventParser.setConnectionCharsetStd(this.connectionCharset);
+            rdsLocalBinlogEventParser.setConnectionCharsetNumber(this.connectionCharsetNumber);
             rdsLocalBinlogEventParser.setEnableTsdb(this.enableTsdb);
             rdsLocalBinlogEventParser.setEventBlackFilter(this.eventBlackFilter);
             rdsLocalBinlogEventParser.setFilterQueryDcl(this.filterQueryDcl);

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/request/AbstractRequest.java

@@ -195,7 +195,7 @@ public abstract class AbstractRequest<T> {
     private final HttpResponse executeHttpRequest(HttpGet getMethod, String host) throws Exception {
         SSLContext sslContext = SSLContexts.custom().loadTrustMaterial(null, (TrustStrategy) (arg0, arg1) -> true).build();
         SSLConnectionSocketFactory sslsf = new SSLConnectionSocketFactory(sslContext,
-            new String[] { "TLSv1.2" },
+            new String[] { "TLSv1" },
             null,
             SSLConnectionSocketFactory.ALLOW_ALL_HOSTNAME_VERIFIER);
         Registry registry = RegistryBuilder.create()

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

@@ -207,11 +207,6 @@ public class DatabaseTableMeta implements TableMetaTSDB {
             }
 
             for (String schema : schemas) {
-                // 如果schema命中黑名单，直接跳过
-                // 解决部分数据库可以看到database，但实际表级别无权限的情况
-                if (blackFilter != null && blackFilter.filter(schema + ".%")) {
-                    continue;
-                }
                 // filter views
                 packet = connection.query("show full tables from `" + schema + "` where Table_type = 'BASE TABLE'");
                 columnSize = packet.getFieldDescriptors().size();

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

@@ -3,22 +3,26 @@ package com.alibaba.otter.canal.parse.inbound.mysql.tsdb;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 
-import org.apache.commons.lang.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.alibaba.otter.canal.parse.inbound.TableMeta;
-import com.alibaba.otter.canal.parse.inbound.TableMeta.FieldMeta;
-import com.alibaba.otter.canal.parse.inbound.mysql.ddl.DruidDdlParser;
-import com.alibaba.otter.canal.protocol.position.EntryPosition;
 import com.alibaba.polardbx.druid.DbType;
 import com.alibaba.polardbx.druid.sql.SQLUtils;
 import com.alibaba.polardbx.druid.sql.ast.SQLDataType;
 import com.alibaba.polardbx.druid.sql.ast.SQLDataTypeImpl;
 import com.alibaba.polardbx.druid.sql.ast.SQLExpr;
 import com.alibaba.polardbx.druid.sql.ast.SQLStatement;
-import com.alibaba.polardbx.druid.sql.ast.expr.*;
+import com.alibaba.polardbx.druid.sql.ast.expr.SQLCharExpr;
-import com.alibaba.polardbx.druid.sql.ast.statement.*;
+import com.alibaba.polardbx.druid.sql.ast.expr.SQLIdentifierExpr;
+import com.alibaba.polardbx.druid.sql.ast.expr.SQLMethodInvokeExpr;
+import com.alibaba.polardbx.druid.sql.ast.expr.SQLNullExpr;
+import com.alibaba.polardbx.druid.sql.ast.expr.SQLPropertyExpr;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLColumnConstraint;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLColumnDefinition;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLColumnPrimaryKey;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLColumnUniqueKey;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLCreateTableStatement;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLNotNullConstraint;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLNullConstraint;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLSelectOrderByItem;
+import com.alibaba.polardbx.druid.sql.ast.statement.SQLTableElement;
 import com.alibaba.polardbx.druid.sql.dialect.mysql.ast.MySqlPrimaryKey;
 import com.alibaba.polardbx.druid.sql.dialect.mysql.ast.MySqlUnique;
 import com.alibaba.polardbx.druid.sql.dialect.mysql.ast.expr.MySqlOrderingExpr;
@@ -28,6 +32,13 @@ import com.alibaba.polardbx.druid.sql.repository.SchemaRepository;
 import com.alibaba.polardbx.druid.sql.visitor.SQLASTOutputVisitor;
 import com.alibaba.polardbx.druid.sql.visitor.VisitorFeature;
 import com.alibaba.polardbx.druid.util.JdbcConstants;
+import org.apache.commons.lang.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import com.alibaba.otter.canal.parse.inbound.TableMeta;
+import com.alibaba.otter.canal.parse.inbound.TableMeta.FieldMeta;
+import com.alibaba.otter.canal.parse.inbound.mysql.ddl.DruidDdlParser;
+import com.alibaba.otter.canal.protocol.position.EntryPosition;
 
 /**
  * 基于DDL维护的内存表结构

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

@@ -30,9 +30,6 @@ public class TableMetaTSDBBuilder {
             if (applicationContext == null) {
                 synchronized (contexts) {
                     if (applicationContext == null) {
-                        if (!StringUtils.startsWithIgnoreCase(springXml, "classpath:")) {
-                            springXml = "classpath:" + springXml;
-                        }
                         applicationContext = new ClassPathXmlApplicationContext(springXml);
                         contexts.put(destination, applicationContext);
                     }

parse/src/main/java/com/alibaba/otter/canal/parse/support/AuthenticationInfo.java

@@ -2,12 +2,10 @@ package com.alibaba.otter.canal.parse.support;
 
 import java.net.InetSocketAddress;
 
+import com.alibaba.otter.canal.common.utils.CommonUtils;
 import org.apache.commons.lang.builder.ToStringBuilder;
 import org.apache.commons.lang.builder.ToStringStyle;
 
-import com.alibaba.otter.canal.common.utils.CommonUtils;
-import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo;
-
 /**
  * 数据库认证信息
  *
@@ -16,15 +14,16 @@ import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo;
  */
 public class AuthenticationInfo {
 
+
+
     private InetSocketAddress address;            // 主库信息
     private String            username;           // 帐号
     private String            password;           // 密码
     private String            defaultDatabaseName;// 默认链接的数据库
-    private String            pwdPublicKey;       // 公钥
+    private String            pwdPublicKey;       //公钥
-    private boolean           enableDruid;        // 是否使用druid加密解密数据库密码
+    private boolean           enableDruid;        //是否使用druid加密解密数据库密码
-    private SslInfo           sslInfo;
 
-    public void initPwd() throws Exception {
+    public void initPwd() throws Exception{
         if (enableDruid) {
             this.password = CommonUtils.decryptDruidPassword(pwdPublicKey, password);
         }
@@ -38,19 +37,13 @@ public class AuthenticationInfo {
         this(address, username, password, "");
     }
 
-    private AuthenticationInfo(InetSocketAddress address, String username, String password, String defaultDatabaseName){
+    public AuthenticationInfo(InetSocketAddress address, String username, String password, String defaultDatabaseName){
         this.address = address;
         this.username = username;
         this.password = password;
         this.defaultDatabaseName = defaultDatabaseName;
     }
 
-    public AuthenticationInfo(InetSocketAddress address, String username, String password, String defaultDatabaseName,
-                              SslInfo sslInfo){
-        this(address, username, password, defaultDatabaseName);
-        this.sslInfo = sslInfo;
-    }
-
     public InetSocketAddress getAddress() {
         return address;
     }
@@ -99,14 +92,6 @@ public class AuthenticationInfo {
         this.enableDruid = enableDruid;
     }
 
-    public SslInfo getSslInfo() {
-        return sslInfo;
-    }
-
-    public void setSslInfo(SslInfo sslInfo) {
-        this.sslInfo = sslInfo;
-    }
-
     @Override
     public String toString() {
         return ToStringBuilder.reflectionToString(this, ToStringStyle.DEFAULT_STYLE);

pom.xml

@@ -99,13 +99,13 @@
         <java_target_version>1.8</java_target_version>
         <file_encoding>UTF-8</file_encoding>
         <javadoc_skip>true</javadoc_skip>
-        <spring_version>5.3.26</spring_version>
+        <spring_version>5.3.9</spring_version>
         <log4j_version>2.17.0</log4j_version>
-        <rocketmq_version>4.9.8</rocketmq_version>
+        <rocketmq_version>4.8.0</rocketmq_version>
         <rabbitmq_version>5.18.0</rabbitmq_version>
         <mq_amqp_client>1.0.3</mq_amqp_client>
         <kafka_version>2.4.0</kafka_version>
-        <pulsar_version>2.11.4</pulsar_version>
+        <pulsar_version>2.8.1</pulsar_version>
         <mysql_driver_version>5.1.48</mysql_driver_version>
         <maven-jacoco-plugin.version>0.8.3</maven-jacoco-plugin.version>
         <maven-surefire.version>2.22.1</maven-surefire.version>

server/src/test/java/com/alibaba/otter/canal/server/CanalServerTest.java

@@ -253,6 +253,7 @@ public class CanalServerTest {
 
         parameter.setDefaultConnectionTimeoutInSeconds(30);
         parameter.setConnectionCharset("UTF-8");
+        parameter.setConnectionCharsetNumber((byte) 33);
         parameter.setReceiveBufferSize(8 * 1024);
         parameter.setSendBufferSize(8 * 1024);
 

server/src/test/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded_FileModeTest.java

@@ -40,6 +40,7 @@ public class CanalServerWithEmbedded_FileModeTest extends BaseCanalServerWithEmb
 
         parameter.setDefaultConnectionTimeoutInSeconds(30);
         parameter.setConnectionCharset("UTF-8");
+        parameter.setConnectionCharsetNumber((byte) 33);
         parameter.setReceiveBufferSize(8 * 1024);
         parameter.setSendBufferSize(8 * 1024);
 

server/src/test/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded_StandaloneTest.java

@@ -43,6 +43,7 @@ public class CanalServerWithEmbedded_StandaloneTest extends BaseCanalServerWithE
 
         parameter.setDefaultConnectionTimeoutInSeconds(30);
         parameter.setConnectionCharset("UTF-8");
+        parameter.setConnectionCharsetNumber((byte) 33);
         parameter.setReceiveBufferSize(8 * 1024);
         parameter.setSendBufferSize(8 * 1024);
 

server/src/test/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded_StandbyTest.java

@@ -55,6 +55,7 @@ public class CanalServerWithEmbedded_StandbyTest extends BaseCanalServerWithEmbe
 
         parameter.setDefaultConnectionTimeoutInSeconds(30);
         parameter.setConnectionCharset("UTF-8");
+        parameter.setConnectionCharsetNumber((byte) 33);
         parameter.setReceiveBufferSize(8 * 1024);
         parameter.setSendBufferSize(8 * 1024);