feat 元数据与target数据源不一致时 进行兼容处理 (#5168)

* feat 元数据与target数据源不一致时 进行兼容处理

* feat 元数据与target数据源不一致时 进行兼容处理  优化代码

* Feature #FR-761 ddl执行失败会造成后续未执行的dml丢弃 针对ddl exec failed catch the ex

README.md

@@ -93,10 +93,6 @@ canal 作为 MySQL binlog 增量获取和解析工具，可将变更记录投递
 
 - 参考文档: [Canal Kafka/RocketMQ QuickStart](https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart)
 
-## 基于canal开发的工具
-
--  canal2sql(基于binlog生成SQL) : [https://github.com/zhuchao941/canal2sql]
-
 ## 相关开源&产品
 
 - [canal 消费端开源项目: Otter](http://github.com/alibaba/otter)

admin/admin-ui/pom.xml

@@ -3,7 +3,7 @@
   <parent>
     <artifactId>canal.admin</artifactId>
     <groupId>com.alibaba.otter</groupId>
-    <version>1.1.8-SNAPSHOT</version>
+    <version>1.1.7</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
 

admin/admin-web/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.admin</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.8-SNAPSHOT</version>
+        <version>1.1.7</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>canal.admin-web</artifactId>

admin/admin-web/src/main/java/com/alibaba/otter/canal/admin/config/EbeanConfig.java

@@ -10,7 +10,6 @@ import java.util.List;
 
 import javax.sql.DataSource;
 
-import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 

admin/admin-web/src/main/java/com/alibaba/otter/canal/admin/config/WebConfig.java

@@ -5,8 +5,6 @@ import java.io.PrintWriter;
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 
-import org.apache.commons.lang.StringUtils;
-import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.http.HttpMethod;
 import org.springframework.http.HttpStatus;
@@ -28,24 +26,8 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 @Configuration
 public class WebConfig implements WebMvcConfigurer {
 
-    @Value(value = "${canal.adminUser}")
-    private String user;
-
-    @Value(value = "${canal.adminPasswd}")
-    private String passwd;
-
     @Override
     public void addInterceptors(InterceptorRegistry registry) {
-        if (StringUtils.isEmpty(user)) {
-            throw new IllegalArgumentException(
-                "canal.adminUser is empty , pls check https://github.com/alibaba/canal/issues/4941");
-        }
-
-        if (StringUtils.isEmpty(passwd)) {
-            throw new IllegalArgumentException(
-                "canal.adminPasswd is empty , pls check https://github.com/alibaba/canal/issues/4941");
-        }
-
         registry.addInterceptor(new HandlerInterceptor() {
 
             @Override

admin/admin-web/src/main/java/com/alibaba/otter/canal/admin/controller/PollingConfigController.java

@@ -2,11 +2,15 @@ package com.alibaba.otter.canal.admin.controller;
 
 import java.security.NoSuchAlgorithmException;
 
-import org.apache.commons.lang.RandomStringUtils;
 import org.apache.commons.lang.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
-import org.springframework.web.bind.annotation.*;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RequestHeader;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
 
 import com.alibaba.otter.canal.admin.model.BaseModel;
 import com.alibaba.otter.canal.admin.model.CanalConfig;
@@ -24,7 +28,7 @@ import com.alibaba.otter.canal.protocol.SecurityUtil;
 @RequestMapping("/api/{env}/config")
 public class PollingConfigController {
 
-    private static final byte[] seeds = RandomStringUtils.random(16).getBytes();
+    private static final byte[] seeds = "canal is best!".getBytes();
 
     @Value(value = "${canal.adminUser}")
     String                      user;

admin/admin-web/src/main/resources/application.yml

@@ -18,4 +18,4 @@ spring.datasource:
 
 canal:
   adminUser: admin
-  adminPasswd:
+  adminPasswd: admin

admin/pom.xml

@@ -3,13 +3,13 @@
     <parent>
         <groupId>com.alibaba.otter</groupId>
         <artifactId>canal</artifactId>
-        <version>1.1.8-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.alibaba.otter</groupId>
     <artifactId>canal.admin</artifactId>
-    <version>1.1.8-SNAPSHOT</version>
+    <version>1.1.7</version>
     <packaging>pom</packaging>
     <name>canal admin module for otter ${project.version}</name>
     <modules>

client-adapter/clickhouse/pom.xml

@@ -1,84 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0"
-         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-    <parent>
-        <artifactId>canal.client-adapter</artifactId>
-        <groupId>com.alibaba.otter</groupId>
-        <version>1.1.8-SNAPSHOT</version>
-        <relativePath>../pom.xml</relativePath>
-    </parent>
-    <modelVersion>4.0.0</modelVersion>
-    <groupId>com.alibaba.otter</groupId>
-    <artifactId>client-adapter.clickhouse</artifactId>
-    <packaging>jar</packaging>
-    <name>canal client adapter clickhouse module for otter ${project.version}</name>
-
-    <dependencies>
-        <dependency>
-            <groupId>com.alibaba.otter</groupId>
-            <artifactId>client-adapter.common</artifactId>
-            <version>${project.version}</version>
-            <scope>provided</scope>
-        </dependency>
-        <dependency>
-            <groupId>ru.yandex.clickhouse</groupId>
-            <artifactId>clickhouse-jdbc</artifactId>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>junit</groupId>
-            <artifactId>junit</artifactId>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.yaml</groupId>
-            <artifactId>snakeyaml</artifactId>
-            <scope>test</scope>
-        </dependency>
-    </dependencies>
-
-    <build>
-        <plugins>
-            <plugin>
-                <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-assembly-plugin</artifactId>
-                <version>2.4</version>
-                <configuration>
-                    <descriptorRefs>
-                        <descriptorRef>jar-with-dependencies</descriptorRef>
-                    </descriptorRefs>
-                </configuration>
-                <executions>
-                    <execution>
-                        <id>make-assembly</id>
-                        <phase>package</phase>
-                        <goals>
-                            <goal>single</goal>
-                        </goals>
-                    </execution>
-                </executions>
-            </plugin>
-            <plugin>
-                <artifactId>maven-antrun-plugin</artifactId>
-                <executions>
-                    <execution>
-                        <phase>package</phase>
-                        <goals>
-                            <goal>run</goal>
-                        </goals>
-                        <configuration>
-                            <tasks>
-                                <copy todir="${project.basedir}/../launcher/target/classes/clickhouse" overwrite="true">
-                                    <fileset dir="${project.basedir}/target/classes/clickhouse" erroronmissingdir="true">
-                                        <include name="*.yml" />
-                                    </fileset>
-                                </copy>
-                            </tasks>
-                        </configuration>
-                    </execution>
-                </executions>
-            </plugin>
-        </plugins>
-    </build>
-</project>

client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/ClickHouseAdapter.java

@@ -1,340 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.clickhouse;
-
-import java.sql.Connection;
-import java.sql.SQLException;
-import java.util.*;
-import java.util.concurrent.ConcurrentHashMap;
-
-import org.apache.commons.lang.BooleanUtils;
-import org.apache.commons.lang.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.alibaba.druid.filter.stat.StatFilter;
-import com.alibaba.druid.pool.DruidDataSource;
-import com.alibaba.druid.util.JdbcUtils;
-import com.alibaba.otter.canal.client.adapter.OuterAdapter;
-import com.alibaba.otter.canal.client.adapter.clickhouse.config.ConfigLoader;
-import com.alibaba.otter.canal.client.adapter.clickhouse.config.MappingConfig;
-import com.alibaba.otter.canal.client.adapter.clickhouse.config.MirrorDbConfig;
-import com.alibaba.otter.canal.client.adapter.clickhouse.monitor.ClickHouseConfigMonitor;
-import com.alibaba.otter.canal.client.adapter.clickhouse.service.ClickHouseBatchSyncService;
-import com.alibaba.otter.canal.client.adapter.clickhouse.service.ClickHouseEtlService;
-import com.alibaba.otter.canal.client.adapter.clickhouse.service.ClickHouseMirrorDbBatchSyncService;
-import com.alibaba.otter.canal.client.adapter.clickhouse.support.SyncUtil;
-import com.alibaba.otter.canal.client.adapter.support.*;
-
-/**
- * ClickHouse Adapter implementation
- *
- * @author: Xander
- * @date: Created in 2023/11/10 1:13
- * @email: zhrunxin33@gmail.com
- */
-@SPI
-public class ClickHouseAdapter implements OuterAdapter {
-
-    private static final Logger                         logger = LoggerFactory.getLogger(ClickHouseAdapter.class);
-
-    // Store the mapping of filename and configuration, load yml files below
-    // resource path
-    private Map<String, MappingConfig>              clickHouseMapping   = new ConcurrentHashMap<>();
-
-    // Schema -> Table -> MappingConfig
-    private Map<String, Map<String, MappingConfig>> mappingConfigCache  = new ConcurrentHashMap<>();
-
-    // Mirror DB Configuration, don't need to load column mapping
-    private Map<String, MirrorDbConfig>             mirrorDbConfigCache = new ConcurrentHashMap<>();
-
-    private DruidDataSource                             dataSource;
-
-    private ClickHouseBatchSyncService clickHouseBatchSyncService;
-
-    private ClickHouseMirrorDbBatchSyncService clickHouseMirrorDbBatchSyncService;
-
-    private Properties                                  envProperties;
-
-    // Launch configuration
-    private OuterAdapterConfig                      configuration;
-
-    private ClickHouseConfigMonitor clickHouseConfigMonitor;
-
-    public Map<String, MappingConfig> getClickHouseMapping() {
-        return clickHouseMapping;
-    }
-
-    @Override
-    public void init(OuterAdapterConfig configuration, Properties envProperties) {
-        this.envProperties = envProperties;
-        this.configuration = configuration;
-        // Load DB type from adapter.launch/bootstrap.yml
-        Map<String, String> properties = configuration.getProperties();
-        String dbType = JdbcUtils.getDbType(properties.get("jdbc.url"), null);
-        // 当.yml文件编码格式存在问题，此处clickhouse yml文件构建 可能会抛出异常
-        Map<String, MappingConfig> clickHouseMappingTmp = ConfigLoader.load(envProperties);
-        // 过滤不匹配的key的配置
-        clickHouseMappingTmp.forEach((key, config) -> {
-            addConfig(key, config);
-        });
-
-        if (clickHouseMapping.isEmpty()) {
-            throw new RuntimeException("No clickhouse adapter found for config key: " + configuration.getKey());
-        }
-
-        // 初始化连接池
-        dataSource = new DruidDataSource();
-        dataSource.setDriverClassName(properties.get("jdbc.driverClassName"));
-        dataSource.setUrl(properties.get("jdbc.url"));
-        dataSource.setUsername(properties.get("jdbc.username"));
-        dataSource.setPassword(properties.get("jdbc.password"));
-        dataSource.setInitialSize(1);
-        dataSource.setMinIdle(1);
-        dataSource.setMaxActive(30);
-        dataSource.setMaxWait(60000);
-        dataSource.setTimeBetweenEvictionRunsMillis(60000);
-        dataSource.setMinEvictableIdleTimeMillis(300000);
-        dataSource.setUseUnfairLock(true);
-        dataSource.setDefaultAutoCommit(false); // disable auto commit or disable Transactional
-        dataSource.setDbType(dbType);
-
-        if ("true".equals(properties.getOrDefault("druid.stat.enable", "true"))) {
-            StatFilter statFilter = new StatFilter();
-            statFilter.setSlowSqlMillis(Long.parseLong(properties.getOrDefault("druid.stat.slowSqlMillis", "1000")));
-            statFilter.setMergeSql(true);
-            statFilter.setLogSlowSql(true);
-            dataSource.setProxyFilters(Collections.singletonList(statFilter));
-        }
-
-        try {
-            dataSource.init();
-        } catch (SQLException e) {
-            logger.error("ERROR ## failed to initial datasource: " + properties.get("jdbc.url"), e);
-        }
-
-        String threads = properties.get("threads");
-        String scheduleTime = properties.get("scheduleTime");
-        String batchSize = properties.get("batchSize");
-
-        boolean skipDupException = BooleanUtils.toBoolean(configuration.getProperties()
-                .getOrDefault("skipDupException", "true"));
-        clickHouseBatchSyncService = new ClickHouseBatchSyncService(dataSource,
-                threads != null ? Integer.valueOf(threads) : null,
-                batchSize != null ? Integer.valueOf(batchSize) : null,
-                scheduleTime != null ? Long.valueOf(scheduleTime) : null,
-                skipDupException);
-
-        clickHouseMirrorDbBatchSyncService = new ClickHouseMirrorDbBatchSyncService(mirrorDbConfigCache,
-                dataSource,
-                threads != null ? Integer.valueOf(threads) : null,
-                batchSize != null ? Integer.valueOf(batchSize) : null,
-                scheduleTime != null ? Long.valueOf(scheduleTime) : null,
-                clickHouseBatchSyncService.getColumnsTypeCache(),
-                skipDupException);
-
-        clickHouseConfigMonitor = new ClickHouseConfigMonitor();
-        clickHouseConfigMonitor.init(configuration.getKey(), this, envProperties);
-    }
-
-    /**
-     * Sync main entrance
-     *
-     * @param dmls 数据包
-     */
-    @Override
-    public void sync(List<Dml> dmls) {
-        if (dmls == null || dmls.isEmpty()) {
-            return;
-        }
-        try {
-            // If mappingConfigCache(column mapping) is empty, that must be mirroring synchronize
-            if (!mappingConfigCache.isEmpty()) {
-                clickHouseBatchSyncService.sync(mappingConfigCache, dmls, envProperties);
-            }
-            clickHouseMirrorDbBatchSyncService.sync(dmls);
-        } catch (Exception e) {
-            throw new RuntimeException(e);
-        }
-    }
-
-    /**
-     * Destroy
-     */
-    @Override
-    public void destroy() {
-        if (clickHouseConfigMonitor != null) {
-            clickHouseConfigMonitor.destroy();
-        }
-
-        if (clickHouseBatchSyncService != null) {
-            clickHouseBatchSyncService.close();
-        }
-
-        if (dataSource != null) {
-            dataSource.close();
-        }
-    }
-
-
-    /**
-     * ETL方法
-     *
-     * @param task 任务名, 对应配置名
-     * @param params etl筛选条件
-     * @return ETL结果
-     */
-    @Override
-    public EtlResult etl(String task, List<String> params) {
-        EtlResult etlResult = new EtlResult();
-        MappingConfig config = clickHouseMapping.get(task);
-        ClickHouseEtlService clickhouseEtlService = new ClickHouseEtlService(dataSource, config);
-        if (config != null) {
-            return clickhouseEtlService.importData(params);
-        } else {
-            StringBuilder resultMsg = new StringBuilder();
-            boolean resSucc = true;
-            for (MappingConfig configTmp : clickHouseMapping.values()) {
-                // 取所有的destination为task的配置
-                if (configTmp.getDestination().equals(task)) {
-                    EtlResult etlRes = clickhouseEtlService.importData(params);
-                    if (!etlRes.getSucceeded()) {
-                        resSucc = false;
-                        resultMsg.append(etlRes.getErrorMessage()).append("\n");
-                    } else {
-                        resultMsg.append(etlRes.getResultMessage()).append("\n");
-                    }
-                }
-            }
-            if (resultMsg.length() > 0) {
-                etlResult.setSucceeded(resSucc);
-                if (resSucc) {
-                    etlResult.setResultMessage(resultMsg.toString());
-                } else {
-                    etlResult.setErrorMessage(resultMsg.toString());
-                }
-                return etlResult;
-            }
-        }
-        etlResult.setSucceeded(false);
-        etlResult.setErrorMessage("Task not found");
-        return etlResult;
-    }
-
-    /**
-     * 获取总数方法
-     *
-     * @param task 任务名, 对应配置名
-     * @return 总数
-     */
-    @Override
-    public Map<String, Object> count(String task) {
-        MappingConfig config = clickHouseMapping.get(task);
-        MappingConfig.DbMapping dbMapping = config.getDbMapping();
-        String sql = "SELECT COUNT(1) AS cnt FROM " + SyncUtil.getDbTableName(dbMapping, dataSource.getDbType());
-        Connection conn = null;
-        Map<String, Object> res = new LinkedHashMap<>();
-        try {
-            conn = dataSource.getConnection();
-            Util.sqlRS(conn, sql, rs -> {
-                try {
-                    if (rs.next()) {
-                        Long rowCount = rs.getLong("cnt");
-                        res.put("count", rowCount);
-                    }
-                } catch (SQLException e) {
-                    logger.error(e.getMessage(), e);
-                }
-            });
-        } catch (SQLException e) {
-            logger.error(e.getMessage(), e);
-        } finally {
-            if (conn != null) {
-                try {
-                    conn.close();
-                } catch (SQLException e) {
-                    logger.error(e.getMessage(), e);
-                }
-            }
-        }
-        res.put("targetTable", SyncUtil.getDbTableName(dbMapping, dataSource.getDbType()));
-
-        return res;
-    }
-
-    /**
-     * 获取对应canal instance name 或 mq topic
-     *
-     * @param task 任务名, 对应配置名
-     * @return destination
-     */
-    @Override
-    public String getDestination(String task) {
-        MappingConfig config = clickHouseMapping.get(task);
-        if (config != null) {
-            return config.getDestination();
-        }
-        return null;
-    }
-
-    private void addSyncConfigToCache(String configName, MappingConfig mappingConfig) {
-        if (!mappingConfig.getDbMapping().getMirrorDb()) {
-            String key;
-            if (envProperties != null && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) {
-                key = StringUtils.trimToEmpty(mappingConfig.getDestination()) + "-"
-                        + StringUtils.trimToEmpty(mappingConfig.getGroupId()) + "_"
-                        + mappingConfig.getDbMapping().getDatabase() + "-" + mappingConfig.getDbMapping().getTable();
-            } else {
-                key = StringUtils.trimToEmpty(mappingConfig.getDestination()) + "_"
-                        + mappingConfig.getDbMapping().getDatabase() + "-" + mappingConfig.getDbMapping().getTable();
-            }
-            Map<String, MappingConfig> configMap = mappingConfigCache.computeIfAbsent(key,
-                    k1 -> new ConcurrentHashMap<>());
-            configMap.put(configName, mappingConfig);
-        } else {
-            // mirrorDB
-            String key = StringUtils.trimToEmpty(mappingConfig.getDestination()) + "."
-                    + mappingConfig.getDbMapping().getDatabase();
-            mirrorDbConfigCache.put(key, MirrorDbConfig.create(configName, mappingConfig));
-        }
-    }
-
-    public boolean addConfig(String fileName, MappingConfig config) {
-        if (match(config)) {
-            clickHouseMapping.put(fileName, config);
-            addSyncConfigToCache(fileName, config);
-            FileName2KeyMapping.register(getClass().getAnnotation(SPI.class).value(), fileName,
-                    configuration.getKey());
-            return true;
-        }
-        return false;
-    }
-
-    public void updateConfig(String fileName, MappingConfig config) {
-        if (config.getOuterAdapterKey() != null && !config.getOuterAdapterKey()
-                .equals(configuration.getKey())) {
-            // 理论上不允许改这个 因为本身就是通过这个关联起Adapter和Config的
-            throw new RuntimeException("not allow to change outAdapterKey");
-        }
-        clickHouseMapping.put(fileName, config);
-        addSyncConfigToCache(fileName, config);
-    }
-
-    public void deleteConfig(String fileName) {
-        clickHouseMapping.remove(fileName);
-        for (Map<String, MappingConfig> configMap : mappingConfigCache.values()) {
-            if (configMap != null) {
-                configMap.remove(fileName);
-            }
-        }
-        FileName2KeyMapping.unregister(getClass().getAnnotation(SPI.class).value(), fileName);
-    }
-
-    private boolean match(MappingConfig config) {
-        boolean sameMatch = config.getOuterAdapterKey() != null && config.getOuterAdapterKey()
-                .equalsIgnoreCase(configuration.getKey());
-        boolean prefixMatch = config.getOuterAdapterKey() == null && configuration.getKey()
-                .startsWith(StringUtils
-                        .join(new String[]{Util.AUTO_GENERATED_PREFIX, config.getDestination(),
-                                config.getGroupId()}, '-'));
-        return sameMatch || prefixMatch;
-    }
-}

client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/config/ConfigLoader.java

@@ -1,52 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.clickhouse.config;
-
-import java.util.LinkedHashMap;
-import java.util.Map;
-import java.util.Properties;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader;
-import com.alibaba.otter.canal.client.adapter.support.YamlUtils;
-
-/**
- * CLICKHOUSE表映射配置加载器
- *
- * @author: Xander
- * @date: Created in 2023/11/10 22:23
- * @email: zhrunxin33@gmail.com
- * @version 1.1.8
- */
-public class ConfigLoader {
-
-    private static Logger logger = LoggerFactory.getLogger(ConfigLoader.class);
-
-    /**
-     * 加载CLICKHOUSE表映射配置
-     *
-     * @return 配置名/配置文件名--对象
-     */
-    public static Map<String, MappingConfig> load(Properties envProperties) {
-        logger.info("## Start loading clickhouse mapping config ... ");
-
-        Map<String, MappingConfig> result = new LinkedHashMap<>();
-
-        Map<String, String> configContentMap = MappingConfigsLoader.loadConfigs("clickhouse");
-        configContentMap.forEach((fileName, content) -> {
-            MappingConfig config = YamlUtils.ymlToObj(null, content, MappingConfig.class, null, envProperties);
-            if (config == null) {
-                return;
-            }
-            try {
-                config.validate();
-            } catch (Exception e) {
-                throw new RuntimeException("ERROR Config: " + fileName + " " + e.getMessage(), e);
-            }
-            result.put(fileName, config);
-        });
-
-        logger.info("## ClickHouse mapping config loaded");
-        return result;
-    }
-}

client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/config/MappingConfig.java

@@ -1,227 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.clickhouse.config;
-
-import java.util.LinkedHashMap;
-import java.util.Map;
-
-import org.apache.commons.lang.StringUtils;
-
-import com.alibaba.otter.canal.client.adapter.support.AdapterConfig;
-
-/**
- * CLICKHOUSE表映射配置
- *
- * @author: Xander
- * @date: Created in 2023/11/10 22:23
- * @email: zhrunxin33@gmail.com
- * @version 1.1.8
- */
-public class MappingConfig implements AdapterConfig {
-
-    private String    dataSourceKey;      // 数据源key
-
-    private String    destination;        // canal实例或MQ的topic
-
-    private String    groupId;            // groupId
-
-    private String    outerAdapterKey;    // 对应适配器的key
-
-    private boolean   concurrent = false; // 是否并行同步
-
-    private DbMapping dbMapping;          // db映射配置
-
-    public String getDataSourceKey() {
-        return dataSourceKey;
-    }
-
-    public void setDataSourceKey(String dataSourceKey) {
-        this.dataSourceKey = dataSourceKey;
-    }
-
-    public String getGroupId() {
-        return groupId;
-    }
-
-    public void setGroupId(String groupId) {
-        this.groupId = groupId;
-    }
-
-    public String getOuterAdapterKey() {
-        return outerAdapterKey;
-    }
-
-    public void setOuterAdapterKey(String outerAdapterKey) {
-        this.outerAdapterKey = outerAdapterKey;
-    }
-
-    public boolean getConcurrent() {
-        return concurrent;
-    }
-
-    public void setConcurrent(boolean concurrent) {
-        this.concurrent = concurrent;
-    }
-
-    public DbMapping getDbMapping() {
-        return dbMapping;
-    }
-
-    public void setDbMapping(DbMapping dbMapping) {
-        this.dbMapping = dbMapping;
-    }
-
-    public String getDestination() {
-        return destination;
-    }
-
-    public void setDestination(String destination) {
-        this.destination = destination;
-    }
-
-    public AdapterMapping getMapping() {
-        return dbMapping;
-    }
-
-    public void validate() {
-        if (dbMapping.database == null || dbMapping.database.isEmpty()) {
-            throw new NullPointerException("dbMapping.database");
-        }
-        if (!dbMapping.getMirrorDb() && (dbMapping.table == null || dbMapping.table.isEmpty())) {
-            throw new NullPointerException("dbMapping.table");
-        }
-        if (!dbMapping.getMirrorDb() && (dbMapping.targetTable == null || dbMapping.targetTable.isEmpty())) {
-            throw new NullPointerException("dbMapping.targetTable");
-        }
-    }
-
-    public static class DbMapping implements AdapterMapping {
-
-        private boolean             mirrorDb        = false;                 // 是否镜像库
-        private String              database;                                // 数据库名或schema名
-        private String              table;                                   // 表名
-        private Map<String, String> targetPk        = new LinkedHashMap<>(); // 目标表主键字段
-        private boolean             mapAll          = false;                 // 映射所有字段
-        private String              targetDb;                                // 目标库名
-        private String              targetTable;                             // 目标表名
-        private Map<String, String> targetColumns;                           // 目标表字段映射
-
-        private boolean             caseInsensitive = false;                 // 目标表不区分大小写，默认是否
-
-        private String              etlCondition;                            // etl条件sql
-
-        private int                 readBatch       = 5000;
-        private int                 commitBatch     = 5000;                  // etl等批量提交大小
-
-        private Map<String, String> allMapColumns;
-
-        public boolean getMirrorDb() {
-            return mirrorDb;
-        }
-
-        public void setMirrorDb(boolean mirrorDb) {
-            this.mirrorDb = mirrorDb;
-        }
-
-        public String getDatabase() {
-            return database;
-        }
-
-        public void setDatabase(String database) {
-            this.database = database;
-        }
-
-        public String getTable() {
-            return table;
-        }
-
-        public void setTable(String table) {
-            this.table = table;
-        }
-
-        public Map<String, String> getTargetPk() {
-            return targetPk;
-        }
-
-        public void setTargetPk(Map<String, String> targetPk) {
-            this.targetPk = targetPk;
-        }
-
-        public Boolean getMapAll() {
-            return mapAll;
-        }
-
-        public void setMapAll(Boolean mapAll) {
-            this.mapAll = mapAll;
-        }
-
-        public String getTargetDb() {
-            return targetDb;
-        }
-
-        public void setTargetDb(String targetDb) {
-            this.targetDb = targetDb;
-        }
-
-        public String getTargetTable() {
-            return targetTable;
-        }
-
-        public void setTargetTable(String targetTable) {
-            this.targetTable = targetTable;
-        }
-
-        public Map<String, String> getTargetColumns() {
-            if (targetColumns != null) {
-                targetColumns.forEach((key, value) -> {
-                    if (StringUtils.isEmpty(value)) {
-                        targetColumns.put(key, key);
-                    }
-                });
-            }
-            return targetColumns;
-        }
-
-        public void setTargetColumns(Map<String, String> targetColumns) {
-            this.targetColumns = targetColumns;
-        }
-
-        public boolean isCaseInsensitive() {
-            return caseInsensitive;
-        }
-
-        public void setCaseInsensitive(boolean caseInsensitive) {
-            this.caseInsensitive = caseInsensitive;
-        }
-
-        public String getEtlCondition() {
-            return etlCondition;
-        }
-
-        public void setEtlCondition(String etlCondition) {
-            this.etlCondition = etlCondition;
-        }
-
-        public int getReadBatch() {
-            return readBatch;
-        }
-
-        public void setReadBatch(int readBatch) {
-            this.readBatch = readBatch;
-        }
-
-        public int getCommitBatch() {
-            return commitBatch;
-        }
-
-        public void setCommitBatch(int commitBatch) {
-            this.commitBatch = commitBatch;
-        }
-
-        public Map<String, String> getAllMapColumns() {
-            return allMapColumns;
-        }
-
-        public void setAllMapColumns(Map<String, String> allMapColumns) {
-            this.allMapColumns = allMapColumns;
-        }
-    }
-}

client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/config/MirrorDbConfig.java

@@ -1,51 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.clickhouse.config;
-
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-
-/**
- * @author: Xander
- * @date: Created in 2023/11/10 22:23
- * @email: zhrunxin33@gmail.com
- * @version 1.1.8
- */
-
-public class MirrorDbConfig {
-
-    private String                     fileName;
-    private MappingConfig              mappingConfig;
-    private Map<String, MappingConfig> tableConfig = new ConcurrentHashMap<>();
-
-    public static MirrorDbConfig create(String fileName, MappingConfig mappingConfig) {
-        return new MirrorDbConfig(fileName, mappingConfig);
-    }
-
-    public MirrorDbConfig(String fileName, MappingConfig mappingConfig){
-        this.fileName = fileName;
-        this.mappingConfig = mappingConfig;
-    }
-
-    public String getFileName() {
-        return fileName;
-    }
-
-    public void setFileName(String fileName) {
-        this.fileName = fileName;
-    }
-
-    public MappingConfig getMappingConfig() {
-        return mappingConfig;
-    }
-
-    public void setMappingConfig(MappingConfig mappingConfig) {
-        this.mappingConfig = mappingConfig;
-    }
-
-    public Map<String, MappingConfig> getTableConfig() {
-        return tableConfig;
-    }
-
-    public void setTableConfig(Map<String, MappingConfig> tableConfig) {
-        this.tableConfig = tableConfig;
-    }
-}

client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/monitor/ClickHouseConfigMonitor.java

@@ -1,131 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.clickhouse.monitor;
-
-import java.io.File;
-import java.util.Properties;
-
-import org.apache.commons.io.filefilter.FileFilterUtils;
-import org.apache.commons.io.monitor.FileAlterationListenerAdaptor;
-import org.apache.commons.io.monitor.FileAlterationMonitor;
-import org.apache.commons.io.monitor.FileAlterationObserver;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.alibaba.otter.canal.client.adapter.clickhouse.ClickHouseAdapter;
-import com.alibaba.otter.canal.client.adapter.clickhouse.config.MappingConfig;
-import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader;
-import com.alibaba.otter.canal.client.adapter.support.Util;
-import com.alibaba.otter.canal.client.adapter.support.YamlUtils;
-
-/**
- * @author: Xander
- * @date: Created in 2023/11/10 22:23
- * @email: zhrunxin33@gmail.com
- * @version 1.1.8
- */
-
-public class ClickHouseConfigMonitor {
-
-    private static final Logger   logger      = LoggerFactory.getLogger(ClickHouseConfigMonitor.class);
-
-    private static final String   adapterName = "clickhouse";
-
-    private String                key;
-
-    private ClickHouseAdapter     clickHouseAdapter;
-
-    private Properties            envProperties;
-
-    private FileAlterationMonitor fileMonitor;
-
-    public void init(String key, ClickHouseAdapter clickHouseAdapter, Properties envProperties) {
-        this.key = key;
-        this.clickHouseAdapter = clickHouseAdapter;
-        this.envProperties = envProperties;
-        File confDir = Util.getConfDirPath(adapterName);
-        try {
-            FileAlterationObserver observer = new FileAlterationObserver(confDir,
-                FileFilterUtils.and(FileFilterUtils.fileFileFilter(), FileFilterUtils.suffixFileFilter("yml")));
-            FileListener listener = new FileListener();
-            observer.addListener(listener);
-            fileMonitor = new FileAlterationMonitor(3000, observer);
-            fileMonitor.start();
-
-        } catch (Exception e) {
-            logger.error(e.getMessage(), e);
-        }
-    }
-
-    public void destroy() {
-        try {
-            fileMonitor.stop();
-        } catch (Exception e) {
-            logger.error(e.getMessage(), e);
-        }
-    }
-
-    private class FileListener extends FileAlterationListenerAdaptor {
-
-        @Override
-        public void onFileCreate(File file) {
-            super.onFileCreate(file);
-            try {
-                // 加载新增的配置文件
-                String configContent = MappingConfigsLoader.loadConfig(adapterName + File.separator + file.getName());
-                MappingConfig config = YamlUtils
-                    .ymlToObj(null, configContent, MappingConfig.class, null, envProperties);
-                if (config == null) {
-                    return;
-                }
-                config.validate();
-                boolean result = clickHouseAdapter.addConfig(file.getName(), config);
-                if (result) {
-                    logger.info("Add a new clickhouse mapping config: {} to canal adapter", file.getName());
-                }
-            } catch (Exception e) {
-                logger.error(e.getMessage(), e);
-            }
-        }
-
-        @Override
-        public void onFileChange(File file) {
-            super.onFileChange(file);
-
-            try {
-                if (clickHouseAdapter.getClickHouseMapping().containsKey(file.getName())) {
-                    // 加载配置文件
-                    String configContent = MappingConfigsLoader
-                        .loadConfig(adapterName + File.separator + file.getName());
-                    if (configContent == null) {
-                        onFileDelete(file);
-                        return;
-                    }
-                    MappingConfig config = YamlUtils
-                        .ymlToObj(null, configContent, MappingConfig.class, null, envProperties);
-                    if (config == null) {
-                        return;
-                    }
-                    config.validate();
-                    clickHouseAdapter.updateConfig(file.getName(), config);
-                    logger.info("Change a clickhouse mapping config: {} of canal adapter", file.getName());
-                }
-            } catch (Exception e) {
-                logger.error(e.getMessage(), e);
-            }
-        }
-
-        @Override
-        public void onFileDelete(File file) {
-            super.onFileDelete(file);
-
-            try {
-                if (clickHouseAdapter.getClickHouseMapping().containsKey(file.getName())) {
-                    clickHouseAdapter.deleteConfig(file.getName());
-
-                    logger.info("Delete a clickhouse mapping config: {} of canal adapter", file.getName());
-                }
-            } catch (Exception e) {
-                logger.error(e.getMessage(), e);
-            }
-        }
-    }
-}

client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/service/ClickHouseBatchSyncService.java

@@ -1,555 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.clickhouse.service;
-
-import java.sql.Connection;
-import java.sql.ResultSetMetaData;
-import java.sql.SQLException;
-import java.sql.Types;
-import java.util.*;
-import java.util.concurrent.*;
-import java.util.function.Function;
-import java.util.stream.Collectors;
-
-import org.apache.commons.lang.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.alibaba.druid.pool.DruidDataSource;
-import com.alibaba.fastjson2.JSON;
-import com.alibaba.fastjson2.JSONWriter.Feature;
-import com.alibaba.otter.canal.client.adapter.clickhouse.config.MappingConfig;
-import com.alibaba.otter.canal.client.adapter.clickhouse.config.MappingConfig.DbMapping;
-import com.alibaba.otter.canal.client.adapter.clickhouse.support.BatchExecutor;
-import com.alibaba.otter.canal.client.adapter.clickhouse.support.SingleDml;
-import com.alibaba.otter.canal.client.adapter.clickhouse.support.SyncUtil;
-import com.alibaba.otter.canal.client.adapter.support.Dml;
-import com.alibaba.otter.canal.client.adapter.support.Util;
-
-/**
- * ClickHouse batch synchronize
- *
- * @author: Xander
- * @date: Created in 2023/11/10 22:23
- * @email: zhrunxin33@gmail.com
- * @version 1.1.8
- */
-public class ClickHouseBatchSyncService {
-
-    private static final Logger                         logger  = LoggerFactory.getLogger(ClickHouseBatchSyncService.class);
-
-    private DruidDataSource                             dataSource;
-
-    private Map<String, Map<String, Integer>>           columnsTypeCache;     // Cache of instance.schema.table -> <columnName, jdbcType>
-
-    private Map<MappingConfig, List<SingleDml>>[]       bufferPools;          // Data buffer pool store sync data, List<Dml> dispersed as arrays according to hash value
-
-    private BatchExecutor[]                             batchExecutors;       // Batch Executor
-
-    private BatchExecutor                               alterExecutors;       // Alter Single Executor(update/delete/truncate)
-
-    private ExecutorService[]                           executorThreads;      // Be initialized once
-
-    private ScheduledExecutorService[]                  scheduledExecutors;
-
-    private int                                         threads = 3;          // Default parallel thread count
-    private int                                         batchSize = 1000;
-    private long                                        scheduleTime = 10;
-    private boolean                                     skipDupException;
-
-    public Map<String, Map<String, Integer>> getColumnsTypeCache() {
-        return columnsTypeCache;
-    }
-
-    public ClickHouseBatchSyncService(DruidDataSource dataSource, Integer threads, Integer batchSize, Long scheduleTime, boolean skipDupException){
-        this(dataSource, threads, batchSize, scheduleTime, new ConcurrentHashMap<>(), skipDupException);
-    }
-
-    @SuppressWarnings("unchecked")
-    public ClickHouseBatchSyncService(DruidDataSource dataSource, Integer threads, Integer batchSize, Long scheduleTime, Map<String, Map<String, Integer>> columnsTypeCache,
-                                      boolean skipDupException){
-        this.dataSource = dataSource;
-        this.columnsTypeCache = columnsTypeCache;
-        this.skipDupException = skipDupException;
-        try {
-            if (threads != null) {
-                this.threads = threads;
-            }
-            if (batchSize != null) {
-                this.batchSize = batchSize;
-            }
-            if (scheduleTime != null) {
-                this.scheduleTime = scheduleTime;
-            }
-            this.alterExecutors = new BatchExecutor(dataSource);
-            this.bufferPools = new ConcurrentHashMap[this.threads];
-            this.batchExecutors = new BatchExecutor[this.threads];
-            this.executorThreads = new ExecutorService[this.threads];
-            this.scheduledExecutors = new ScheduledExecutorService[this.threads];
-            for (int i = 0; i < this.threads; i++) {
-                bufferPools[i] = new ConcurrentHashMap<>();
-                batchExecutors[i] = new BatchExecutor(dataSource);
-                executorThreads[i] = Executors.newSingleThreadExecutor();
-                scheduledExecutors[i] = Executors.newSingleThreadScheduledExecutor();
-            }
-            scheduleBatchSync();
-        } catch (Exception e) {
-            throw new RuntimeException(e);
-        }
-    }
-
-    /**
-     * Timing-driven event
-     * start schedule batch sync threadPool
-     */
-    private void scheduleBatchSync() {
-        for (int i = 0; i < scheduledExecutors.length; i++) {
-            int index = i;
-            scheduledExecutors[i].scheduleAtFixedRate(()->{
-                List<Future<Boolean>> futures = new ArrayList<>();
-                for (MappingConfig mappingConfig : bufferPools[index].keySet()) {
-                    List<SingleDml> dmls = bufferPools[index].get(mappingConfig);
-                    if (dmls == null || dmls.isEmpty()) {
-                        return;
-                    }
-                    List<SingleDml> tempDmls;
-                    synchronized (dmls) {
-                        tempDmls = new ArrayList<>(dmls);
-                        dmls.clear();
-                    }
-                    futures.add(executorThreads[index].submit(()->{
-                        try {
-                            insert(batchExecutors[index], mappingConfig, tempDmls);
-                            batchExecutors[index].commit();
-                            return true;
-                        } catch (Exception e){
-                            batchExecutors[index].rollback();
-                            throw new RuntimeException(e);
-                        }
-                    }));
-                }
-            }, 0, scheduleTime, TimeUnit.SECONDS);
-        }
-        logger.info("Schedule batch executors has started successfully!");
-    }
-
-    /**
-     * 批量同步回调
-     *
-     * @param dmls 批量 DML
-     * @param function 回调方法
-     */
-    public void sync(List<Dml> dmls, Function<Dml, Boolean> function) {
-        boolean toExecute = false;
-        for (Dml dml : dmls) {
-            if (!toExecute) {
-                toExecute = function.apply(dml);
-            } else {
-                function.apply(dml);
-            }
-        }
-    }
-
-    /**
-     * Distribute dmls into different partition
-     *
-     * @param mappingConfig {@link com.alibaba.otter.canal.client.adapter.clickhouse.ClickHouseAdapter#mappingConfigCache }
-     * @param dmls received DML
-     */
-    public void sync(Map<String, Map<String, MappingConfig>> mappingConfig, List<Dml> dmls, Properties envProperties) {
-        sync(dmls, dml -> {
-            if (dml.getIsDdl() != null && dml.getIsDdl() && StringUtils.isNotEmpty(dml.getSql())) {
-                // DDL(Cache need to update when DDL was executed)
-            columnsTypeCache.remove(dml.getDestination() + "." + dml.getDatabase() + "." + dml.getTable());
-            return false;
-        } else {
-            // DML
-            String destination = StringUtils.trimToEmpty(dml.getDestination());
-            String groupId = StringUtils.trimToEmpty(dml.getGroupId());
-            String database = dml.getDatabase();
-            String table = dml.getTable();
-            Map<String, MappingConfig> configMap;
-            if (envProperties != null && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) {
-                configMap = mappingConfig.get(destination + "-" + groupId + "_" + database + "-" + table);
-            } else {
-                configMap = mappingConfig.get(destination + "_" + database + "-" + table);
-            }
-
-            if (configMap == null) {
-                return false;
-            }
-
-            if (configMap.values().isEmpty()) {
-                return false;
-            }
-
-            for (MappingConfig config : configMap.values()) {
-                distributeDml(config, dml);
-            }
-            return true;
-        }
-    }   );
-    }
-
-    /**
-     * Dml distributor
-     */
-    private void distributeDml(MappingConfig config, Dml dml) {
-        if (config != null) {
-            try {
-                String type = dml.getType();
-                if (type == null) return;
-
-                if (type.equalsIgnoreCase("INSERT")) {
-                    appendDmlBufferPartition(config, dml);
-                } else {
-                    boolean caseInsensitive = config.getDbMapping().isCaseInsensitive();
-                    List<SingleDml> singleDmls = SingleDml.dml2SingleDmls(dml, caseInsensitive);
-
-                    if (type.equalsIgnoreCase("UPDATE")) {
-                        for (SingleDml singleDml : singleDmls) {
-                            update(alterExecutors, config, singleDml);
-                        }
-                    } else if (type.equalsIgnoreCase("DELETE")) {
-                        for (SingleDml singleDml : singleDmls) {
-                            delete(alterExecutors, config, singleDml);
-                        }
-                    } else if (type.equalsIgnoreCase("TRUNCATE")) {
-                        truncate(alterExecutors, config);
-                    }
-                }
-                if (logger.isDebugEnabled()) {
-                    logger.debug("DML: {}", JSON.toJSONString(dml, Feature.WriteNulls));
-                }
-            } catch (SQLException e) {
-                throw new RuntimeException(e);
-            }
-        }
-    }
-
-    public void appendDmlBufferPartition(MappingConfig config, Dml dml) {
-        boolean caseInsensitive = config.getDbMapping().isCaseInsensitive();
-        List<SingleDml> singleDmls = SingleDml.dml2SingleDmls(dml, caseInsensitive);
-
-        singleDmls.forEach(singleDml -> {
-            int hash = mappingHash(config.getDbMapping());
-            if (!config.getConcurrent()) {
-                hash = 0;
-            }
-            List<SingleDml> dmls = bufferPools[hash].computeIfAbsent(config, k -> new ArrayList<>());
-            synchronized (dmls) {
-                dmls.add(singleDml);
-                logger.info("Append one data into pool, id {}", singleDml.getData().get("id"));
-            }
-            // Check the size of the List, achieve when it reaches the maximum value
-            if (dmls.size() >= batchSize) syncToClickHouse(config, hash);
-        });
-    }
-
-    /**
-     * sync when size of list{@link #bufferPools } reaches the maximum value
-     *
-     * @param config key
-     * @param index parallel thread index
-     */
-    private void syncToClickHouse(MappingConfig config, int index) {
-        List<SingleDml> dmls = bufferPools[index].get(config);
-        logger.info("schema:{} table:{} reaches the maximum value, ready to synchronize, size {}", config.getDbMapping().getDatabase(), config.getDbMapping().getTable(), dmls.size());
-        if (dmls ==null || dmls.isEmpty()) {
-            return;
-        }
-        List<SingleDml> tempDmls;
-        synchronized (dmls) {
-            tempDmls = new ArrayList<>(dmls);
-            dmls.clear();
-        }
-        executorThreads[index].submit(() -> {
-            try {
-                insert(batchExecutors[index], config, tempDmls);
-                batchExecutors[index].commit();
-                return true;
-            } catch (Exception e) {
-                batchExecutors[index].rollback();
-                throw new RuntimeException(e);
-            }
-        });
-    }
-
-    /**
-     * Insert
-     *
-     * @param batchExecutor batch translational executor
-     * @param config corresponding configuration object
-     * @param dmls DMLs
-     */
-    private void insert(BatchExecutor batchExecutor, MappingConfig config, List<SingleDml> dmls) throws SQLException {
-        if (dmls == null || dmls.isEmpty()) {
-            return;
-        }
-        List<SingleDml> clearDmls = dmls.stream().filter(e -> e.getData() != null && !e.getData().isEmpty()).collect(Collectors.toList());
-        if (clearDmls == null || clearDmls.isEmpty()) {
-            return;
-        }
-
-        DbMapping dbMapping = config.getDbMapping();
-        String backtick = SyncUtil.getBacktickByDbType(dataSource.getDbType());
-        Map<String, String> columnsMap = SyncUtil.getColumnsMap(dbMapping, clearDmls.get(0).getData());
-
-        StringBuilder insertSql = new StringBuilder();
-        insertSql.append("INSERT INTO ").append(SyncUtil.getDbTableName(dbMapping, dataSource.getDbType())).append(" (");
-
-        columnsMap.forEach((targetColumnName, srcColumnName) -> insertSql.append(backtick)
-            .append(targetColumnName)
-            .append(backtick)
-            .append(","));
-        int len = insertSql.length();
-        insertSql.delete(len - 1, len).append(") VALUES (");
-        int mapLen = columnsMap.size();
-        for (int i = 0; i < mapLen; i++) {
-            insertSql.append("?,");
-        }
-        len = insertSql.length();
-        insertSql.delete(len - 1, len).append(")");
-
-        Map<String, Integer> ctype = getTargetColumnType(batchExecutor.getConn(), config);
-
-        List<List<Map<String, ?>>> values = new ArrayList<>();
-        boolean flag = false;
-        for (Map.Entry<String, String> entry : columnsMap.entrySet()) {
-            String targetColumnName = entry.getKey();
-            String srcColumnName = entry.getValue();
-            if (srcColumnName == null) {
-                srcColumnName = Util.cleanColumn(targetColumnName);
-            }
-
-            Integer type = ctype.get(Util.cleanColumn(targetColumnName).toLowerCase());
-            if (type == null) {
-                throw new RuntimeException("Target column: " + targetColumnName + " not matched");
-            }
-            for (int i = 0; i < clearDmls.size(); i++) {
-                Map<String, Object> dmlData = clearDmls.get(i).getData();
-                List<Map<String, ?>> item;
-                if (flag == false) {
-                    item = new ArrayList<>();
-                    values.add(item);
-                } else {
-                    item = values.get(i);
-                }
-                Object value = dmlData.get(srcColumnName);
-                BatchExecutor.setValue(item, type, value);
-            }
-            flag = true;
-        }
-
-        try {
-            batchExecutor.batchExecute(insertSql.toString(), values);
-        } catch (SQLException e) {
-            if (skipDupException
-                && (e.getMessage().contains("Duplicate entry") || e.getMessage().startsWith("ORA-00001:"))) {
-                // ignore
-                // TODO 增加更多关系数据库的主键冲突的错误码
-            } else {
-                throw e;
-            }
-        }
-        if (logger.isTraceEnabled()) {
-            logger.trace("Insert into target table, sql: {}", insertSql);
-        }
-
-    }
-
-    /**
-     * 更新操作
-     *
-     * @param config 配置项
-     * @param dml DML数据
-     */
-    private void update(BatchExecutor batchExecutor, MappingConfig config, SingleDml dml) throws SQLException {
-        Map<String, Object> data = dml.getData();
-        if (data == null || data.isEmpty()) {
-            return;
-        }
-
-        Map<String, Object> old = dml.getOld();
-        if (old == null || old.isEmpty()) {
-            return;
-        }
-
-        DbMapping dbMapping = config.getDbMapping();
-        String backtick = SyncUtil.getBacktickByDbType(dataSource.getDbType());
-        Map<String, String> columnsMap = SyncUtil.getColumnsMap(dbMapping, data);
-
-        Map<String, Integer> ctype = getTargetColumnType(batchExecutor.getConn(), config);
-
-        StringBuilder updateSql = new StringBuilder();
-        updateSql.append("ALTER TABLE ").append(SyncUtil.getDbTableName(dbMapping, dataSource.getDbType())).append(" UPDATE ");
-        List<Map<String, ?>> values = new ArrayList<>();
-        boolean hasMatched = false;
-        for (String srcColumnName : old.keySet()) {
-            List<String> targetColumnNames = new ArrayList<>();
-            columnsMap.forEach((targetColumn, srcColumn) -> {
-                if (srcColumnName.equalsIgnoreCase(srcColumn)) {
-                    targetColumnNames.add(targetColumn);
-                }
-            });
-            if (!targetColumnNames.isEmpty()) {
-                hasMatched = true;
-                for (String targetColumnName : targetColumnNames) {
-                    updateSql.append(backtick).append(targetColumnName).append(backtick).append("=?, ");
-                    Integer type = ctype.get(Util.cleanColumn(targetColumnName).toLowerCase());
-                    if (type == null) {
-                        throw new RuntimeException("Target column: " + targetColumnName + " not matched");
-                    }
-                    BatchExecutor.setValue(values, type, data.get(srcColumnName));
-                }
-            }
-        }
-        if (!hasMatched) {
-            logger.warn("Did not matched any columns to update ");
-            return;
-        }
-        int len = updateSql.length();
-        updateSql.delete(len - 2, len).append(" WHERE ");
-
-        // 拼接主键
-        appendCondition(dbMapping, updateSql, ctype, values, data, old);
-        batchExecutor.execute(updateSql.toString(), values);
-        if (logger.isTraceEnabled()) {
-            logger.trace("Alter update target table, sql: {}", updateSql);
-        }
-    }
-
-    /**
-     * 删除操作
-     *
-     * @param config
-     * @param dml
-     */
-    private void delete(BatchExecutor batchExecutor, MappingConfig config, SingleDml dml) throws SQLException {
-        Map<String, Object> data = dml.getData();
-        if (data == null || data.isEmpty()) {
-            return;
-        }
-
-        DbMapping dbMapping = config.getDbMapping();
-        Map<String, Integer> ctype = getTargetColumnType(batchExecutor.getConn(), config);
-
-        StringBuilder sql = new StringBuilder();
-        sql.append("ALTER TABLE ").append(SyncUtil.getDbTableName(dbMapping, dataSource.getDbType())).append(" DELETE WHERE ");
-
-        List<Map<String, ?>> values = new ArrayList<>();
-        // 拼接主键
-        appendCondition(dbMapping, sql, ctype, values, data);
-        batchExecutor.execute(sql.toString(), values);
-        if (logger.isTraceEnabled()) {
-            logger.trace("Alter delete from target table, sql: {}", sql);
-        }
-    }
-
-    /**
-     * truncate操作
-     *
-     * @param config
-     */
-    private void truncate(BatchExecutor batchExecutor, MappingConfig config) throws SQLException {
-        DbMapping dbMapping = config.getDbMapping();
-        StringBuilder sql = new StringBuilder();
-        sql.append("TRUNCATE TABLE ").append(SyncUtil.getDbTableName(dbMapping, dataSource.getDbType()));
-        batchExecutor.execute(sql.toString(), new ArrayList<>());
-        if (logger.isTraceEnabled()) {
-            logger.trace("Truncate target table, sql: {}", sql);
-        }
-    }
-
-    /**
-     * 获取目标字段类型
-     *
-     * @param conn sql connection
-     * @param config 映射配置
-     * @return 字段sqlType
-     */
-    private Map<String, Integer> getTargetColumnType(Connection conn, MappingConfig config) {
-        DbMapping dbMapping = config.getDbMapping();
-        String cacheKey = config.getDestination() + "." + dbMapping.getDatabase() + "." + dbMapping.getTable();
-        Map<String, Integer> columnType = columnsTypeCache.get(cacheKey);
-        if (columnType == null) {
-            synchronized (ClickHouseBatchSyncService.class) {
-                columnType = columnsTypeCache.get(cacheKey);
-                if (columnType == null) {
-                    columnType = new LinkedHashMap<>();
-                    final Map<String, Integer> columnTypeTmp = columnType;
-                    String sql = "SELECT * FROM " + SyncUtil.getDbTableName(dbMapping, dataSource.getDbType()) + " WHERE 1=2";
-                    Util.sqlRS(conn, sql, rs -> {
-                        try {
-                            ResultSetMetaData rsd = rs.getMetaData();
-                            int columnCount = rsd.getColumnCount();
-                            for (int i = 1; i <= columnCount; i++) {
-                                int colType = rsd.getColumnType(i);
-                                // 修复year类型作为date处理时的data truncated问题
-                                if ("YEAR".equals(rsd.getColumnTypeName(i))) {
-                                    colType = Types.VARCHAR;
-                                }
-                                columnTypeTmp.put(rsd.getColumnName(i).toLowerCase(), colType);
-                            }
-                            columnsTypeCache.put(cacheKey, columnTypeTmp);
-                        } catch (SQLException e) {
-                            logger.error(e.getMessage(), e);
-                        }
-                    });
-                }
-            }
-        }
-        return columnType;
-    }
-
-    /**
-     * 拼接主键 where条件
-     */
-    private void appendCondition(MappingConfig.DbMapping dbMapping, StringBuilder sql, Map<String, Integer> ctype,
-                                 List<Map<String, ?>> values, Map<String, Object> d) {
-        appendCondition(dbMapping, sql, ctype, values, d, null);
-    }
-
-    private void appendCondition(MappingConfig.DbMapping dbMapping, StringBuilder sql, Map<String, Integer> ctype,
-                                 List<Map<String, ?>> values, Map<String, Object> d, Map<String, Object> o) {
-        String backtick = SyncUtil.getBacktickByDbType(dataSource.getDbType());
-
-        // 拼接主键
-        for (Map.Entry<String, String> entry : dbMapping.getTargetPk().entrySet()) {
-            String targetColumnName = entry.getKey();
-            String srcColumnName = entry.getValue();
-            if (srcColumnName == null) {
-                srcColumnName = Util.cleanColumn(targetColumnName);
-            }
-            sql.append(backtick).append(targetColumnName).append(backtick).append("=? AND ");
-            Integer type = ctype.get(Util.cleanColumn(targetColumnName).toLowerCase());
-            if (type == null) {
-                throw new RuntimeException("Target column: " + targetColumnName + " not matched");
-            }
-            // 如果有修改主键的情况
-            if (o != null && o.containsKey(srcColumnName)) {
-                BatchExecutor.setValue(values, type, o.get(srcColumnName));
-            } else {
-                BatchExecutor.setValue(values, type, d.get(srcColumnName));
-            }
-        }
-        int len = sql.length();
-        sql.delete(len - 4, len);
-    }
-
-    /**
-     * make sure the same table in one index
-     *
-     * @param dbMapping
-     * @return
-     */
-    private int mappingHash(MappingConfig.DbMapping dbMapping) {
-        int hash = dbMapping.getDatabase().toLowerCase().hashCode() + dbMapping.getTable().toLowerCase().hashCode();
-        hash = Math.abs(hash) % threads;
-        return Math.abs(hash);
-    }
-
-    public void close() {
-        for (int i = 0; i < threads; i++) {
-            executorThreads[i].shutdown();
-        }
-    }
-}

client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/service/ClickHouseEtlService.java

@@ -1,197 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.clickhouse.service;
-
-import java.sql.*;
-import java.util.ArrayList;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicLong;
-
-import javax.sql.DataSource;
-
-import com.alibaba.druid.pool.DruidDataSource;
-import com.alibaba.otter.canal.client.adapter.clickhouse.config.MappingConfig;
-import com.alibaba.otter.canal.client.adapter.clickhouse.config.MappingConfig.DbMapping;
-import com.alibaba.otter.canal.client.adapter.clickhouse.support.SyncUtil;
-import com.alibaba.otter.canal.client.adapter.support.*;
-
-/**
- * ClickHouse ETL 操作业务类
- *
- * @author: Xander
- * @date: Created in 2023/11/10 22:23
- * @email: zhrunxin33@gmail.com
- * @version 1.1.8
- */
-public class ClickHouseEtlService extends AbstractEtlService {
-
-    private DataSource    targetDS;
-    private MappingConfig config;
-
-    public ClickHouseEtlService(DataSource targetDS, MappingConfig config){
-        super("CLICKHOUSE", config);
-        this.targetDS = targetDS;
-        this.config = config;
-    }
-
-    /**
-     * 导入数据
-     */
-    public EtlResult importData(List<String> params) {
-        DbMapping dbMapping = config.getDbMapping();
-        DruidDataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
-        String sql = "SELECT * FROM " + SyncUtil.getDbTableName(dbMapping, dataSource.getDbType());
-        return importData(sql, params);
-    }
-
-    /**
-     * 执行导入
-     */
-    protected boolean executeSqlImport(DataSource srcDS, String sql, List<Object> values,
-                                       AdapterConfig.AdapterMapping mapping, AtomicLong impCount, List<String> errMsg) {
-        try {
-            DbMapping dbMapping = (DbMapping) mapping;
-            Map<String, String> columnsMap = new LinkedHashMap<>();
-            Map<String, Integer> columnType = new LinkedHashMap<>();
-            DruidDataSource dataSource = (DruidDataSource) srcDS;
-            String backtick = SyncUtil.getBacktickByDbType(dataSource.getDbType());
-
-            Util.sqlRS(targetDS,
-                "SELECT * FROM " + SyncUtil.getDbTableName(dbMapping, dataSource.getDbType()) + " LIMIT 1 ",
-                rs -> {
-                    try {
-
-                        ResultSetMetaData rsd = rs.getMetaData();
-                        int columnCount = rsd.getColumnCount();
-                        List<String> columns = new ArrayList<>();
-                        for (int i = 1; i <= columnCount; i++) {
-                            columnType.put(rsd.getColumnName(i).toLowerCase(), rsd.getColumnType(i));
-                            columns.add(rsd.getColumnName(i));
-                        }
-
-                        columnsMap.putAll(SyncUtil.getColumnsMap(dbMapping, columns));
-                        return true;
-                    } catch (Exception e) {
-                        logger.error(e.getMessage(), e);
-                        return false;
-                    }
-                });
-
-            Util.sqlRS(srcDS, sql, values, rs -> {
-                int idx = 1;
-
-                try {
-                    boolean completed = false;
-
-                    StringBuilder insertSql = new StringBuilder();
-                    insertSql.append("INSERT INTO ")
-                        .append(SyncUtil.getDbTableName(dbMapping, dataSource.getDbType()))
-                        .append(" (");
-                    columnsMap.forEach((targetColumnName, srcColumnName) -> insertSql.append(backtick)
-                        .append(targetColumnName)
-                        .append(backtick)
-                        .append(","));
-
-                    int len = insertSql.length();
-                    insertSql.delete(len - 1, len).append(") VALUES (");
-                    int mapLen = columnsMap.size();
-                    for (int i = 0; i < mapLen; i++) {
-                        insertSql.append("?,");
-                    }
-                    len = insertSql.length();
-                    insertSql.delete(len - 1, len).append(")");
-                    logger.info("executeSqlImport sql:{}", insertSql.toString());
-                    try (Connection connTarget = targetDS.getConnection();
-                            PreparedStatement pstmt = connTarget.prepareStatement(insertSql.toString())) {
-                        connTarget.setAutoCommit(false);
-
-                        while (rs.next()) {
-                            completed = false;
-
-                            pstmt.clearParameters();
-
-                            // 删除数据
-                            Map<String, Object> pkVal = new LinkedHashMap<>();
-                            StringBuilder deleteSql = new StringBuilder(
-                                "ALTER TABLE " + SyncUtil.getDbTableName(dbMapping, dataSource.getDbType())
-                                                                        + " DELETE WHERE ");
-                            appendCondition(dbMapping, deleteSql, pkVal, rs, backtick);
-                            try (PreparedStatement pstmt2 = connTarget.prepareStatement(deleteSql.toString())) {
-                                int k = 1;
-                                for (Object val : pkVal.values()) {
-                                    pstmt2.setObject(k++, val);
-                                }
-                                pstmt2.execute();
-                            }
-
-                            int i = 1;
-                            for (Map.Entry<String, String> entry : columnsMap.entrySet()) {
-                                String targetClolumnName = entry.getKey();
-                                String srcColumnName = entry.getValue();
-                                if (srcColumnName == null) {
-                                    srcColumnName = targetClolumnName;
-                                }
-
-                                Integer type = columnType.get(targetClolumnName.toLowerCase());
-                                Object value = rs.getObject(srcColumnName);
-                                if (value != null) {
-                                    SyncUtil.setPStmt(type, pstmt, value, i);
-                                } else {
-                                    pstmt.setNull(i, type);
-                                }
-
-                                i++;
-                            }
-
-                            pstmt.execute();
-                            if (logger.isTraceEnabled()) {
-                                logger.trace("Insert into target table, sql: {}", insertSql);
-                            }
-
-                            if (idx % dbMapping.getCommitBatch() == 0) {
-                                connTarget.commit();
-                                completed = true;
-                            }
-                            idx++;
-                            impCount.incrementAndGet();
-                            if (logger.isDebugEnabled()) {
-                                logger.debug("successful import count:" + impCount.get());
-                            }
-                        }
-                        if (!completed) {
-                            connTarget.commit();
-                        }
-                    }
-
-                } catch (Exception e) {
-                    logger.error(dbMapping.getTable() + " etl failed! ==>" + e.getMessage(), e);
-                    errMsg.add(dbMapping.getTable() + " etl failed! ==>" + e.getMessage());
-                }
-                return idx;
-            });
-            return true;
-        } catch (Exception e) {
-            logger.error(e.getMessage(), e);
-            return false;
-        }
-    }
-
-    /**
-     * 拼接目标表主键where条件
-     */
-    private static void appendCondition(DbMapping dbMapping, StringBuilder sql, Map<String, Object> values,
-                                        ResultSet rs, String backtick) throws SQLException {
-        // 拼接主键
-        for (Map.Entry<String, String> entry : dbMapping.getTargetPk().entrySet()) {
-            String targetColumnName = entry.getKey();
-            String srcColumnName = entry.getValue();
-            if (srcColumnName == null) {
-                srcColumnName = targetColumnName;
-            }
-            sql.append(backtick).append(targetColumnName).append(backtick).append("=? AND ");
-            values.put(targetColumnName, rs.getObject(srcColumnName));
-        }
-        int len = sql.length();
-        sql.delete(len - 4, len);
-    }
-}

client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/service/ClickHouseMirrorDbBatchSyncService.java

@@ -1,180 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.clickhouse.service;
-
-import java.sql.Connection;
-import java.sql.Statement;
-import java.util.ArrayList;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-
-import org.apache.commons.lang.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.alibaba.druid.pool.DruidDataSource;
-import com.alibaba.fastjson2.JSON;
-import com.alibaba.fastjson2.JSONWriter.Feature;
-import com.alibaba.otter.canal.client.adapter.clickhouse.config.MappingConfig;
-import com.alibaba.otter.canal.client.adapter.clickhouse.config.MirrorDbConfig;
-import com.alibaba.otter.canal.client.adapter.clickhouse.support.SyncUtil;
-import com.alibaba.otter.canal.client.adapter.support.Dml;
-
-/**
- * CLICKHOUSE镜像库同步操作业务
- *
- * @author: Xander
- * @date: Created in 2023/11/10 22:23
- * @email: zhrunxin33@gmail.com
- * @version 1.1.8
- */
-public class ClickHouseMirrorDbBatchSyncService {
-
-    private static final Logger         logger = LoggerFactory.getLogger(ClickHouseMirrorDbBatchSyncService.class);
-
-    // 镜像库配置
-    private Map<String, MirrorDbConfig> mirrorDbConfigCache;
-    private DruidDataSource             dataSource;
-    // clickhouseSyncService代理
-    private ClickHouseBatchSyncService  clickHouseBatchSyncService;
-
-    public ClickHouseMirrorDbBatchSyncService(Map<String, MirrorDbConfig> mirrorDbConfigCache,
-                                              DruidDataSource dataSource, Integer threads, Integer batchSize,
-                                              Long scheduleTime, Map<String, Map<String, Integer>> columnsTypeCache,
-                                              boolean skipDupException){
-        this.mirrorDbConfigCache = mirrorDbConfigCache;
-        this.dataSource = dataSource;
-        this.clickHouseBatchSyncService = new ClickHouseBatchSyncService(dataSource,
-            threads,
-            batchSize,
-            scheduleTime,
-            columnsTypeCache,
-            skipDupException);
-    }
-
-    /**
-     * 批量同步方法
-     *
-     * @param dmls 批量 DML，包含DDL
-     */
-    public void sync(List<Dml> dmls) {
-        List<Dml> dmlList = new ArrayList<>();
-        for (Dml dml : dmls) {
-            String destination = StringUtils.trimToEmpty(dml.getDestination());
-            String database = dml.getDatabase();
-            MirrorDbConfig mirrorDbConfig = mirrorDbConfigCache.get(destination + "." + database);
-            if (mirrorDbConfig == null) {
-                continue;
-            }
-            if (mirrorDbConfig.getMappingConfig() == null) {
-                continue;
-            }
-            if (dml.getGroupId() != null && StringUtils.isNotEmpty(mirrorDbConfig.getMappingConfig().getGroupId())) {
-                if (!mirrorDbConfig.getMappingConfig().getGroupId().equals(dml.getGroupId())) {
-                    continue; // 如果groupId不匹配则过滤
-                }
-            }
-
-            if (dml.getIsDdl() != null && dml.getIsDdl() && StringUtils.isNotEmpty(dml.getSql())) {
-                // 确保执行DDL前DML已执行完
-                syncDml(dmlList);
-                dmlList.clear();
-
-                // DDL
-                if (logger.isDebugEnabled()) {
-                    logger.debug("DDL: {}", JSON.toJSONString(dml, Feature.WriteNulls));
-                }
-                executeDdl(mirrorDbConfig, dml);
-                clickHouseBatchSyncService.getColumnsTypeCache()
-                    .remove(destination + "." + database + "." + dml.getTable());
-                mirrorDbConfig.getTableConfig().remove(dml.getTable()); // 删除对应库表配置
-            } else {
-                // DML
-                initMappingConfig(dml.getTable(), mirrorDbConfig.getMappingConfig(), mirrorDbConfig, dml);
-                dmlList.add(dml);
-            }
-        }
-        syncDml(dmlList);
-    }
-
-    /**
-     * 批量同步Dml
-     *
-     * @param dmlList Dml列表，不包含DDL
-     */
-    private void syncDml(List<Dml> dmlList) {
-        if (dmlList == null || dmlList.isEmpty()) {
-            return;
-        }
-        clickHouseBatchSyncService.sync(dmlList, dml -> {
-            MirrorDbConfig mirrorDbConfig = mirrorDbConfigCache.get(dml.getDestination() + "." + dml.getDatabase());
-            if (mirrorDbConfig == null) {
-                return false;
-            }
-            String table = dml.getTable();
-            MappingConfig config = mirrorDbConfig.getTableConfig().get(table);
-
-            if (config == null) {
-                return false;
-            }
-            clickHouseBatchSyncService.appendDmlBufferPartition(config, dml);
-            return true;
-        });
-    }
-
-    /**
-     * 初始化表配置
-     *
-     * @param key 配置key: destination.database.table
-     * @param baseConfigMap db sync config
-     * @param dml DML
-     */
-    private void initMappingConfig(String key, MappingConfig baseConfigMap, MirrorDbConfig mirrorDbConfig, Dml dml) {
-        MappingConfig mappingConfig = mirrorDbConfig.getTableConfig().get(key);
-        if (mappingConfig == null) {
-            // 构造表配置
-            mappingConfig = new MappingConfig();
-            mappingConfig.setDataSourceKey(baseConfigMap.getDataSourceKey());
-            mappingConfig.setDestination(baseConfigMap.getDestination());
-            mappingConfig.setGroupId(baseConfigMap.getGroupId());
-            mappingConfig.setOuterAdapterKey(baseConfigMap.getOuterAdapterKey());
-            mappingConfig.setConcurrent(baseConfigMap.getConcurrent());
-            MappingConfig.DbMapping dbMapping = new MappingConfig.DbMapping();
-            mappingConfig.setDbMapping(dbMapping);
-            dbMapping.setDatabase(dml.getDatabase());
-            dbMapping.setTable(dml.getTable());
-            dbMapping.setTargetDb(dml.getDatabase());
-            dbMapping.setTargetTable(dml.getTable());
-            dbMapping.setMapAll(true);
-            List<String> pkNames = dml.getPkNames();
-            Map<String, String> pkMapping = new LinkedHashMap<>();
-            pkNames.forEach(pkName -> pkMapping.put(pkName, pkName));
-            dbMapping.setTargetPk(pkMapping);
-
-            mirrorDbConfig.getTableConfig().put(key, mappingConfig);
-        }
-    }
-
-    /**
-     * DDL 操作
-     *
-     * @param ddl DDL
-     */
-    private void executeDdl(MirrorDbConfig mirrorDbConfig, Dml ddl) {
-        try (Connection conn = dataSource.getConnection(); Statement statement = conn.createStatement()) {
-            // 替换反引号
-            String sql = ddl.getSql();
-            String backtick = SyncUtil.getBacktickByDbType(dataSource.getDbType());
-            if (!"`".equals(backtick)) {
-                sql = sql.replaceAll("`", backtick);
-            }
-            statement.execute(sql);
-            // 移除对应配置
-            mirrorDbConfig.getTableConfig().remove(ddl.getTable());
-            if (logger.isTraceEnabled()) {
-                logger.trace("Execute DDL sql: {} for database: {}", ddl.getSql(), ddl.getDatabase());
-            }
-        } catch (Exception e) {
-            throw new RuntimeException(e);
-        }
-    }
-}

client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/support/BatchExecutor.java

@@ -1,113 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.clickhouse.support;
-
-import java.io.Closeable;
-import java.sql.Connection;
-import java.sql.PreparedStatement;
-import java.sql.SQLException;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import javax.sql.DataSource;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * sql批量执行器
- *
- * @author rewerma 2018-11-7 下午06:45:49
- * @version 1.0.0
- */
-public class BatchExecutor implements Closeable {
-
-    private static final Logger logger = LoggerFactory.getLogger(BatchExecutor.class);
-
-    private DataSource          dataSource;
-    private Connection          conn;
-    private AtomicInteger       idx    = new AtomicInteger(0);
-
-    public BatchExecutor(DataSource dataSource){
-        this.dataSource = dataSource;
-    }
-
-    public Connection getConn() {
-        if (conn == null) {
-            try {
-                conn = dataSource.getConnection();
-                this.conn.setAutoCommit(false);
-            } catch (SQLException e) {
-                logger.error(e.getMessage(), e);
-            }
-        }
-        return conn;
-    }
-
-    public static void setValue(List<Map<String, ?>> values, int type, Object value) {
-        Map<String, Object> valueItem = new HashMap<>();
-        valueItem.put("type", type);
-        valueItem.put("value", value);
-        values.add(valueItem);
-    }
-
-    public void execute(String sql, List<Map<String, ?>> values) throws SQLException {
-        PreparedStatement pstmt = getConn().prepareStatement(sql);
-        int len = values.size();
-        for (int i = 0; i < len; i++) {
-            int type = (Integer) values.get(i).get("type");
-            Object value = values.get(i).get("value");
-            SyncUtil.setPStmt(type, pstmt, value, i + 1);
-        }
-
-        pstmt.execute();
-        idx.incrementAndGet();
-        pstmt.close();
-    }
-
-    public void batchExecute(String sql, List<List<Map<String, ?>>> batchValues) throws SQLException {
-        PreparedStatement pstmt = getConn().prepareStatement(sql);
-        for (int i = 0; i < batchValues.size(); i++) {
-            List<Map<String, ?>> values = batchValues.get(i);
-            for (int j = 0; j < values.size(); j++) {
-                int type = (Integer) values.get(j).get("type");
-                Object value = values.get(j).get("value");
-                SyncUtil.setPStmt(type, pstmt, value, j + 1);
-            }
-            pstmt.addBatch();
-        }
-
-        pstmt.executeBatch();
-        idx.addAndGet(batchValues.size());
-        pstmt.close();
-    }
-
-    public void commit() throws SQLException {
-        getConn().commit();
-        if (logger.isTraceEnabled()) {
-            logger.trace("Batch executor commit " + idx.get() + " rows");
-        }
-        idx.set(0);
-    }
-
-    public void rollback() throws SQLException {
-        getConn().rollback();
-        if (logger.isTraceEnabled()) {
-            logger.trace("Batch executor rollback " + idx.get() + " rows");
-        }
-        idx.set(0);
-    }
-
-    @Override
-    public void close() {
-        if (conn != null) {
-            try {
-                conn.close();
-            } catch (SQLException e) {
-                logger.error(e.getMessage(), e);
-            } finally {
-                conn = null;
-            }
-        }
-    }
-}

client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/support/SingleDml.java

@@ -1,112 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.clickhouse.support;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-
-import org.springframework.util.LinkedCaseInsensitiveMap;
-
-import com.alibaba.otter.canal.client.adapter.support.Dml;
-
-public class SingleDml {
-
-    private String              destination;
-    private String              database;
-    private String              table;
-    private String              type;
-    private Map<String, Object> data;
-    private Map<String, Object> old;
-
-    public String getDestination() {
-        return destination;
-    }
-
-    public void setDestination(String destination) {
-        this.destination = destination;
-    }
-
-    public String getDatabase() {
-        return database;
-    }
-
-    public void setDatabase(String database) {
-        this.database = database;
-    }
-
-    public String getTable() {
-        return table;
-    }
-
-    public void setTable(String table) {
-        this.table = table;
-    }
-
-    public String getType() {
-        return type;
-    }
-
-    public void setType(String type) {
-        this.type = type;
-    }
-
-    public Map<String, Object> getData() {
-        return data;
-    }
-
-    public void setData(Map<String, Object> data) {
-        this.data = data;
-    }
-
-    public Map<String, Object> getOld() {
-        return old;
-    }
-
-    public void setOld(Map<String, Object> old) {
-        this.old = old;
-    }
-
-    public static List<SingleDml> dml2SingleDmls(Dml dml, boolean caseInsensitive) {
-        List<SingleDml> singleDmls = new ArrayList<>();
-        if (dml.getData() != null) {
-            int size = dml.getData().size();
-            for (int i = 0; i < size; i++) {
-                SingleDml singleDml = new SingleDml();
-                singleDml.setDestination(dml.getDestination());
-                singleDml.setDatabase(dml.getDatabase());
-                singleDml.setTable(dml.getTable());
-                singleDml.setType(dml.getType());
-                Map<String, Object> data = dml.getData().get(i);
-                if (caseInsensitive) {
-                    data = toCaseInsensitiveMap(data);
-                }
-                singleDml.setData(data);
-                if (dml.getOld() != null) {
-                    Map<String, Object> oldData = dml.getOld().get(i);
-                    if (caseInsensitive) {
-                        oldData = toCaseInsensitiveMap(oldData);
-                    }
-                    singleDml.setOld(oldData);
-                }
-                singleDmls.add(singleDml);
-            }
-        } else if ("TRUNCATE".equalsIgnoreCase(dml.getType())) {
-            SingleDml singleDml = new SingleDml();
-            singleDml.setDestination(dml.getDestination());
-            singleDml.setDatabase(dml.getDatabase());
-            singleDml.setTable(dml.getTable());
-            singleDml.setType(dml.getType());
-            singleDmls.add(singleDml);
-        }
-        return singleDmls;
-    }
-
-    public static List<SingleDml> dml2SingleDmls(Dml dml) {
-        return dml2SingleDmls(dml, false);
-    }
-
-    private static <V> LinkedCaseInsensitiveMap<V> toCaseInsensitiveMap(Map<String, V> data) {
-        LinkedCaseInsensitiveMap map = new LinkedCaseInsensitiveMap();
-        map.putAll(data);
-        return map;
-    }
-}

client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/support/SyncUtil.java

@@ -1,293 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.clickhouse.support;
-
-import java.io.Reader;
-import java.io.StringReader;
-import java.math.BigDecimal;
-import java.nio.charset.StandardCharsets;
-import java.sql.*;
-import java.util.Collection;
-import java.util.LinkedHashMap;
-import java.util.Map;
-
-import org.apache.commons.lang.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.alibaba.druid.DbType;
-import com.alibaba.otter.canal.client.adapter.clickhouse.config.MappingConfig;
-import com.alibaba.otter.canal.client.adapter.support.Util;
-
-public class SyncUtil {
-
-    private static final Logger logger = LoggerFactory.getLogger(SyncUtil.class);
-
-    public static Map<String, String> getColumnsMap(MappingConfig.DbMapping dbMapping, Map<String, Object> data) {
-        return getColumnsMap(dbMapping, data.keySet());
-    }
-
-    public static Map<String, String> getColumnsMap(MappingConfig.DbMapping dbMapping, Collection<String> columns) {
-        Map<String, String> columnsMap;
-        if (dbMapping.getMapAll()) {
-            if (dbMapping.getAllMapColumns() != null) {
-                return dbMapping.getAllMapColumns();
-            }
-            columnsMap = new LinkedHashMap<>();
-            for (String srcColumn : columns) {
-                boolean flag = true;
-                if (dbMapping.getTargetColumns() != null) {
-                    for (Map.Entry<String, String> entry : dbMapping.getTargetColumns().entrySet()) {
-                        if (srcColumn.equals(entry.getValue())) {
-                            columnsMap.put(entry.getKey(), srcColumn);
-                            flag = false;
-                            break;
-                        }
-                    }
-                }
-                if (flag) {
-                    columnsMap.put(srcColumn, srcColumn);
-                }
-            }
-            dbMapping.setAllMapColumns(columnsMap);
-        } else {
-            columnsMap = dbMapping.getTargetColumns();
-        }
-        return columnsMap;
-    }
-
-    /**
-     * 设置 preparedStatement
-     *
-     * @param type sqlType
-     * @param pstmt 需要设置的preparedStatement
-     * @param value 值
-     * @param i 索引号
-     */
-    public static void setPStmt(int type, PreparedStatement pstmt, Object value, int i) throws SQLException {
-        switch (type) {
-            case Types.BIT:
-            case Types.BOOLEAN:
-                if (value instanceof Boolean) {
-                    pstmt.setBoolean(i, (Boolean) value);
-                } else if (value instanceof String) {
-                    boolean v = !value.equals("0");
-                    pstmt.setBoolean(i, v);
-                } else if (value instanceof Number) {
-                    boolean v = ((Number) value).intValue() != 0;
-                    pstmt.setBoolean(i, v);
-                } else {
-                    pstmt.setNull(i, type);
-                }
-                break;
-            case Types.CHAR:
-            case Types.NCHAR:
-            case Types.VARCHAR:
-            case Types.LONGVARCHAR:
-                if (value instanceof String) {
-                    pstmt.setString(i, (String) value);
-                } else if (value == null) {
-                    pstmt.setNull(i, type);
-                } else {
-                    pstmt.setString(i, value.toString());
-                }
-                break;
-            case Types.TINYINT:
-                // 向上提升一级，处理unsigned情况
-                if (value instanceof Number) {
-                    pstmt.setShort(i, ((Number) value).shortValue());
-                } else if (value instanceof String) {
-                    pstmt.setShort(i, Short.parseShort((String) value));
-                } else {
-                    pstmt.setNull(i, type);
-                }
-                break;
-            case Types.SMALLINT:
-                if (value instanceof Number) {
-                    pstmt.setInt(i, ((Number) value).intValue());
-                } else if (value instanceof String) {
-                    pstmt.setInt(i, Integer.parseInt((String) value));
-                } else {
-                    pstmt.setNull(i, type);
-                }
-                break;
-            case Types.INTEGER:
-                if (value instanceof Number) {
-                    pstmt.setLong(i, ((Number) value).longValue());
-                } else if (value instanceof String) {
-                    pstmt.setLong(i, Long.parseLong((String) value));
-                } else {
-                    pstmt.setNull(i, type);
-                }
-                break;
-            case Types.BIGINT:
-                if (value instanceof Number) {
-                    pstmt.setBigDecimal(i, new BigDecimal(value.toString()));
-                } else if (value instanceof String) {
-                    pstmt.setBigDecimal(i, new BigDecimal(value.toString()));
-                } else {
-                    pstmt.setNull(i, type);
-                }
-                break;
-            case Types.DECIMAL:
-            case Types.NUMERIC:
-                if (value instanceof BigDecimal) {
-                    pstmt.setBigDecimal(i, (BigDecimal) value);
-                } else if (value instanceof Byte) {
-                    pstmt.setInt(i, ((Byte) value).intValue());
-                } else if (value instanceof Short) {
-                    pstmt.setInt(i, ((Short) value).intValue());
-                } else if (value instanceof Integer) {
-                    pstmt.setInt(i, (Integer) value);
-                } else if (value instanceof Long) {
-                    pstmt.setLong(i, (Long) value);
-                } else if (value instanceof Float) {
-                    pstmt.setBigDecimal(i, new BigDecimal((float) value));
-                } else if (value instanceof Double) {
-                    pstmt.setBigDecimal(i, new BigDecimal((double) value));
-                } else if (value != null) {
-                    pstmt.setBigDecimal(i, new BigDecimal(value.toString()));
-                } else {
-                    pstmt.setNull(i, type);
-                }
-                break;
-            case Types.REAL:
-                if (value instanceof Number) {
-                    pstmt.setFloat(i, ((Number) value).floatValue());
-                } else if (value instanceof String) {
-                    pstmt.setFloat(i, Float.parseFloat((String) value));
-                } else {
-                    pstmt.setNull(i, type);
-                }
-                break;
-            case Types.FLOAT:
-            case Types.DOUBLE:
-                if (value instanceof Number) {
-                    pstmt.setDouble(i, ((Number) value).doubleValue());
-                } else if (value instanceof String) {
-                    pstmt.setDouble(i, Double.parseDouble((String) value));
-                } else {
-                    pstmt.setNull(i, type);
-                }
-                break;
-            case Types.BINARY:
-            case Types.VARBINARY:
-            case Types.LONGVARBINARY:
-            case Types.BLOB:
-                if (value instanceof Blob) {
-                    pstmt.setBlob(i, (Blob) value);
-                } else if (value instanceof byte[]) {
-                    pstmt.setBytes(i, (byte[]) value);
-                } else if (value instanceof String) {
-                    pstmt.setBytes(i, ((String) value).getBytes(StandardCharsets.ISO_8859_1));
-                } else {
-                    pstmt.setNull(i, type);
-                }
-                break;
-            case Types.CLOB:
-                if (value instanceof Clob) {
-                    pstmt.setClob(i, (Clob) value);
-                } else if (value instanceof byte[]) {
-                    pstmt.setBytes(i, (byte[]) value);
-                } else if (value instanceof String) {
-                    Reader clobReader = new StringReader((String) value);
-                    pstmt.setCharacterStream(i, clobReader);
-                } else {
-                    pstmt.setNull(i, type);
-                }
-                break;
-            case Types.DATE:
-                if (value instanceof Date) {
-                    pstmt.setDate(i, (Date) value);
-                } else if (value instanceof java.util.Date) {
-                    pstmt.setDate(i, new Date(((java.util.Date) value).getTime()));
-                } else if (value instanceof String) {
-                    String v = (String) value;
-                    if (!v.startsWith("0000-00-00")) {
-                        java.util.Date date = Util.parseDate(v);
-                        if (date != null) {
-                            pstmt.setDate(i, new Date(date.getTime()));
-                        } else {
-                            pstmt.setNull(i, type);
-                        }
-                    } else {
-                        pstmt.setObject(i, value);
-                    }
-                } else {
-                    pstmt.setNull(i, type);
-                }
-                break;
-            case Types.TIME:
-                if (value instanceof Time) {
-                    pstmt.setTime(i, (Time) value);
-                } else if (value instanceof java.util.Date) {
-                    pstmt.setTime(i, new Time(((java.util.Date) value).getTime()));
-                } else if (value instanceof String) {
-                    String v = (String) value;
-                    java.util.Date date = Util.parseDate(v);
-                    if (date != null) {
-                        pstmt.setTime(i, new Time(date.getTime()));
-                    } else {
-                        pstmt.setNull(i, type);
-                    }
-                } else {
-                    pstmt.setNull(i, type);
-                }
-                break;
-            case Types.TIMESTAMP:
-                if (value instanceof Timestamp) {
-                    pstmt.setTimestamp(i, (Timestamp) value);
-                } else if (value instanceof java.util.Date) {
-                    pstmt.setTimestamp(i, new Timestamp(((java.util.Date) value).getTime()));
-                } else if (value instanceof String) {
-                    String v = (String) value;
-                    if (!v.startsWith("0000-00-00")) {
-                        java.util.Date date = Util.parseDate(v);
-                        if (date != null) {
-                            pstmt.setTimestamp(i, new Timestamp(date.getTime()));
-                        } else {
-                            pstmt.setNull(i, type);
-                        }
-                    } else {
-                        pstmt.setObject(i, value);
-                    }
-                } else {
-                    pstmt.setNull(i, type);
-                }
-                break;
-            default:
-                pstmt.setObject(i, value, type);
-        }
-    }
-
-    public static String getDbTableName(MappingConfig.DbMapping dbMapping, String dbType) {
-        String result = "";
-        String backtick = getBacktickByDbType(dbType);
-        if (StringUtils.isNotEmpty(dbMapping.getTargetDb())) {
-            result += (backtick + dbMapping.getTargetDb() + backtick + ".");
-        }
-        result += (backtick + dbMapping.getTargetTable() + backtick);
-        return result;
-    }
-
-    /**
-     * 根据DbType返回反引号或空字符串
-     *
-     * @param dbTypeName DbType名称
-     * @return 反引号或空字符串
-     */
-    public static String getBacktickByDbType(String dbTypeName) {
-        DbType dbType = DbType.of(dbTypeName);
-        if (dbType == null) {
-            dbType = DbType.other;
-        }
-
-        // 只有当dbType为MySQL/MariaDB或OceanBase时返回反引号
-        switch (dbType) {
-            case mysql:
-            case mariadb:
-            case oceanbase:
-                return "`";
-            default:
-                return "";
-        }
-    }
-}

client-adapter/clickhouse/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.OuterAdapter

@@ -1 +0,0 @@
-clickhouse=com.alibaba.otter.canal.client.adapter.clickhouse.ClickHouseAdapter

client-adapter/clickhouse/src/main/resources/clickhouse/mytest_user.yml

@@ -1,31 +0,0 @@
-dataSourceKey: defaultDS
-destination: example
-groupId: g1
-outerAdapterKey: clickhouse1
-concurrent: true
-dbMapping:
-  database: mytest
-  table: user
-  targetTable: mytest.user
-  targetPk:
-    id: id
-#  mapAll: true
-  targetColumns:
-    id:
-    name:
-    role_id:
-    c_time:
-    test1:
-  etlCondition: "where c_time>={}"
-  commitBatch: 3000 # 批量提交的大小
-
-
-## Mirror schema synchronize config
-#dataSourceKey: defaultDS
-#destination: example
-#groupId: g1
-#outerAdapterKey: mysql1
-#concurrent: true
-#dbMapping:
-#  mirrorDb: true
-#  database: mytest

client-adapter/clickhouse/src/test/java/com/alibaba/otter/canal/client/adapter/clickhouse/ClickHouseBatchSyncServiceTest.java

@@ -1,109 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.clickhouse;
-
-import java.util.*;
-
-import org.junit.Before;
-import org.junit.Ignore;
-import org.junit.Test;
-
-import com.alibaba.otter.canal.client.adapter.clickhouse.sync.Common;
-import com.alibaba.otter.canal.client.adapter.support.Dml;
-
-@Ignore
-public class ClickHouseBatchSyncServiceTest {
-
-    private ClickHouseAdapter clickHouseAdapter;
-
-    @Before
-    public void init() {
-        clickHouseAdapter = Common.init();
-    }
-
-    @Test
-    public void insert() throws InterruptedException {
-        Dml dml = new Dml();
-        dml.setDestination("example");
-        dml.setTs(new Date().getTime());
-        dml.setType("INSERT");
-        dml.setDatabase("mytest");
-        dml.setTable("user");
-        List<Map<String, Object>> dataList = new ArrayList<>();
-        Map<String, Object> data = new LinkedHashMap<>();
-        dataList.add(data);
-        data.put("id", 1L);
-        data.put("name", "Eric");
-        data.put("role_id", 1L);
-        data.put("c_time", new Date());
-        data.put("test1", "sdfasdfawe中国asfwef");
-        dml.setData(dataList);
-
-        clickHouseAdapter.sync(Collections.singletonList(dml));
-        Thread.sleep(10000L);
-    }
-
-    @Test
-    public void update() {
-        Dml dml = new Dml();
-        dml.setDestination("example");
-        dml.setTs(new Date().getTime());
-        dml.setType("UPDATE");
-        dml.setDatabase("mytest");
-        dml.setTable("user");
-        List<Map<String, Object>> dataList = new ArrayList<>();
-        Map<String, Object> data = new LinkedHashMap<>();
-        dataList.add(data);
-        data.put("id", 1L);
-        data.put("name", "Eric2");
-        dml.setData(dataList);
-        List<Map<String, Object>> oldList = new ArrayList<>();
-        Map<String, Object> old = new LinkedHashMap<>();
-        oldList.add(old);
-        old.put("name", "Eric");
-        dml.setOld(oldList);
-
-        clickHouseAdapter.sync(Collections.singletonList(dml));
-    }
-
-    @Test
-    public void delete() {
-        Dml dml = new Dml();
-        dml.setDestination("example");
-        dml.setTs(new Date().getTime());
-        dml.setType("DELETE");
-        dml.setDatabase("mytest");
-        dml.setTable("user");
-        List<Map<String, Object>> dataList = new ArrayList<>();
-        Map<String, Object> data = new LinkedHashMap<>();
-        dataList.add(data);
-        data.put("id", 1L);
-        data.put("name", "Eric2");
-        dml.setData(dataList);
-
-        clickHouseAdapter.sync(Collections.singletonList(dml));
-    }
-
-    @Test
-    public void truncate() throws InterruptedException {
-        Dml dml = new Dml();
-        dml.setDestination("example");
-        dml.setTs(new Date().getTime());
-        dml.setType("TRUNCATE");
-        dml.setDatabase("mytest");
-        dml.setTable("user");
-        List<Map<String, Object>> dataList = new ArrayList<>();
-        Map<String, Object> data = new LinkedHashMap<>();
-        dataList.add(data);
-        data.put("id", 1L);
-        data.put("name", "Eric2");
-        dml.setData(dataList);
-        List<Map<String, Object>> oldList = new ArrayList<>();
-        Map<String, Object> old = new LinkedHashMap<>();
-        oldList.add(old);
-        old.put("name", "Eric");
-        dml.setOld(oldList);
-
-        clickHouseAdapter.sync(Collections.singletonList(dml));
-        Thread.sleep(1000L);
-
-    }
-}

client-adapter/clickhouse/src/test/java/com/alibaba/otter/canal/client/adapter/clickhouse/ClickHouseBatchSyncThreadSafeTest.java

@@ -1,81 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.clickhouse;
-
-import java.util.*;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import org.junit.Before;
-import org.junit.Ignore;
-import org.junit.Test;
-
-import com.alibaba.otter.canal.client.adapter.clickhouse.sync.Common;
-import com.alibaba.otter.canal.client.adapter.support.Dml;
-
-import ch.qos.logback.classic.Level;
-
-/**
- * @author: Xander
- * @date: Created in 2023/11/13 22:27
- * @email: zhrunxin33@gmail.com
- * @description: Testing thread safe
- */
-
-@Ignore
-public class ClickHouseBatchSyncThreadSafeTest {
-
-    private ClickHouseAdapter clickHouseAdapter;
-
-    private ExecutorService   executorService;
-
-    private String[]          operations = new String[] { "INSERT", "UPDATE" };
-
-    private String[]          tables     = new String[] { "user", "customer" };
-
-    @Before
-    public void init() {
-        clickHouseAdapter = Common.init();
-        Common.setLogLevel(Level.INFO);
-        executorService = Executors.newFixedThreadPool(5);
-    }
-
-    @Test
-    public void test01() throws InterruptedException, ExecutionException {
-        ArrayList<Future> list = new ArrayList();
-        AtomicInteger count = new AtomicInteger();
-        for (int i = 0; i < 10; i++) {
-            list.add(executorService.submit(() -> {
-                for (int j = 0; j < 300; j++) {
-                    Random random = new Random();
-                    int cou = count.incrementAndGet();
-                    // test insert
-                    String dmlType = operations[random.nextInt(1)];
-                    Dml dml = new Dml();
-                    dml.setDestination("example");
-                    dml.setTs(new Date().getTime());
-                    dml.setType(dmlType);
-                    dml.setDatabase("mytest");
-                    dml.setTable(tables[(int) Math.round(Math.random())]);
-                    List<Map<String, Object>> dataList = new ArrayList<>();
-                    Map<String, Object> data = new LinkedHashMap<>();
-                    dataList.add(data);
-                    data.put("id", cou);
-                    data.put("name", "Eric" + cou);
-                    data.put("role_id", cou);
-                    data.put("c_time", new Date());
-                    data.put("test1", "sdfasdfawe中国asfwef");
-                    dml.setData(dataList);
-                    clickHouseAdapter.sync(Collections.singletonList(dml));
-                }
-            }));
-
-        }
-        for (Future future : list) {
-            future.get();
-        }
-        Thread.sleep(10000L); // waiting multiple threads execute successfully.
-    }
-
-}

client-adapter/clickhouse/src/test/java/com/alibaba/otter/canal/client/adapter/clickhouse/TestConstant.java

@@ -1,37 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.clickhouse;
-
-import java.sql.SQLException;
-
-import com.alibaba.druid.pool.DruidDataSource;
-
-public class TestConstant {
-
-    public final static String          jdbcUrl      = "jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true";
-    public final static String          jdbcUser     = "root";
-    public final static String          jdbcPassword = "121212";
-
-    public final static DruidDataSource dataSource;
-
-    static {
-        dataSource = new DruidDataSource();
-        dataSource.setDriverClassName("com.mysql.jdbc.Driver");
-        dataSource.setUrl(jdbcUrl);
-        dataSource.setUsername(jdbcUser);
-        dataSource.setPassword(jdbcPassword);
-        dataSource.setInitialSize(1);
-        dataSource.setMinIdle(1);
-        dataSource.setMaxActive(1);
-        dataSource.setMaxWait(60000);
-        dataSource.setTimeBetweenEvictionRunsMillis(60000);
-        dataSource.setMinEvictableIdleTimeMillis(300000);
-        dataSource.setPoolPreparedStatements(false);
-        dataSource.setMaxPoolPreparedStatementPerConnectionSize(20);
-        dataSource.setValidationQuery("select 1");
-        try {
-            dataSource.init();
-        } catch (SQLException e) {
-            e.printStackTrace();
-        }
-    }
-
-}

client-adapter/clickhouse/src/test/java/com/alibaba/otter/canal/client/adapter/clickhouse/sync/Common.java

@@ -1,51 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.clickhouse.sync;
-
-import java.util.HashMap;
-import java.util.Map;
-
-import org.slf4j.LoggerFactory;
-
-import com.alibaba.otter.canal.client.adapter.clickhouse.ClickHouseAdapter;
-import com.alibaba.otter.canal.client.adapter.clickhouse.TestConstant;
-import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
-import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig;
-
-import ch.qos.logback.classic.Level;
-import ch.qos.logback.classic.Logger;
-
-/**
- * @author: Xander
- * @date: Created in 2023/11/13 0:16
- * @email: zhrunxin33@gmail.com @description：
- */
-public class Common {
-
-    public static ClickHouseAdapter init() {
-        DatasourceConfig.DATA_SOURCES.put("defaultDS", TestConstant.dataSource);
-
-        OuterAdapterConfig outerAdapterConfig = new OuterAdapterConfig();
-        outerAdapterConfig.setName("clickhouse");
-        outerAdapterConfig.setKey("clickhouse1");
-        Map<String, String> properties = new HashMap<>();
-        properties.put("jdbc.driveClassName", "ru.yandex.clickhouse.ClickHouseDriver");
-        properties.put("jdbc.url", "jdbc:clickhouse://127.0.0.1:8123/default");
-        properties.put("jdbc.username", "default");
-        properties.put("jdbc.password", "123456");
-        outerAdapterConfig.setProperties(properties);
-
-        ClickHouseAdapter adapter = new ClickHouseAdapter();
-        adapter.init(outerAdapterConfig, null);
-        return adapter;
-    }
-
-    /**
-     * set global log level
-     *
-     * @param logLevel
-     */
-    public static void setLogLevel(Level logLevel) {
-        ch.qos.logback.classic.Logger logger = (ch.qos.logback.classic.Logger) LoggerFactory
-            .getLogger(Logger.ROOT_LOGGER_NAME);
-        logger.setLevel(logLevel);
-    }
-}

client-adapter/clickhouse/src/test/resources/clickhouse/mytest_customer.yml

@@ -1,31 +0,0 @@
-dataSourceKey: defaultDS
-destination: example
-groupId: g1
-outerAdapterKey: clickhouse1
-concurrent: true
-dbMapping:
-  database: mytest
-  table: customer
-  targetTable: mytest.customer
-  targetPk:
-    id: id
-#  mapAll: true
-  targetColumns:
-    id:
-    name:
-    role_id:
-    c_time:
-    test1:
-  etlCondition: "where c_time>={}"
-  commitBatch: 3000 # 批量提交的大小
-
-
-## Mirror schema synchronize config
-#dataSourceKey: defaultDS
-#destination: example
-#groupId: g1
-#outerAdapterKey: mysql1
-#concurrent: true
-#dbMapping:
-#  mirrorDb: true
-#  database: mytest

client-adapter/clickhouse/src/test/resources/clickhouse/mytest_user.yml

@@ -1,31 +0,0 @@
-dataSourceKey: defaultDS
-destination: example
-groupId: g1
-outerAdapterKey: clickhouse1
-concurrent: true
-dbMapping:
-  database: mytest
-  table: user
-  targetTable: mytest.user
-  targetPk:
-    id: id
-#  mapAll: true
-  targetColumns:
-    id:
-    name:
-    role_id:
-    c_time:
-    test1:
-  etlCondition: "where c_time>={}"
-  commitBatch: 3000 # 批量提交的大小
-
-
-## Mirror schema synchronize config
-#dataSourceKey: defaultDS
-#destination: example
-#groupId: g1
-#outerAdapterKey: mysql1
-#concurrent: true
-#dbMapping:
-#  mirrorDb: true
-#  database: mytest

client-adapter/common/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.8-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

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

@@ -178,7 +178,8 @@ public class Util {
                             // ignore
                         }
                     }
-            });
+                }
+        );
     }
 
     public static ThreadPoolExecutor newSingleDaemonThreadExecutor(long keepAliveTime) {
@@ -326,56 +327,4 @@ public class Util {
 
         return null;
     }
-
-    /**
-     * Check if the time string has millisecond or microsecond units
-     * 
-     * @param timeStr time string
-     * @return boolean
-     */
-    public static boolean isAccuracyOverSecond(String timeStr) {
-        if (StringUtils.isEmpty(timeStr)) {
-            return false;
-        }
-        String[] times = StringUtils.split(timeStr, ".");
-        return times.length > 1 && !times[times.length - 1].isEmpty();
-    }
-
-    /**
-     * Check if the datetime string has microsecond or nanosecond units
-     * 
-     * @param datetimeStr datetime string
-     * @return boolean
-     */
-    public static boolean isAccuracyOverMillisecond(String datetimeStr) {
-        if (StringUtils.isEmpty(datetimeStr)) {
-            return false;
-        }
-        String[] times = StringUtils.split(datetimeStr, ".");
-        return times.length > 1 && times[times.length - 1].length() > 3;
-    }
-
-    /**
-     * MySQL has fractional seconds support for TIME, DATETIME, and TIMESTAMP
-     * values, with up to microseconds (6 digits) precision. ISO-8601 standard
-     * format is with up to nanoseconds (9 digits) precision, which is sufficient
-     * for storing MySQL time-related data.
-     * 
-     * @param datetimeStr datetime string
-     * @return LocalDateTime
-     */
-    public static LocalDateTime parseISOLocalDateTime(String datetimeStr) {
-        if (StringUtils.isEmpty(datetimeStr)) {
-            return null;
-        }
-        datetimeStr = datetimeStr.trim();
-        try {
-            // Replace SPACE in middle with `T` to comply with ISO-8601 standard format
-            return LocalDateTime.parse(datetimeStr.replace(" ", "T"));
-        } catch (Exception e) {
-            logger.error("Convert datetime string to ISOLocalDateTime fail:{}", datetimeStr, e);
-        }
-        return null;
-    }
-
 }

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

@@ -4,11 +4,9 @@ import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.charset.StandardCharsets;
-import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
-
 import org.apache.commons.lang.StringUtils;
 import org.springframework.boot.context.properties.bind.Bindable;
 import org.springframework.boot.context.properties.bind.Binder;
@@ -20,7 +18,6 @@ import org.springframework.core.env.PropertySource;
 import org.springframework.core.io.ByteArrayResource;
 import org.springframework.core.io.ClassPathResource;
 import org.springframework.core.io.Resource;
-import org.springframework.util.PropertyPlaceholderHelper;
 
 public class YamlUtils {
 
@@ -90,13 +87,7 @@ public class YamlUtils {
             }
             
             ConfigurationPropertySource sources = new MapConfigurationPropertySource(properties);
-            PropertyPlaceholderHelper propertyPlaceholderHelper = new PropertyPlaceholderHelper("${", "}");
+            Binder binder = new Binder(sources);
-            Binder binder = new Binder(Arrays.asList(sources), value -> {
-                if (value instanceof String) {
-                    return propertyPlaceholderHelper.replacePlaceholders((String) value, baseProperties);
-                }
-                return value;
-            });
             return binder.bind(prefix, Bindable.of(clazz)).get();
         } catch (Exception e) {
             throw new RuntimeException(e);

client-adapter/es6x/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.8-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

client-adapter/es7x/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.8-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

client-adapter/es7x/src/main/java/com/alibaba/otter/canal/client/adapter/es7x/support/ES7xTemplate.java

@@ -2,7 +2,11 @@ package com.alibaba.otter.canal.client.adapter.es7x.support;
 
 import java.sql.ResultSet;
 import java.sql.SQLException;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
@@ -314,15 +318,11 @@ public class ES7xTemplate implements ESTemplate {
         Object resultIdVal = null;
         for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
             ColumnItem columnItem = fieldItem.getColumnItems().iterator().next();
-            if (columnItem.getOwner() == null || columnItem.getColumnName() == null) {
-                continue;
-            }
-
             if (!columnItem.getOwner().equals(owner)) {
                 continue;
             }
-
             String columnName = columnItem.getColumnName();
+
             if (fieldItem.getFieldName().equals(idFieldName)) {
                 resultIdVal = getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName);
             }

client-adapter/es8x/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.8-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

client-adapter/es8x/src/main/java/com/alibaba/otter/canal/client/adapter/es8x/support/ES8xTemplate.java

@@ -65,9 +65,8 @@ public class ES8xTemplate implements ESTemplate {
         if (mapping.getId() != null) {
             String parentVal = (String) esFieldData.remove("$parent_routing");
             if (mapping.isUpsert()) {
-                ESUpdateRequest updateRequest = esConnection.new ES8xUpdateRequest(mapping.getIndex(), pkVal.toString())
+                ESUpdateRequest updateRequest = esConnection.new ES8xUpdateRequest(mapping.getIndex(),
-                    .setDoc(esFieldData)
+                    pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true);
-                    .setDocAsUpsert(true);
                 if (StringUtils.isNotEmpty(parentVal)) {
                     updateRequest.setRouting(parentVal);
                 }

client-adapter/es8x/src/main/java/com/alibaba/otter/canal/client/adapter/es8x/support/ESConnection.java

@@ -1,29 +1,18 @@
 package com.alibaba.otter.canal.client.adapter.es8x.support;
 
 import java.io.IOException;
-import java.io.InputStream;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.net.UnknownHostException;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.security.KeyStore;
-import java.security.cert.Certificate;
-import java.security.cert.CertificateFactory;
 import java.util.Arrays;
 import java.util.Map;
 
-import javax.net.ssl.SSLContext;
-
 import org.apache.commons.lang.StringUtils;
 import org.apache.http.HttpHost;
 import org.apache.http.auth.AuthScope;
 import org.apache.http.auth.UsernamePasswordCredentials;
 import org.apache.http.client.CredentialsProvider;
 import org.apache.http.impl.client.BasicCredentialsProvider;
-import org.apache.http.ssl.SSLContextBuilder;
-import org.apache.http.ssl.SSLContexts;
 import org.elasticsearch.action.bulk.BulkItemResponse;
 import org.elasticsearch.action.bulk.BulkRequest;
 import org.elasticsearch.action.bulk.BulkRequestBuilder;
@@ -62,47 +51,6 @@ public class ESConnection {
     private RestHighLevelClient restHighLevelClient;
 
     public ESConnection(String[] hosts, Map<String, String> properties) throws UnknownHostException{
-        String caPath = properties.get("security.ca.path");
-        if (StringUtils.isNotEmpty(caPath)) {
-            connectEsWithCa(hosts, properties, caPath);
-        } else {
-            connectEsWithoutCa(hosts, properties);
-        }
-    }
-
-    private void connectEsWithCa(String[] hosts, Map<String, String> properties, String caPath) {
-        Path caCertificatePath = Paths.get(caPath);
-        try (InputStream is = Files.newInputStream(caCertificatePath)) {
-            CertificateFactory factory = CertificateFactory.getInstance("X.509");
-            Certificate trustedCa = factory.generateCertificate(is);
-            KeyStore trustStore = KeyStore.getInstance("pkcs12");
-            trustStore.load(null, null);
-            trustStore.setCertificateEntry("ca", trustedCa);
-            SSLContextBuilder sslContextBuilder = SSLContexts.custom().loadTrustMaterial(trustStore, null);
-            final SSLContext sslContext = sslContextBuilder.build();
-
-            HttpHost[] httpHosts = Arrays.stream(hosts).map(this::createHttpHost).toArray(HttpHost[]::new);
-            RestClientBuilder restClientBuilder = RestClient.builder(httpHosts);
-            String nameAndPwd = properties.get("security.auth");
-            if (StringUtils.isNotEmpty(nameAndPwd) && nameAndPwd.contains(":")) {
-                String[] nameAndPwdArr = nameAndPwd.split(":");
-                final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
-                credentialsProvider.setCredentials(AuthScope.ANY,
-                    new UsernamePasswordCredentials(nameAndPwdArr[0], nameAndPwdArr[1]));
-                restClientBuilder.setHttpClientConfigCallback(httpClientBuilder -> {
-                    httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
-                    return httpClientBuilder.setSSLContext(sslContext);
-                });
-            }
-            restHighLevelClient = new RestHighLevelClientBuilder(restClientBuilder.build())
-                .setApiCompatibilityMode(true)
-                .build();
-        } catch (Exception e) {
-            throw new RuntimeException(e);
-        }
-    }
-
-    private void connectEsWithoutCa(String[] hosts, Map<String, String> properties) {
         HttpHost[] httpHosts = Arrays.stream(hosts).map(this::createHttpHost).toArray(HttpHost[]::new);
         RestClientBuilder restClientBuilder = RestClient.builder(httpHosts);
         String nameAndPwd = properties.get("security.auth");
@@ -332,7 +280,9 @@ public class ESConnection {
         private BulkRequest        bulkRequest;
 
         public ES8xBulkRequest(){
+
             bulkRequest = new BulkRequest();
+
         }
 
         public void resetBulk() {

client-adapter/escore/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.8-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

client-adapter/hbase/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.8-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

client-adapter/kudu/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.8-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

client-adapter/kudu/src/main/java/com/alibaba/otter/canal/client/adapter/kudu/service/KuduEtlService.java

@@ -2,7 +2,11 @@ package com.alibaba.otter.canal.client.adapter.kudu.service;
 
 import java.sql.ResultSetMetaData;
 import java.sql.SQLException;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicLong;
 
 import javax.sql.DataSource;
@@ -41,13 +45,13 @@ public class KuduEtlService extends AbstractEtlService {
         boolean flag = kuduTemplate.tableExists(kuduMapping.getTargetTable());
         // 表不存在，停止导入
         if (!flag) {
-            logger.info("{} is don't exist , please check your kudu table !", kuduMapping.getTargetTable());
+            logger.info("{} is don't hava,please check your kudu table !", kuduMapping.getTargetTable());
             errMsg.add(kuduMapping.getTargetTable() + " is don't hava,please check your kudu table !");
             etlResult.setErrorMessage(Joiner.on("\n").join(errMsg));
             return etlResult;
         }
         logger.info("{} etl is starting!", kuduMapping.getTargetTable());
-        String sql = "SELECT * FROM " + SyncUtil.getDbTableName(kuduMapping);
+        String sql = "SELECT * FROM " + kuduMapping.getDatabase() + "." + kuduMapping.getTable();
         return importData(sql, params);
     }
 

client-adapter/launcher/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.8-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>
@@ -74,10 +74,6 @@
             <artifactId>hbase-shaded-client</artifactId>
             <scope>runtime</scope>
         </dependency>
-        <dependency>
-            <groupId>ru.yandex.clickhouse</groupId>
-            <artifactId>clickhouse-jdbc</artifactId>
-        </dependency>
 
         <!-- outer adapter jar with dependencies-->
         <dependency>
@@ -171,19 +167,6 @@
             <classifier>jar-with-dependencies</classifier>
             <scope>provided</scope>
         </dependency>
-        <dependency>
-            <groupId>com.alibaba.otter</groupId>
-            <artifactId>client-adapter.clickhouse</artifactId>
-            <version>${project.version}</version>
-            <exclusions>
-                <exclusion>
-                    <artifactId>*</artifactId>
-                    <groupId>*</groupId>
-                </exclusion>
-            </exclusions>
-            <classifier>jar-with-dependencies</classifier>
-            <scope>provided</scope>
-        </dependency>
         <!-- connector plugin -->
         <dependency>
             <groupId>com.alibaba.otter</groupId>

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

@@ -224,7 +224,7 @@ public class AdapterProcessor {
                                     logger.error("finish turn off switch of destination:" + canalDestination);
                                 } else {
                                     canalMsgConsumer.ack();
-                                    logger.error(e.getMessage() + " Error sync but ACK!", e);
+                                    logger.error(e.getMessage() + " Error sync but ACK!");
                                 }
                             }
                             Thread.sleep(500);

client-adapter/launcher/src/main/resources/application.yml

@@ -91,7 +91,6 @@ canal.conf:
 #        hosts: 127.0.0.1:9300 # 127.0.0.1:9200 for rest mode
 #        properties:
 #          mode: transport # or rest
-#          # security.ca.path: /etc/es8/ca.crt
 #          # security.auth: test:123456 #  only used for rest mode
 #          cluster.name: elasticsearch
 #      - name: kudu
@@ -105,13 +104,3 @@ canal.conf:
 #          jdbc.url: jdbc:phoenix:127.0.0.1:2181:/hbase/db
 #          jdbc.username:
 #          jdbc.password:
-#      - name: clickhouse
-#        key: clickhouse1
-#        properties:
-#          jdbc.driverClassName: ru.yandex.clickhouse.ClickHouseDriver
-#          jdbc.url: jdbc:clickhouse://127.0.0.1:8123/default
-#          jdbc.username: default
-#          jdbc.password: 123456
-#          batchSize: 3000
-#          scheduleTime: 600   # second unit
-#          threads: 3          # parallel threads

client-adapter/logger/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.8-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

client-adapter/phoenix/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.8-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

client-adapter/pom.xml

@@ -4,12 +4,12 @@
     <parent>
         <groupId>com.alibaba.otter</groupId>
         <artifactId>canal</artifactId>
-        <version>1.1.8-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <groupId>com.alibaba.otter</groupId>
     <artifactId>canal.client-adapter</artifactId>
-    <version>1.1.8-SNAPSHOT</version>
+    <version>1.1.7</version>
     <packaging>pom</packaging>
     <name>canal client adapter module for otter ${project.version}</name>
 
@@ -36,7 +36,6 @@
         <module>kudu</module>
         <module>phoenix</module>
         <module>tablestore</module>
-        <module>clickhouse</module>
     </modules>
 
     <licenses>
@@ -193,11 +192,6 @@
                 <artifactId>hbase-client</artifactId>
                 <version>1.4.8</version>
             </dependency>
-            <dependency>
-                <groupId>ru.yandex.clickhouse</groupId>
-                <artifactId>clickhouse-jdbc</artifactId>
-                <version>0.3.1</version>
-            </dependency>
             <dependency>
                 <groupId>com.aliyun.openservices</groupId>
                 <artifactId>tablestore</artifactId>

client-adapter/rdb/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.8-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

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

@@ -1,6 +1,10 @@
 package com.alibaba.otter.canal.client.adapter.rdb.service;
 
-import java.sql.*;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -13,7 +17,10 @@ import com.alibaba.druid.pool.DruidDataSource;
 import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig;
 import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig.DbMapping;
 import com.alibaba.otter.canal.client.adapter.rdb.support.SyncUtil;
-import com.alibaba.otter.canal.client.adapter.support.*;
+import com.alibaba.otter.canal.client.adapter.support.AbstractEtlService;
+import com.alibaba.otter.canal.client.adapter.support.AdapterConfig;
+import com.alibaba.otter.canal.client.adapter.support.EtlResult;
+import com.alibaba.otter.canal.client.adapter.support.Util;
 
 /**
  * RDB ETL 操作业务类
@@ -37,9 +44,7 @@ public class RdbEtlService extends AbstractEtlService {
      */
     public EtlResult importData(List<String> params) {
         DbMapping dbMapping = config.getDbMapping();
-        // 获取源数据源，根据数据库类型拼装数据库名和表名
+        String sql = "SELECT * FROM " + dbMapping.getDatabase() + "." + dbMapping.getTable();
-        DruidDataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
-        String sql = "SELECT * FROM " + SyncUtil.getDbTableName(dbMapping, dataSource.getDbType());
         return importData(sql, params);
     }
 

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

@@ -164,7 +164,8 @@ public class RdbMirrorDbSyncService {
                 logger.trace("Execute DDL sql: {} for database: {}", ddl.getSql(), ddl.getDatabase());
             }
         } catch (Exception e) {
-            throw new RuntimeException(e);
+            logger.error("Execute DDL sql failed sql: {} for database: {} cause: {}",
+                    ddl.getSql(), ddl.getDatabase(), e.getMessage());
         }
     }
 }

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

@@ -1,25 +1,5 @@
 package com.alibaba.otter.canal.client.adapter.rdb.service;
 
-import java.sql.Connection;
-import java.sql.ResultSetMetaData;
-import java.sql.SQLException;
-import java.sql.Types;
-import java.util.ArrayList;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.function.Function;
-
-import org.apache.commons.lang.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import com.alibaba.druid.pool.DruidDataSource;
 import com.alibaba.fastjson2.JSON;
 import com.alibaba.fastjson2.JSONWriter.Feature;
@@ -30,6 +10,27 @@ import com.alibaba.otter.canal.client.adapter.rdb.support.SingleDml;
 import com.alibaba.otter.canal.client.adapter.rdb.support.SyncUtil;
 import com.alibaba.otter.canal.client.adapter.support.Dml;
 import com.alibaba.otter.canal.client.adapter.support.Util;
+import org.apache.commons.lang.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Types;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.function.Function;
+import java.util.stream.Collectors;
 
 /**
  * RDB同步操作业务
@@ -272,7 +273,7 @@ public class RdbSyncService {
         len = insertSql.length();
         insertSql.delete(len - 1, len).append(")");
 
-        Map<String, Integer> ctype = getTargetColumnType(batchExecutor.getConn(), config);
+        Map<String, Integer> ctype = getTargetColumnTypeByColumns(batchExecutor.getConn(), config, columnsMap);
 
         List<Map<String, ?>> values = new ArrayList<>();
         for (Map.Entry<String, String> entry : columnsMap.entrySet()) {
@@ -328,7 +329,7 @@ public class RdbSyncService {
         String backtick = SyncUtil.getBacktickByDbType(dataSource.getDbType());
         Map<String, String> columnsMap = SyncUtil.getColumnsMap(dbMapping, data);
 
-        Map<String, Integer> ctype = getTargetColumnType(batchExecutor.getConn(), config);
+        Map<String, Integer> ctype = getTargetColumnTypeByColumns(batchExecutor.getConn(), config, columnsMap);
 
         StringBuilder updateSql = new StringBuilder();
         updateSql.append("UPDATE ").append(SyncUtil.getDbTableName(dbMapping, dataSource.getDbType())).append(" SET ");
@@ -410,6 +411,57 @@ public class RdbSyncService {
         }
     }
 
+    /**
+     * Obtain metadata for the corresponding table columns of the target data source
+     * Check if the metadata of the columns in the src is consistent with the metadata cache of the columns in the cache
+     * (first determine the number of columns and then determine if each column is consistent).
+     * If not, reload
+     * @param conn
+     * @param config
+     * @param columnsMap
+     * @date 2024/6/17 14:45
+     */
+    private Map<String, Integer> getTargetColumnTypeByColumns(Connection conn, MappingConfig config,
+                                                              Map<String, String> columnsMap) {
+        DbMapping dbMapping = config.getDbMapping();
+        String cacheKey = config.getDestination() + "." + dbMapping.getDatabase() + "." + dbMapping.getTable();
+        Map<String, Integer> columnType = getTargetColumnType(conn, config);
+        if(null != columnType && null != columnsMap &&
+                !checkCloumnConsistent(columnType.keySet(), columnsMap.keySet())){
+            logger.info("column meta data not match, reload the column meta data from target db. key:{}", cacheKey);
+            columnsTypeCache.remove(cacheKey);
+            columnType = getTargetColumnType(conn, config);
+            if(null != columnType && null != columnsMap &&
+                    !checkCloumnConsistent(columnType.keySet(), columnsMap.keySet())) {
+                logger.info("the column meta data from db still not match. dbdata:{}", columnType.keySet());
+            }
+        }
+        return columnType;
+    }
+
+
+    /**
+     * 判断binlog的中的列的数据和缓存的列数据是否一致
+     *
+     * @param <T>  集合元素类型
+     * @param column
+     * @param anotherColumn
+     * @return {@link Boolean}
+     */
+    public static<T> Boolean checkCloumnConsistent(Collection<T> column, Collection<T> anotherColumn) {
+        if(null == column || null == anotherColumn || column.size() !=  anotherColumn.size()){
+            return false;
+        }
+        Map<T, T> anotherColumnMap = column.stream().collect(Collectors.toMap(Function.identity(), Function.identity()));
+
+        for (T t : column) {
+            if(null == anotherColumnMap.get(t)){
+                return false;
+            }
+        }
+        return true;
+    }
+
     /**
      * 获取目标字段类型
      *

client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java

@@ -1,23 +1,21 @@
 package com.alibaba.otter.canal.client.adapter.rdb.support;
 
+import com.alibaba.druid.DbType;
+import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig;
+import com.alibaba.otter.canal.client.adapter.support.Util;
+import org.apache.commons.lang.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.io.Reader;
 import java.io.StringReader;
 import java.math.BigDecimal;
 import java.nio.charset.StandardCharsets;
 import java.sql.*;
-import java.time.LocalDateTime;
 import java.util.Collection;
 import java.util.LinkedHashMap;
 import java.util.Map;
 
-import org.apache.commons.lang.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.alibaba.druid.DbType;
-import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig;
-import com.alibaba.otter.canal.client.adapter.support.Util;
-
 public class SyncUtil {
     private static final Logger logger  = LoggerFactory.getLogger(SyncUtil.class);
 
@@ -28,7 +26,7 @@ public class SyncUtil {
     public static Map<String, String> getColumnsMap(MappingConfig.DbMapping dbMapping, Collection<String> columns) {
         Map<String, String> columnsMap;
         if (dbMapping.getMapAll()) {
-            if (dbMapping.getAllMapColumns() != null) {
+            if (dbMapping.getAllMapColumns() != null && columns.size() == dbMapping.getAllMapColumns().size() ) {
                 return dbMapping.getAllMapColumns();
             }
             columnsMap = new LinkedHashMap<>();
@@ -222,17 +220,12 @@ public class SyncUtil {
                     pstmt.setTime(i, new java.sql.Time(((java.util.Date) value).getTime()));
                 } else if (value instanceof String) {
                     String v = (String) value;
-                    if(Util.isAccuracyOverSecond(v)) {
-                        //the java.sql.time doesn't support for even millisecond, only setObject works here.
-                        pstmt.setObject(i, v);
-                    }else {
                     java.util.Date date = Util.parseDate(v);
                     if (date != null) {
                         pstmt.setTime(i, new Time(date.getTime()));
                     } else {
                         pstmt.setNull(i, type);
                     }
-                    }
                 } else {
                     pstmt.setNull(i, type);
                 }
@@ -245,23 +238,12 @@ public class SyncUtil {
                 } else if (value instanceof String) {
                     String v = (String) value;
                     if (!v.startsWith("0000-00-00")) {
-                        if(Util.isAccuracyOverMillisecond(v)){
-                            //convert to ISO-8601 standard format, with up to nanoseconds (9 digits) precision
-                            LocalDateTime isoDatetime = Util.parseISOLocalDateTime(v);
-                            if (isoDatetime != null) {
-                                pstmt.setTimestamp(i, Timestamp.valueOf(isoDatetime));
-                            } else {
-                                //if can't convert, set to null
-                                pstmt.setNull(i, type);
-                            }
-                        }else {
                         java.util.Date date = Util.parseDate(v);
                         if (date != null) {
                             pstmt.setTimestamp(i, new Timestamp(date.getTime()));
                         } else {
                             pstmt.setNull(i, type);
                         }
-                        }
                     } else {
                         pstmt.setObject(i, value);
                     }

client-adapter/tablestore/pom.xml

@@ -3,14 +3,14 @@
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.8-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.alibaba.otter</groupId>
     <artifactId>client-adapter.tablestore</artifactId>
     <packaging>jar</packaging>
-    <name>canal client adapter rdb module for otter ${project.version}</name>
+    <name>canal client adapter tablestore module for otter ${project.version}</name>
 
     <dependencies>
         <dependency>

client/pom.xml

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

client/src/main/java/com/alibaba/otter/canal/client/CanalMessageDeserializer.java

@@ -51,7 +51,7 @@ public class CanalMessageDeserializer {
                 }
             }
         } catch (Exception e) {
-            throw new CanalClientException("deserializer failed by " + e.getMessage(), e);
+            throw new CanalClientException("deserializer failed", e);
         }
     }
 }

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

@@ -1,5 +1,6 @@
 package com.alibaba.otter.canal.client.kafka;
 
+import com.alibaba.fastjson.JSON;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
@@ -10,7 +11,6 @@ import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.serialization.StringDeserializer;
 
-import com.alibaba.fastjson2.JSON;
 import com.alibaba.otter.canal.client.CanalMQConnector;
 import com.alibaba.otter.canal.client.impl.SimpleCanalConnector;
 import com.alibaba.otter.canal.protocol.FlatMessage;
@@ -78,15 +78,14 @@ public class KafkaCanalConnector implements CanalMQConnector {
             return;
         }
 
+        connected = true;
         if (kafkaConsumer == null && !flatMessage) {
             kafkaConsumer = new KafkaConsumer<>(properties);
-        }
 
+        }
         if (kafkaConsumer2 == null && flatMessage) {
             kafkaConsumer2 = new KafkaConsumer<>(properties);
         }
-
-        connected = true;
     }
 
     /**
@@ -325,3 +324,4 @@ public class KafkaCanalConnector implements CanalMQConnector {
     }
 
 }
+

client/src/main/java/com/alibaba/otter/canal/client/pulsarmq/PulsarMQCanalConnector.java

@@ -1,13 +1,5 @@
 package com.alibaba.otter.canal.client.pulsarmq;
 
-import java.util.List;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.commons.lang.StringUtils;
-import org.apache.pulsar.client.api.*;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import com.alibaba.fastjson2.JSON;
 import com.alibaba.otter.canal.client.CanalMQConnector;
 import com.alibaba.otter.canal.client.CanalMessageDeserializer;
@@ -17,6 +9,13 @@ import com.alibaba.otter.canal.protocol.FlatMessage;
 import com.alibaba.otter.canal.protocol.Message;
 import com.alibaba.otter.canal.protocol.exception.CanalClientException;
 import com.google.common.collect.Lists;
+import org.apache.commons.lang.StringUtils;
+import org.apache.pulsar.client.api.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
 
 /**
  * PulsarMQ的连接
@@ -25,15 +24,13 @@ import com.google.common.collect.Lists;
  * 注意点:
  * 1. 相比于canal {@linkplain SimpleCanalConnector}, 这里get和ack操作不能有并发, 必须是一个线程执行get后，内存里执行完毕ack后再取下一个get
  * </pre>
- * 
  * todo 重复消费的概率相当高。一次批处理中，只要有一个消息处理失败，则该批次全部重试
  *
  * @since 1.1.1
  */
 public class PulsarMQCanalConnector implements CanalMQConnector {
 
-    private static final Logger       logger                     = LoggerFactory
+    private static final Logger logger = LoggerFactory.getLogger(PulsarMQCanalConnector.class);
-        .getLogger(PulsarMQCanalConnector.class);
 
     private volatile Messages<byte[]> lastGetBatchMessage;
 
@@ -61,12 +58,6 @@ public class PulsarMQCanalConnector implements CanalMQConnector {
      * 角色认证token
      */
     private String roleToken;
-
-    /**
-     * listener name
-     */
-    private String                    listenerName;
-
     /**
      * 订阅客户端名称
      */
@@ -76,7 +67,8 @@ public class PulsarMQCanalConnector implements CanalMQConnector {
      */
     private int batchSize = 30;
     /**
-     * 与{@code batchSize}一起决定批量获取的数据大小 当：
+     * 与{@code batchSize}一起决定批量获取的数据大小
+     * 当：
      * <p>
      * 1. {@code batchSize} 条消息未消费时<br/>
      * 2. 距上一次批量消费时间达到{@code batchTimeoutSeconds}秒时
@@ -134,13 +126,8 @@ public class PulsarMQCanalConnector implements CanalMQConnector {
      * @author chad
      * @since 1 by chad at 2021/9/18 完善
      */
-    public PulsarMQCanalConnector(boolean isFlatMessage, String serviceUrl, String roleToken, String topic,
+    public PulsarMQCanalConnector(boolean isFlatMessage, String serviceUrl, String roleToken, String topic
-                                  String subscriptName){
+            , String subscriptName) {
-        this(isFlatMessage, serviceUrl, roleToken, topic, subscriptName, null);
-    }
-
-    public PulsarMQCanalConnector(boolean isFlatMessage, String serviceUrl, String roleToken, String topic,
-                                  String subscriptName, String listenerName){
         this.isFlatMessage = isFlatMessage;
         this.serviceUrl = serviceUrl;
         this.roleToken = roleToken;
@@ -149,7 +136,6 @@ public class PulsarMQCanalConnector implements CanalMQConnector {
         if (StringUtils.isEmpty(this.subscriptName)) {
             throw new RuntimeException("Pulsar Consumer subscriptName required");
         }
-        this.listenerName = listenerName;
     }
 
     /**
@@ -159,31 +145,10 @@ public class PulsarMQCanalConnector implements CanalMQConnector {
      * @author chad
      * @since 1 by chad at 2021/9/18 完善
      */
-    public PulsarMQCanalConnector(boolean isFlatMessage, String serviceUrl, String roleToken, String topic,
+    public PulsarMQCanalConnector(boolean isFlatMessage, String serviceUrl, String roleToken, String topic
-                                  String subscriptName, int batchSize, int getBatchTimeoutSeconds,
+            , String subscriptName, int batchSize, int getBatchTimeoutSeconds, int batchProcessTimeoutSeconds
-                                  int batchProcessTimeoutSeconds, int redeliveryDelaySeconds, int ackTimeoutSeconds,
+            , int redeliveryDelaySeconds, int ackTimeoutSeconds, boolean isRetry, boolean isRetryDLQUpperCase
-                                  boolean isRetry, boolean isRetryDLQUpperCase, int maxRedeliveryCount){
+            , int maxRedeliveryCount) {
-        this(isFlatMessage,
-            serviceUrl,
-            roleToken,
-            topic,
-            subscriptName,
-            batchSize,
-            getBatchTimeoutSeconds,
-            batchProcessTimeoutSeconds,
-            redeliveryDelaySeconds,
-            ackTimeoutSeconds,
-            isRetry,
-            isRetryDLQUpperCase,
-            maxRedeliveryCount,
-            null);
-    }
-
-    public PulsarMQCanalConnector(boolean isFlatMessage, String serviceUrl, String roleToken, String topic,
-                                  String subscriptName, int batchSize, int getBatchTimeoutSeconds,
-                                  int batchProcessTimeoutSeconds, int redeliveryDelaySeconds, int ackTimeoutSeconds,
-                                  boolean isRetry, boolean isRetryDLQUpperCase, int maxRedeliveryCount,
-                                  String listenerName){
         this.isFlatMessage = isFlatMessage;
         this.serviceUrl = serviceUrl;
         this.roleToken = roleToken;
@@ -200,20 +165,16 @@ public class PulsarMQCanalConnector implements CanalMQConnector {
         this.isRetry = isRetry;
         this.isRetryDLQUpperCase = isRetryDLQUpperCase;
         this.maxRedeliveryCount = maxRedeliveryCount;
-        this.listenerName = listenerName;
     }
 
     @Override
     public void connect() throws CanalClientException {
         // 连接创建客户端
         try {
-            ClientBuilder builder = PulsarClient.builder()
+            pulsarClient = PulsarClient.builder()
                     .serviceUrl(serviceUrl)
-                .authentication(AuthenticationFactory.token(roleToken));
+                    .authentication(AuthenticationFactory.token(roleToken))
-            if (StringUtils.isNotEmpty(listenerName)) {
+                    .build();
-                builder.listenerName(listenerName);
-            }
-            pulsarClient = builder.build();
         } catch (PulsarClientException e) {
             throw new RuntimeException(e);
         }
@@ -265,7 +226,8 @@ public class PulsarMQCanalConnector implements CanalMQConnector {
                 // 调用consumer.negativeAcknowledge(message) （即nack）来表示消费失败的消息
                 // 在指定的时间进行重新消费，默认是1分钟。
                 .negativeAckRedeliveryDelay(this.redeliveryDelaySeconds, TimeUnit.SECONDS)
-            .subscriptionName(this.subscriptName);
+                .subscriptionName(this.subscriptName)
+        ;
         if (this.isRetry) {
             DeadLetterPolicy.DeadLetterPolicyBuilder dlqBuilder = DeadLetterPolicy.builder()
                     // 最大重试次数
@@ -278,26 +240,29 @@ public class PulsarMQCanalConnector implements CanalMQConnector {
                 dlqBuilder.deadLetterTopic(dlqTopic);
             }
 
-            // 默认关闭，如果需要重试则开启
+            //默认关闭，如果需要重试则开启
-            builder.enableRetry(true).deadLetterPolicy(dlqBuilder.build());
+            builder.enableRetry(true)
+                    .deadLetterPolicy(dlqBuilder.build());
         }
 
         // ack超时
         builder.ackTimeout(this.ackTimeoutSeconds, TimeUnit.SECONDS);
 
         // pulsar批量获取消息设置
-        builder.batchReceivePolicy(new BatchReceivePolicy.Builder().maxNumMessages(this.batchSize)
+        builder.batchReceivePolicy(new BatchReceivePolicy.Builder()
+                .maxNumMessages(this.batchSize)
                 .timeout(this.getBatchTimeoutSeconds, TimeUnit.SECONDS)
                 .build());
 
         try {
             this.consumer = builder.subscribe();
-            connected = true;
         } catch (PulsarClientException e) {
             throw new RuntimeException(e);
         }
+        connected = true;
     }
 
+
     @Override
     public void subscribe() throws CanalClientException {
         this.subscribe(null);
@@ -334,8 +299,7 @@ public class PulsarMQCanalConnector implements CanalMQConnector {
     }
 
     /**
-     * 关心业务执行结果，业务侧根据执行结果调用
+     * 关心业务执行结果，业务侧根据执行结果调用 {@link PulsarMQCanalConnector#ack()}或{@link PulsarMQCanalConnector#rollback()}
-     * {@link PulsarMQCanalConnector#ack()}或{@link PulsarMQCanalConnector#rollback()}
      * <p>
      * 本方法示支持多线程，在MQ保障顺序的前提下，也无法提供单Topic多线程
      * </p>
@@ -375,8 +339,7 @@ public class PulsarMQCanalConnector implements CanalMQConnector {
      * @return java.util.List<T>
      * @date 2021/9/14 15:20
      * @author chad
-     * @since 1 by chad at 2021/9/14
+     * @since 1 by chad at 2021/9/14 供{@link PulsarMQCanalConnector#getListWithoutAck(Long, TimeUnit)}
-     * 供{@link PulsarMQCanalConnector#getListWithoutAck(Long, TimeUnit)}
      * 和{@link PulsarMQCanalConnector#getFlatListWithoutAck(Long, TimeUnit)}调用
      */
     private <T> List<T> getListWithoutAck() {

client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java

@@ -6,6 +6,7 @@ import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 
+import com.alibaba.otter.canal.client.ConsumerBatchMessage;
 import org.apache.commons.lang.StringUtils;
 import org.apache.rocketmq.acl.common.AclClientRPCHook;
 import org.apache.rocketmq.acl.common.SessionCredentials;
@@ -24,7 +25,6 @@ import org.slf4j.LoggerFactory;
 import com.alibaba.fastjson2.JSON;
 import com.alibaba.otter.canal.client.CanalMQConnector;
 import com.alibaba.otter.canal.client.CanalMessageDeserializer;
-import com.alibaba.otter.canal.client.ConsumerBatchMessage;
 import com.alibaba.otter.canal.client.impl.SimpleCanalConnector;
 import com.alibaba.otter.canal.protocol.FlatMessage;
 import com.alibaba.otter.canal.protocol.Message;
@@ -155,10 +155,11 @@ public class RocketMQCanalConnector implements CanalMQConnector {
                 }
             });
             rocketMQConsumer.start();
-            connected = true;
         } catch (MQClientException ex) {
-            throw new RuntimeException("Start RocketMQ consumer error", ex);
+            connected = false;
+            logger.error("Start RocketMQ consumer error", ex);
         }
+        connected = true;
     }
 
     private boolean process(List<MessageExt> messageExts) {

common/pom.xml

@@ -3,7 +3,7 @@
 	<parent>
 		<groupId>com.alibaba.otter</groupId>
 		<artifactId>canal</artifactId>
-		<version>1.1.8-SNAPSHOT</version>
+		<version>1.1.7</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<artifactId>canal.common</artifactId>
@@ -87,11 +87,6 @@
 			<groupId>com.google.code.findbugs</groupId>
 			<artifactId>jsr305</artifactId>
 		</dependency>
-		<dependency>
-			<groupId>com.alibaba</groupId>
-			<artifactId>druid</artifactId>
-			<optional>true</optional>
-		</dependency>
 		<dependency>
 			<groupId>org.springframework</groupId>
 			<artifactId>spring-test</artifactId>

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

@@ -7,7 +7,6 @@ import java.net.ServerSocket;
 import java.util.Enumeration;
 import java.util.regex.Pattern;
 
-import org.apache.commons.lang.ArrayUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -76,27 +75,4 @@ public class AddressUtils {
         logger.error("Could not get local host ip address, will use 127.0.0.1 instead.");
         return localAddress;
     }
-
-    /**
-     * 拆分IP地址和端口号
-     *
-     * @param text ip地址和端口号，ip和端口号以英文冒号(:)分隔;
-     * 
-     *     <pre>
-     *             ipv4 127.0.0.1:3306
-     *             ipv6 [::1]:3306
-     *     </pre>
-     *
-     * @return
-     */
-    public static String[] splitIPAndPort(String text) {
-        text = text.replace("[", "").replace("]", "");
-        int idx = text.lastIndexOf(':');
-        if (idx > 0) {
-            String ip = text.substring(0, idx);
-            String port = text.substring(idx + 1);
-            return new String[] { ip, port };
-        }
-        return ArrayUtils.EMPTY_STRING_ARRAY;
-    }
 }

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

@@ -1,7 +1,5 @@
 package com.alibaba.otter.canal.common.utils;
 
-import com.alibaba.druid.filter.config.ConfigTools;
-
 import java.io.File;
 
 /**
@@ -52,8 +50,4 @@ public class CommonUtils {
 
         return dirFile.delete();
     }
-
-    public static String decryptDruidPassword(String publicKeyText, String cipherText) throws Exception {
-        return ConfigTools.decrypt(publicKeyText, cipherText);
-    }
 }

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

@@ -44,7 +44,7 @@ public class ExecutorTemplate {
         check(future);
     }
 
-    public void submit(Callable task) {
+    public void submit(Callable<Exception> task) {
         Future future = executor.submit(task);
         futures.add(future);
         check(future);

connector/core/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.connector</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.8-SNAPSHOT</version>
+        <version>1.1.7</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.alibaba.otter</groupId>

connector/core/src/main/java/com/alibaba/otter/canal/connector/core/util/CanalMessageSerializerUtil.java

@@ -79,7 +79,7 @@ public class CanalMessageSerializerUtil {
                 }
             }
         } catch (Exception e) {
-            throw new RuntimeException("Error when serializing message to byte[] by " + e.getMessage() , e);
+            throw new RuntimeException("Error when serializing message to byte[] ");
         }
         return null;
     }
@@ -125,7 +125,7 @@ public class CanalMessageSerializerUtil {
                 }
             }
         } catch (Exception e) {
-            throw new CanalClientException("deserializer failed by " + e.getMessage(), e);
+            throw new CanalClientException("deserializer failed", e);
         }
     }
 }

connector/kafka-connector/pom.xml

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

connector/kafka-connector/src/main/java/com/alibaba/otter/canal/connector/kafka/producer/CanalKafkaProducer.java

@@ -9,6 +9,7 @@ import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 
+import com.alibaba.otter.canal.common.utils.PropertiesUtils;
 import org.apache.commons.lang.StringUtils;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.Producer;
@@ -20,7 +21,6 @@ import org.slf4j.LoggerFactory;
 import com.alibaba.fastjson2.JSON;
 import com.alibaba.fastjson2.JSONWriter;
 import com.alibaba.otter.canal.common.utils.ExecutorTemplate;
-import com.alibaba.otter.canal.common.utils.PropertiesUtils;
 import com.alibaba.otter.canal.connector.core.producer.AbstractMQProducer;
 import com.alibaba.otter.canal.connector.core.producer.MQDestination;
 import com.alibaba.otter.canal.connector.core.producer.MQMessageUtils;
@@ -249,15 +249,13 @@ public class CanalKafkaProducer extends AbstractMQProducer implements CanalMQPro
                         FlatMessage flatMessagePart = partitionFlatMessage[i];
                         if (flatMessagePart != null) {
                             records.add(new ProducerRecord<>(topicName, i, null, JSON.toJSONBytes(flatMessagePart,
-                                JSONWriter.Feature.WriteNulls,
+                                JSONWriter.Feature.WriteNulls)));
-                                JSONWriter.Feature.LargeObject)));
                         }
                     }
                 } else {
                     final int partition = mqDestination.getPartition() != null ? mqDestination.getPartition() : 0;
                     records.add(new ProducerRecord<>(topicName, partition, null, JSON.toJSONBytes(flatMessage,
-                        JSONWriter.Feature.WriteNulls,
+                        JSONWriter.Feature.WriteNulls)));
-                        JSONWriter.Feature.LargeObject)));
                 }
             }
         }

connector/pom.xml

@@ -4,13 +4,13 @@
     <parent>
         <groupId>com.alibaba.otter</groupId>
         <artifactId>canal</artifactId>
-        <version>1.1.8-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 
     <groupId>com.alibaba.otter</groupId>
     <artifactId>canal.connector</artifactId>
-    <version>1.1.8-SNAPSHOT</version>
+    <version>1.1.7</version>
     <packaging>pom</packaging>
     <name>canal connector module for otter ${project.version}</name>
 

connector/pulsarmq-connector/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.connector</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.8-SNAPSHOT</version>
+        <version>1.1.7</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

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

@@ -61,9 +61,4 @@ public class PulsarMQConstants {
      */
     public static final String PULSARMQ_ADMIN_SERVER_URL          = ROOT + "." + "adminServerUrl";
 
-    /**
-     * Pulsar admin服务器地址
-     */
-    public static final String PULSARMQ_LISTENER_NAME             = ROOT + "." + "listenerName";
-
 }

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

@@ -37,11 +37,6 @@ public class PulsarMQProducerConfig extends MQProperties {
      */
     private String adminServerUrl;
 
-    /**
-     * listener name
-     */
-    private String listenerName;
-
     public String getServerUrl() {
         return serverUrl;
     }
@@ -73,12 +68,4 @@ public class PulsarMQProducerConfig extends MQProperties {
     public void setAdminServerUrl(String adminServerUrl) {
         this.adminServerUrl = adminServerUrl;
     }
-
-    public String getListenerName() {
-        return listenerName;
-    }
-
-    public void setListenerName(String listenerName) {
-        this.listenerName = listenerName;
-    }
 }

connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumer.java

@@ -54,12 +54,6 @@ public class CanalPulsarMQConsumer implements CanalMsgConsumer {
      * 角色认证token
      */
     private String                    roleToken;
-
-    /**
-     * listener name
-     */
-    private String                    listenerName;
-
     /**
      * 订阅客户端名称
      */
@@ -116,7 +110,6 @@ public class CanalPulsarMQConsumer implements CanalMsgConsumer {
         }
         this.serviceUrl = properties.getProperty(PulsarMQConstants.PULSARMQ_SERVER_URL);
         this.roleToken = properties.getProperty(PulsarMQConstants.PULSARMQ_ROLE_TOKEN);
-        this.listenerName = properties.getProperty(PulsarMQConstants.PULSARMQ_LISTENER_NAME);
         this.subscriptName = properties.getProperty(PulsarMQConstants.PULSARMQ_SUBSCRIPT_NAME);
         // 采用groupId作为subscriptName，避免所有的都是同一个订阅者名称
         if (StringUtils.isEmpty(this.subscriptName)) {
@@ -172,9 +165,6 @@ public class CanalPulsarMQConsumer implements CanalMsgConsumer {
             if (StringUtils.isNotEmpty(roleToken)) {
                 builder.authentication(AuthenticationFactory.token(roleToken));
             }
-            if (StringUtils.isNotEmpty(listenerName)) {
-                builder.authentication(AuthenticationFactory.token(listenerName));
-            }
             pulsarClient = builder.build();
         } catch (PulsarClientException e) {
             throw new RuntimeException(e);

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

@@ -13,7 +13,6 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import com.alibaba.fastjson2.JSON;
-import com.alibaba.fastjson2.JSONWriter;
 import com.alibaba.fastjson2.JSONWriter.Feature;
 import com.alibaba.otter.canal.common.utils.ExecutorTemplate;
 import com.alibaba.otter.canal.common.utils.NamedThreadFactory;
@@ -73,11 +72,6 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
                 // 角色权限认证的token
                 builder.authentication(AuthenticationFactory.token(pulsarMQProducerConfig.getRoleToken()));
             }
-            if (StringUtils.isNotEmpty(pulsarMQProducerConfig.getListenerName())) {
-                // listener name
-                builder.listenerName(pulsarMQProducerConfig.getListenerName());
-            }
-
             client = builder.build();
         } catch (PulsarClientException e) {
             throw new RuntimeException(e);
@@ -132,11 +126,6 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
         if (!StringUtils.isEmpty(adminServerUrl)) {
             tmpProperties.setAdminServerUrl(adminServerUrl);
         }
-        String listenerName = PropertiesUtils.getProperty(properties, PulsarMQConstants.PULSARMQ_LISTENER_NAME);
-        if (!StringUtils.isEmpty(listenerName)) {
-            tmpProperties.setListenerName(listenerName);
-        }
-
         if (logger.isDebugEnabled()) {
             logger.debug("Load pulsar properties ==> {}", JSON.toJSON(this.mqProperties));
         }
@@ -341,7 +330,7 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
             try {
                 MessageId msgResultId = producer.newMessage()
                     .property(MSG_PROPERTY_PARTITION_NAME, String.valueOf(partition))
-                    .value(JSON.toJSONBytes(f, Feature.WriteNulls, JSONWriter.Feature.LargeObject))
+                    .value(JSON.toJSONBytes(f, Feature.WriteNulls))
                     .send()
                 //
                 ;

connector/rabbitmq-connector/pom.xml

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

connector/rabbitmq-connector/src/main/java/com/alibaba/otter/canal/connector/rabbitmq/consumer/CanalRabbitMQConsumer.java

@@ -9,7 +9,6 @@ import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
-import com.alibaba.otter.canal.common.utils.AddressUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -89,7 +88,7 @@ public class CanalRabbitMQConsumer implements CanalMsgConsumer {
         }
         // 解析出端口 modified by 16075140
         if (nameServer != null && nameServer.contains(":")) {
-            String[] serverHostAndPort = AddressUtils.splitIPAndPort(nameServer);
+            String[] serverHostAndPort = nameServer.split(":");
             factory.setHost(serverHostAndPort[0]);
             factory.setPort(Integer.parseInt(serverHostAndPort[1]));
         } else {

connector/rabbitmq-connector/src/main/java/com/alibaba/otter/canal/connector/rabbitmq/producer/CanalRabbitMQProducer.java

@@ -1,9 +1,6 @@
 package com.alibaba.otter.canal.connector.rabbitmq.producer;
 
 import java.io.IOException;
-import java.net.URISyntaxException;
-import java.security.KeyManagementException;
-import java.security.NoSuchAlgorithmException;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
@@ -16,7 +13,6 @@ import org.slf4j.LoggerFactory;
 import com.alibaba.fastjson2.JSON;
 import com.alibaba.fastjson2.JSONWriter;
 import com.alibaba.otter.canal.common.CanalException;
-import com.alibaba.otter.canal.common.utils.AddressUtils;
 import com.alibaba.otter.canal.common.utils.ExecutorTemplate;
 import com.alibaba.otter.canal.common.utils.PropertiesUtils;
 import com.alibaba.otter.canal.connector.core.producer.AbstractMQProducer;
@@ -55,14 +51,8 @@ public class CanalRabbitMQProducer extends AbstractMQProducer implements CanalMQ
 
         ConnectionFactory factory = new ConnectionFactory();
         String servers = rabbitMQProperties.getHost();
-        if (servers.startsWith("amqp")) {
+        if (servers.contains(":")) {
-            try {
+            String[] serverHostAndPort = servers.split(":");
-                factory.setUri(servers);
-            } catch (URISyntaxException | NoSuchAlgorithmException | KeyManagementException ex) {
-                throw new CanalException("failed to parse host", ex);
-            }
-        } else if (servers.contains(":")) {
-            String[] serverHostAndPort = AddressUtils.splitIPAndPort(servers);
             factory.setHost(serverHostAndPort[0]);
             factory.setPort(Integer.parseInt(serverHostAndPort[1]));
         } else {
@@ -82,18 +72,24 @@ public class CanalRabbitMQProducer extends AbstractMQProducer implements CanalMQ
         try {
             connect = factory.newConnection();
             channel = connect.createChannel();
-            String queue = rabbitMQProperties.getQueue();
+            String queue=rabbitMQProperties.getQueue();
-            String exchange = rabbitMQProperties.getExchange();
+            String exchange=rabbitMQProperties.getExchange();
-            String deliveryMode = rabbitMQProperties.getDeliveryMode();
+            String deliveryMode=rabbitMQProperties.getDeliveryMode();
-            String routingKey = rabbitMQProperties.getRoutingKey();
+            String routingKey=rabbitMQProperties.getRoutingKey();
-            if (!StringUtils.isEmpty(queue)) {
+            if(!StringUtils.isEmpty(queue))
+            {
                 channel.queueDeclare(queue, true, false, false, null);
             }
-            if (!StringUtils.isEmpty(queue) && !StringUtils.isEmpty(exchange) && !StringUtils.isEmpty(deliveryMode)
+            if(!StringUtils.isEmpty(queue)
-                && !StringUtils.isEmpty(routingKey)) {
+                &&!StringUtils.isEmpty(exchange)
+                &&!StringUtils.isEmpty(deliveryMode)
+                &&!StringUtils.isEmpty(routingKey))
+            {
                 channel.exchangeDeclare(exchange, deliveryMode, true, false, false, null);
-                channel.queueBind(queue, exchange, routingKey);
+                channel.queueBind(queue,exchange,routingKey);
             }
+            
+
         } catch (IOException | TimeoutException ex) {
             throw new CanalException("Start RabbitMQ producer error", ex);
         }
@@ -180,8 +176,7 @@ public class CanalRabbitMQProducer extends AbstractMQProducer implements CanalMQ
             // 串行分区
             List<FlatMessage> flatMessages = MQMessageUtils.messageConverter(datas, messageSub.getId());
             for (FlatMessage flatMessage : flatMessages) {
-                byte[] message = JSON
+                byte[] message = JSON.toJSONBytes(flatMessage, JSONWriter.Feature.WriteNulls);
-                    .toJSONBytes(flatMessage, JSONWriter.Feature.WriteNulls, JSONWriter.Feature.LargeObject);
                 if (logger.isDebugEnabled()) {
                     logger.debug("send message:{} to destination:{}", message, canalDestination.getCanalDestination());
                 }

connector/rocketmq-connector/pom.xml

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

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

@@ -263,7 +263,7 @@ import java.util.stream.Collectors;
                             List<Message> messages = flatMessagePart.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)))
                                     .collect(Collectors.toList());
                             // 批量发送
                             sendMessage(messages, index);
@@ -278,7 +278,7 @@ import java.util.stream.Collectors;
                 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)))
                         .collect(Collectors.toList());
                 // 批量发送
                 sendMessage(messages, partition);

connector/tcp-connector/pom.xml

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

connector/tcp-connector/src/main/java/com/alibaba/otter/canal/connector/tcp/consumer/CanalTCPConsumer.java

@@ -6,7 +6,6 @@ import java.util.List;
 import java.util.Properties;
 import java.util.concurrent.TimeUnit;
 
-import com.alibaba.otter.canal.common.utils.AddressUtils;
 import org.apache.commons.lang.StringUtils;
 
 import com.alibaba.otter.canal.client.CanalConnector;
@@ -48,7 +47,7 @@ public class CanalTCPConsumer implements CanalMsgConsumer {
             batchSize = Integer.parseInt(batchSizePro);
         }
         if (StringUtils.isNotBlank(host)) {
-            String[] ipPort =  AddressUtils.splitIPAndPort(host);
+            String[] ipPort = host.split(":");
             SocketAddress sa = new InetSocketAddress(ipPort[0], Integer.parseInt(ipPort[1]));
             this.canalConnector = new SimpleCanalConnector(sa, username, password, destination);
         } else {

dbsync/pom.xml

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

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

@@ -1,9 +1,11 @@
 package com.taobao.tddl.dbsync.binlog;
 
-import static com.taobao.tddl.dbsync.binlog.event.RowsLogBuffer.*;
-
 import java.nio.charset.Charset;
 
+import static com.taobao.tddl.dbsync.binlog.event.RowsLogBuffer.appendNumber2;
+import static com.taobao.tddl.dbsync.binlog.event.RowsLogBuffer.appendNumber4;
+import static com.taobao.tddl.dbsync.binlog.event.RowsLogBuffer.usecondsToStr;
+
 /**
  * 处理下MySQL json二进制转化为可读的字符串
  * 
@@ -32,25 +34,26 @@ public class JsonConversion {
     public static final char JSONB_FALSE_LITERAL     = '\2';
 
     /*
-     * The size of offset or size fields in the small and the large storage format
+     * The size of offset or size fields in the small and the large storage
-     * for JSON objects and JSON arrays.
+     * format for JSON objects and JSON arrays.
      */
     public static final int  SMALL_OFFSET_SIZE       = 2;
     public static final int  LARGE_OFFSET_SIZE       = 4;
 
     /*
-     * The size of key entries for objects when using the small storage format or
+     * The size of key entries for objects when using the small storage format
-     * the large storage format. In the small format it is 4 bytes (2 bytes for key
+     * or the large storage format. In the small format it is 4 bytes (2 bytes
-     * length and 2 bytes for key offset). In the large format it is 6 (2 bytes for
+     * for key length and 2 bytes for key offset). In the large format it is 6
-     * length, 4 bytes for offset).
+     * (2 bytes for length, 4 bytes for offset).
      */
     public static final int  KEY_ENTRY_SIZE_SMALL    = (2 + SMALL_OFFSET_SIZE);
     public static final int  KEY_ENTRY_SIZE_LARGE    = (2 + LARGE_OFFSET_SIZE);
 
     /*
-     * The size of value entries for objects or arrays. When using the small storage
+     * The size of value entries for objects or arrays. When using the small
-     * format, the entry size is 3 (1 byte for type, 2 bytes for offset). When using
+     * storage format, the entry size is 3 (1 byte for type, 2 bytes for
-     * the large storage format, it is 5 (1 byte for type, 4 bytes for offset).
+     * offset). When using the large storage format, it is 5 (1 byte for type, 4
+     * bytes for offset).
      */
     public static final int  VALUE_ENTRY_SIZE_SMALL  = (1 + SMALL_OFFSET_SIZE);
     public static final int  VALUE_ENTRY_SIZE_LARGE  = (1 + LARGE_OFFSET_SIZE);
@@ -157,12 +160,13 @@ public class JsonConversion {
                 if (len < n + str_len) {
                     throw new IllegalArgumentException("illegal json data");
                 }
-                return new Json_Value(Json_enum_type.STRING,
+                return new Json_Value(Json_enum_type.STRING, buffer.rewind()
-                    buffer.rewind().forward((int) n).getFixString((int) str_len, charset));
+                    .forward((int) n)
+                    .getFixString((int) str_len, charset));
             case JSONB_TYPE_OPAQUE:
                 /*
-                 * There should always be at least one byte, which tells the field type of the
+                 * There should always be at least one byte, which tells the
-                 * opaque value.
+                 * field type of the opaque value.
                  */
                 // The type is encoded as a uint8 that maps to an
                 // enum_field_types.
@@ -287,7 +291,7 @@ public class JsonConversion {
                         if (i > 0) {
                             buf.append(", ");
                         }
-                        buf.append('"').append(escapse(key(i, charset))).append('"');
+                        buf.append('"').append(key(i, charset)).append('"');
                         buf.append(": ");
                         element(i, charset).toJsonString(buf, charset);
                     }
@@ -448,8 +452,7 @@ public class JsonConversion {
     }
 
     public static enum Json_enum_type {
-                                       OBJECT, ARRAY, STRING, INT, UINT, DOUBLE, LITERAL_NULL, LITERAL_TRUE,
+        OBJECT, ARRAY, STRING, INT, UINT, DOUBLE, LITERAL_NULL, LITERAL_TRUE, LITERAL_FALSE, OPAQUE, ERROR
-                                       LITERAL_FALSE, OPAQUE, ERROR
     }
 
 }

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

@@ -3,7 +3,6 @@ package com.taobao.tddl.dbsync.binlog;
 import java.nio.charset.Charset;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Objects;
 
 import com.taobao.tddl.dbsync.binlog.JsonConversion.Json_Value;
 import com.taobao.tddl.dbsync.binlog.JsonConversion.Json_enum_type;
@@ -72,21 +71,12 @@ public class JsonDiffConversion {
 
                 buffer.forward((int) value_length);
             }
-
-            // see https://github.com/alibaba/canal/pull/5018
-            if (buffer.position() - position >= len) {
-                break;
-            }
-        }
-
-        if (buffer.position() - position != len) {
-            throw new IllegalArgumentException("reading json diff");
         }
 
         // Print function names in reverse order.
         StringBuilder builder = new StringBuilder();
         for (int i = operation_names.size() - 1; i >= 0; i--) {
-            if (i == 0 || !Objects.equals(operation_names.get(i - 1), operation_names.get(i))) {
+            if (i == 0 || operation_names.get(i - 1) != operation_names.get(i)) {
                 builder.append(operation_names.get(i)).append("(");
             }
         }
@@ -100,7 +90,7 @@ public class JsonDiffConversion {
 
         // In case this vector is empty (a no-op), make an early return
         // after printing only the column name
-        if (operation_names.isEmpty()) {
+        if (operation_names.size() == 0) {
             return builder;
         }
 
@@ -138,14 +128,8 @@ public class JsonDiffConversion {
                 builder.append(jsonBuilder);
             }
 
-            // see https://github.com/alibaba/canal/pull/5018
-            if (buffer.position() - position >= len) {
-                builder.append(")");
-                break;
-            }
-
             // Print closing parenthesis
-            if (!buffer.hasRemaining() || !Objects.equals(operation_names.get(diff_i + 1), operation_names.get(diff_i))) {
+            if (!buffer.hasRemaining() || operation_names.get(diff_i + 1) != operation_names.get(diff_i)) {
                 builder.append(")");
             }
 
@@ -155,10 +139,6 @@ public class JsonDiffConversion {
             diff_i++;
         }
 
-        if (buffer.position() - position != len) {
-            throw new IllegalArgumentException("reading json diff");
-        }
-
         return builder;
     }
 

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

@@ -5,6 +5,7 @@ import java.util.Map;
 
 import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet;
 import com.taobao.tddl.dbsync.binlog.event.FormatDescriptionLogEvent;
+import com.taobao.tddl.dbsync.binlog.event.GtidLogEvent;
 import com.taobao.tddl.dbsync.binlog.event.TableMapLogEvent;
 
 /**
@@ -27,8 +28,6 @@ public final class LogContext {
 
     private boolean                           iterateDecode = false;
 
-    private boolean                           compatiablePercona = false;
-
     public LogContext(){
         this.formatDescription = FormatDescriptionLogEvent.FORMAT_DESCRIPTION_EVENT_5_x;
     }
@@ -93,12 +92,4 @@ public final class LogContext {
     public void setIterateDecode(boolean iterateDecode) {
         this.iterateDecode = iterateDecode;
     }
-
-    public boolean isCompatiablePercona() {
-        return compatiablePercona;
-    }
-
-    public void setCompatiablePercona(boolean compatiablePercona) {
-        this.compatiablePercona = compatiablePercona;
-    }
 }

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

@@ -158,6 +158,8 @@ public final class LogDecoder {
             } finally {
                 context.setIterateDecode(false);
             }
+        } else {
+            // TODO support mariadb compress binlog
         }
         return events;
     }
@@ -191,10 +193,7 @@ public final class LogDecoder {
         LogEvent gtidLogEvent = context.getGtidLogEvent();
         switch (header.getType()) {
             case LogEvent.QUERY_EVENT: {
-                QueryLogEvent event = new QueryLogEvent(header,
+                QueryLogEvent event = new QueryLogEvent(header, buffer, descriptionEvent);
-                    buffer,
-                    descriptionEvent,
-                    context.isCompatiablePercona());
                 /* updating position in context */
                 logPosition.position = header.getLogPos();
                 header.putGtid(context.getGtidSet(), gtidLogEvent);
@@ -355,10 +354,7 @@ public final class LogDecoder {
                 return event;
             }
             case LogEvent.EXECUTE_LOAD_QUERY_EVENT: {
-                ExecuteLoadQueryLogEvent event = new ExecuteLoadQueryLogEvent(header,
+                ExecuteLoadQueryLogEvent event = new ExecuteLoadQueryLogEvent(header, buffer, descriptionEvent);
-                    buffer,
-                    descriptionEvent,
-                    context.isCompatiablePercona());
                 /* updating position in context */
                 logPosition.position = header.getLogPos();
                 return event;

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

@@ -52,9 +52,9 @@ public final class ExecuteLoadQueryLogEvent extends QueryLogEvent {
     public static final int ELQ_FN_POS_END_OFFSET   = ELQ_FILE_ID_OFFSET + 8;
     public static final int ELQ_DUP_HANDLING_OFFSET = ELQ_FILE_ID_OFFSET + 12;
 
-    public ExecuteLoadQueryLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent,
+    public ExecuteLoadQueryLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent)
-                                    boolean compatiablePercona) throws IOException{
+                                                                                                                   throws IOException{
-        super(header, buffer, descriptionEvent, compatiablePercona);
+        super(header, buffer, descriptionEvent);
 
         buffer.position(descriptionEvent.commonHeaderLen + ELQ_FILE_ID_OFFSET);
 

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

@@ -434,8 +434,8 @@ public class QueryLogEvent extends LogEvent {
     private int             tvSec                     = -1;
     private BigInteger      ddlXid                    = BigInteger.valueOf(-1L);
     private Charset         charset;
+
     private String          timezone;
-    private boolean         compatiablePercona        = false;
 
     public QueryLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent)
                                                                                                          throws IOException{
@@ -443,15 +443,8 @@ public class QueryLogEvent extends LogEvent {
     }
 
     public QueryLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent,
-                         boolean compatiablePercona) throws IOException{
-        this(header, buffer, descriptionEvent, compatiablePercona, false);
-    }
-
-    public QueryLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent,
-                         boolean compatiablePercona,
                          boolean compress) throws IOException{
         super(header);
-        this.compatiablePercona = compatiablePercona;
         final int commonHeaderLen = descriptionEvent.commonHeaderLen;
         final int postHeaderLen = descriptionEvent.postHeaderLen[header.type - 1];
         /*
@@ -614,21 +607,6 @@ public class QueryLogEvent extends LogEvent {
      */
     public static final int Q_SQL_REQUIRE_PRIMARY_KEY         = 19;
 
-    /**
-     * Replicate default_table_encryption.
-     */
-    public static final int Q_DEFAULT_TABLE_ENCRYPTION        = 20;
-
-    /**
-     * @since percona 8.0.31 Replicate ddl_skip_rewrite.
-     */
-    public static final int Q_DDL_SKIP_REWRITE                = 21;
-
-    /**
-     * @since percona 8.0.31 Replicate Q_WSREP_SKIP_READONLY_CHECKS.
-     */
-    public static final int Q_WSREP_SKIP_READONLY_CHECKS      = 128;
-
     /**
      * FROM MariaDB 5.5.34
      */
@@ -641,8 +619,6 @@ public class QueryLogEvent extends LogEvent {
 
     public static final int Q_GTID_FLAGS3                     = 130;
 
-    public static final int Q_CHARACTER_SET_COLLATIONS        = 131;
-
     private final void unpackVariables(LogBuffer buffer, final int end) throws IOException {
         int code = -1;
         try {
@@ -737,24 +713,9 @@ public class QueryLogEvent extends LogEvent {
                         // *start++ = thd->variables.sql_require_primary_key;
                         buffer.forward(1);
                         break;
-                    case Q_DEFAULT_TABLE_ENCRYPTION:
-                        // *start++ = thd->variables.default_table_encryption;
-                        buffer.forward(1);
-                    case Q_DDL_SKIP_REWRITE:
-                        // *start++ = thd->variables.binlog_ddl_skip_rewrite;
-                        buffer.forward(1);
                     case Q_HRNOW:
-                        // https://github.com/alibaba/canal/issues/4940
-                        // percona 和 mariadb各自扩展mysql binlog的格式后有冲突
-                        // 需要精确识别一下数据库类型做兼容处理
-                        if (compatiablePercona) {
-                            // percona 8.0.31
-                            // Q_WSREP_SKIP_READONLY_CHECKS *start++ = 1;
-                            buffer.forward(1);
-                        } else {
                         // int when_sec_part = buffer.getUint24();
                         buffer.forward(3);
-                        }
                         break;
                     case Q_XID:
                         // xid= uint8korr(pos);
@@ -767,6 +728,7 @@ public class QueryLogEvent extends LogEvent {
                         // sa_seq_no = uint8korr(pos);
                         // pos+= 8;
                         // }
+
                         int gtid_flags_extra = buffer.getUint8();
                         final int FL_COMMIT_ALTER_E1= 4;
                         final int FL_ROLLBACK_ALTER_E1= 8;
@@ -774,11 +736,6 @@ public class QueryLogEvent extends LogEvent {
                             buffer.forward(8);
                         }
                         break;
-                    case Q_CHARACTER_SET_COLLATIONS :
-                        // mariadb
-                        int count = buffer.getUint8();
-                        // character_set_collations= Lex_cstring((const char *) pos0 , (const char *) pos);
-                        buffer.forward(count * 4);
                     default:
                         /*
                          * That's why you must write status vars in growing
@@ -832,18 +789,8 @@ public class QueryLogEvent extends LogEvent {
                 return "Q_DEFAULT_COLLATION_FOR_UTF8MB4";
             case Q_SQL_REQUIRE_PRIMARY_KEY:
                 return "Q_SQL_REQUIRE_PRIMARY_KEY";
-            case Q_DEFAULT_TABLE_ENCRYPTION:
-                return "Q_DEFAULT_TABLE_ENCRYPTION";
-            case Q_DDL_SKIP_REWRITE:
-                return "Q_DDL_SKIP_REWRITE";
             case Q_HRNOW:
                 return "Q_HRNOW";
-            case Q_XID:
-                return "Q_XID";
-            case Q_GTID_FLAGS3:
-                return "Q_GTID_FLAGS3";
-            case Q_CHARACTER_SET_COLLATIONS :
-                return "Q_CHARACTER_SET_COLLATIONS";
         }
         return "CODE#" + code;
     }

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

@@ -4,6 +4,8 @@ import java.io.Serializable;
 import java.nio.charset.Charset;
 import java.sql.Timestamp;
 import java.sql.Types;
+import java.time.ZoneId;
+import java.time.zone.ZoneRules;
 import java.util.BitSet;
 
 import org.apache.commons.logging.Log;
@@ -69,7 +71,8 @@ public final class RowsLogBuffer {
     /**
      * Extracting next row from packed buffer.
      * 
-     * @see mysql-5.1.60/sql/log_event.cc - Rows_log_event::print_verbose_one_row
+     * @see mysql-5.1.60/sql/log_event.cc -
+     * Rows_log_event::print_verbose_one_row
      */
     public final boolean nextOneRow(BitSet columns, boolean after) {
         final boolean hasOneRow = buffer.hasRemaining();
@@ -102,7 +105,8 @@ public final class RowsLogBuffer {
     /**
      * Extracting next field value from packed buffer.
      * 
-     * @see mysql-5.1.60/sql/log_event.cc - Rows_log_event::print_verbose_one_row
+     * @see mysql-5.1.60/sql/log_event.cc -
+     * Rows_log_event::print_verbose_one_row
      */
     public final Serializable nextValue(final String columName, final int columnIndex, final int type, final int meta) {
         return nextValue(columName, columnIndex, type, meta, false);
@@ -111,7 +115,8 @@ public final class RowsLogBuffer {
     /**
      * Extracting next field value from packed buffer.
      * 
-     * @see mysql-5.1.60/sql/log_event.cc - Rows_log_event::print_verbose_one_row
+     * @see mysql-5.1.60/sql/log_event.cc -
+     * Rows_log_event::print_verbose_one_row
      */
     public final Serializable nextValue(final String columName, final int columnIndex, final int type, final int meta,
                                         boolean isBinary) {
@@ -295,8 +300,10 @@ public final class RowsLogBuffer {
                             len = byte1;
                             break;
                         default:
-                            throw new IllegalArgumentException(String
+                            throw new IllegalArgumentException(String.format("!! Don't know how to handle column type=%d meta=%d (%04X)",
-                                .format("!! Don't know how to handle column type=%d meta=%d (%04X)", type, meta, meta));
+                                type,
+                                meta,
+                                meta));
                     }
                 }
             } else {
@@ -352,8 +359,8 @@ public final class RowsLogBuffer {
             }
             case LogEvent.MYSQL_TYPE_DECIMAL: {
                 /*
-                 * log_event.h : This enumeration value is only used internally and cannot exist
+                 * log_event.h : This enumeration value is only used internally
-                 * in a binlog.
+                 * and cannot exist in a binlog.
                  */
                 logger.warn("MYSQL_TYPE_DECIMAL : This enumeration value is "
                             + "only used internally and cannot exist in a binlog!");
@@ -534,10 +541,12 @@ public final class RowsLogBuffer {
             }
             case LogEvent.MYSQL_TYPE_DATETIME2: {
                 /*
-                 * DATETIME and DATE low-level memory and disk representation routines 1 bit
+                 * DATETIME and DATE low-level memory and disk representation
-                 * sign (used when on disk) 17 bits year*13+month (year 0-9999, month 0-12) 5
+                 * routines 1 bit sign (used when on disk) 17 bits year*13+month
-                 * bits day (0-31) 5 bits hour (0-23) 6 bits minute (0-59) 6 bits second (0-59)
+                 * (year 0-9999, month 0-12) 5 bits day (0-31) 5 bits hour
-                 * 24 bits microseconds (0-999999) Total: 64 bits = 8 bytes SYYYYYYY.YYYYYYYY
+                 * (0-23) 6 bits minute (0-59) 6 bits second (0-59) 24 bits
+                 * microseconds (0-999999) Total: 64 bits = 8 bytes
+                 * SYYYYYYY.YYYYYYYY
                  * .YYdddddh.hhhhmmmm.mmssssss.ffffffff.ffffffff.ffffffff
                  */
                 long intpart = buffer.getBeUlong40() - DATETIMEF_INT_OFS; // big-endian
@@ -656,10 +665,12 @@ public final class RowsLogBuffer {
             }
             case LogEvent.MYSQL_TYPE_TIME2: {
                 /*
-                 * TIME low-level memory and disk representation routines In-memory format: 1
+                 * TIME low-level memory and disk representation routines
-                 * bit sign (Used for sign, when on disk) 1 bit unused (Reserved for wider hour
+                 * In-memory format: 1 bit sign (Used for sign, when on disk) 1
-                 * range, e.g. for intervals) 10 bit hour (0-836) 6 bit minute (0-59) 6 bit
+                 * bit unused (Reserved for wider hour range, e.g. for
-                 * second (0-59) 24 bits microseconds (0-999999) Total: 48 bits = 6 bytes
+                 * intervals) 10 bit hour (0-836) 6 bit minute (0-59) 6 bit
+                 * second (0-59) 24 bits microseconds (0-999999) Total: 48 bits
+                 * = 6 bytes
                  * Suhhhhhh.hhhhmmmm.mmssssss.ffffffff.ffffffff.ffffffff
                  */
                 long intpart = 0;
@@ -676,15 +687,20 @@ public final class RowsLogBuffer {
                         frac = buffer.getUint8();
                         if (intpart < 0 && frac > 0) {
                             /*
-                             * Negative values are stored with reverse fractional part order, for binary
+                             * Negative values are stored with reverse
-                             * sort compatibility. Disk value intpart frac Time value Memory value 800000.00
+                             * fractional part order, for binary sort
-                             * 0 0 00:00:00.00 0000000000.000000 7FFFFF.FF -1 255 -00:00:00.01
+                             * compatibility. Disk value intpart frac Time value
-                             * FFFFFFFFFF.FFD8F0 7FFFFF.9D -1 99 -00:00:00.99 FFFFFFFFFF.F0E4D0 7FFFFF.00 -1
+                             * Memory value 800000.00 0 0 00:00:00.00
-                             * 0 -00:00:01.00 FFFFFFFFFF.000000 7FFFFE.FF -1 255 -00:00:01.01
+                             * 0000000000.000000 7FFFFF.FF -1 255 -00:00:00.01
-                             * FFFFFFFFFE.FFD8F0 7FFFFE.F6 -2 246 -00:00:01.10 FFFFFFFFFE.FE7960 Formula to
+                             * FFFFFFFFFF.FFD8F0 7FFFFF.9D -1 99 -00:00:00.99
-                             * convert fractional part from disk format (now stored in "frac" variable) to
+                             * FFFFFFFFFF.F0E4D0 7FFFFF.00 -1 0 -00:00:01.00
-                             * absolute value: "0x100 - frac". To reconstruct in-memory value, we shift to
+                             * FFFFFFFFFF.000000 7FFFFE.FF -1 255 -00:00:01.01
-                             * the next integer value and then substruct fractional part.
+                             * FFFFFFFFFE.FFD8F0 7FFFFE.F6 -2 246 -00:00:01.10
+                             * FFFFFFFFFE.FE7960 Formula to convert fractional
+                             * part from disk format (now stored in "frac"
+                             * variable) to absolute value: "0x100 - frac". To
+                             * reconstruct in-memory value, we shift to the next
+                             * integer value and then substruct fractional part.
                              */
                             intpart++; /* Shift to the next integer value */
                             frac -= 0x100; /* -(0x100 - frac) */
@@ -699,8 +715,9 @@ public final class RowsLogBuffer {
                         frac = buffer.getBeUint16();
                         if (intpart < 0 && frac > 0) {
                             /*
-                             * Fix reverse fractional part order: "0x10000 - frac". See comments for FSP=1
+                             * Fix reverse fractional part order:
-                             * and FSP=2 above.
+                             * "0x10000 - frac". See comments for FSP=1 and
+                             * FSP=2 above.
                              */
                             intpart++; /* Shift to the next integer value */
                             frac -= 0x10000; /* -(0x10000-frac) */
@@ -772,8 +789,8 @@ public final class RowsLogBuffer {
             }
             case LogEvent.MYSQL_TYPE_NEWDATE: {
                 /*
-                 * log_event.h : This enumeration value is only used internally and cannot exist
+                 * log_event.h : This enumeration value is only used internally
-                 * in a binlog.
+                 * and cannot exist in a binlog.
                  */
                 logger.warn("MYSQL_TYPE_NEWDATE : This enumeration value is "
                             + "only used internally and cannot exist in a binlog!");
@@ -845,8 +862,8 @@ public final class RowsLogBuffer {
             case LogEvent.MYSQL_TYPE_ENUM: {
                 final int int32;
                 /*
-                 * log_event.h : This enumeration value is only used internally and cannot exist
+                 * log_event.h : This enumeration value is only used internally
-                 * in a binlog.
+                 * and cannot exist in a binlog.
                  */
                 switch (len) {
                     case 1:
@@ -912,24 +929,24 @@ public final class RowsLogBuffer {
             }
             case LogEvent.MYSQL_TYPE_TINY_BLOB: {
                 /*
-                 * log_event.h : This enumeration value is only used internally and cannot exist
+                 * log_event.h : This enumeration value is only used internally
-                 * in a binlog.
+                 * and cannot exist in a binlog.
                  */
                 logger.warn("MYSQL_TYPE_TINY_BLOB : This enumeration value is "
                             + "only used internally and cannot exist in a binlog!");
             }
             case LogEvent.MYSQL_TYPE_MEDIUM_BLOB: {
                 /*
-                 * log_event.h : This enumeration value is only used internally and cannot exist
+                 * log_event.h : This enumeration value is only used internally
-                 * in a binlog.
+                 * and cannot exist in a binlog.
                  */
                 logger.warn("MYSQL_TYPE_MEDIUM_BLOB : This enumeration value is "
                             + "only used internally and cannot exist in a binlog!");
             }
             case LogEvent.MYSQL_TYPE_LONG_BLOB: {
                 /*
-                 * log_event.h : This enumeration value is only used internally and cannot exist
+                 * log_event.h : This enumeration value is only used internally
-                 * in a binlog.
+                 * and cannot exist in a binlog.
                  */
                 logger.warn("MYSQL_TYPE_LONG_BLOB : This enumeration value is "
                             + "only used internally and cannot exist in a binlog!");
@@ -988,7 +1005,8 @@ public final class RowsLogBuffer {
             case LogEvent.MYSQL_TYPE_VAR_STRING: {
                 /*
                  * Except for the data length calculation, MYSQL_TYPE_VARCHAR,
-                 * MYSQL_TYPE_VAR_STRING and MYSQL_TYPE_STRING are handled the same way.
+                 * MYSQL_TYPE_VAR_STRING and MYSQL_TYPE_STRING are handled the
+                 * same way.
                  */
                 len = meta;
                 if (len < 256) {
@@ -1059,19 +1077,30 @@ public final class RowsLogBuffer {
                 if (partialBits.get(1)) {
                     // print_json_diff
                     int position = buffer.position();
-                    try {
+                    StringBuilder builder = JsonDiffConversion.print_json_diff(buffer,
-                        // https://github.com/alibaba/canal/pull/5018
+                        len,
-                        StringBuilder builder = JsonDiffConversion
+                        columnName,
-                            .print_json_diff(buffer, len, columnName, columnIndex, charset);
+                        columnIndex,
+                            charset);
                     value = builder.toString();
                     buffer.position(position + len);
-                    } catch (IllegalArgumentException e) {
-                        buffer.position(position);
-                        // print_json_diff failed, fallback to parse_value
-                        parseJsonFromFullValue(len);
-                    }
                 } else {
-                    parseJsonFromFullValue(len);
+                    if (0 == len) {
+                        // fixed issue #1 by lava, json column of zero length
+                        // has no
+                        // value, value parsing should be skipped
+                        value = "";
+                    } else {
+                        int position = buffer.position();
+                        Json_Value jsonValue = JsonConversion.parse_value(buffer.getUint8(),
+                            buffer,
+                            len - 1,
+                                charset);
+                        StringBuilder builder = new StringBuilder();
+                        jsonValue.toJsonString(builder, charset);
+                        value = builder.toString();
+                        buffer.position(position + len);
+                    }
                 }
                 javaType = Types.VARCHAR;
                 length = len;
@@ -1100,19 +1129,24 @@ public final class RowsLogBuffer {
                 /* fill binary */
                 byte[] binary = new byte[len];
                 buffer.fillBytes(binary, 0, len);
-                // Warning unsupport cloumn type
+
-                // logger.warn(String.format("!! Unsupport column type MYSQL_TYPE_GEOMETRY:
+                /* Warning unsupport cloumn type */
-                // meta=%d (%04X), len = %d", meta,meta, len));
+                // logger.warn(String.format("!! Unsupport column type MYSQL_TYPE_GEOMETRY: meta=%d (%04X), len = %d",
+                // meta,
+                // meta,
+                // len));
                 javaType = Types.BINARY;
                 value = binary;
                 length = len;
                 break;
             }
-            case LogEvent.MYSQL_TYPE_BOOL:
+            case LogEvent.MYSQL_TYPE_BOOL :
-            case LogEvent.MYSQL_TYPE_INVALID:
+            case LogEvent.MYSQL_TYPE_INVALID :
             default:
-                logger.error(
+                logger.error(String.format("!! Don't know how to handle column type=%d meta=%d (%04X)",
-                    String.format("!! Don't know how to handle column type=%d meta=%d (%04X)", type, meta, meta));
+                    type,
+                    meta,
+                    meta));
                 javaType = Types.OTHER;
                 value = null;
                 length = 0;
@@ -1121,21 +1155,6 @@ public final class RowsLogBuffer {
         return value;
     }
 
-    private void parseJsonFromFullValue(int len) {
-        if (0 == len) {
-            // fixed issue #1 by lava, json column of zero length has no value, value
-            // parsing should be skipped
-            value = "";
-        } else {
-            int position = buffer.position();
-            Json_Value jsonValue = JsonConversion.parse_value(buffer.getUint8(), buffer, len - 1, charset);
-            StringBuilder builder = new StringBuilder();
-            jsonValue.toJsonString(builder, charset);
-            value = builder.toString();
-            buffer.position(position + len);
-        }
-    }
-
     public final boolean isNull() {
         return fNull;
     }

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

@@ -1 +1 @@
-package com.taobao.tddl.dbsync.binlog.event.mariadb;
+package com.taobao.tddl.dbsync.binlog.event.mariadb;

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

@@ -8,13 +8,11 @@ import java.util.List;
 
 import com.taobao.tddl.dbsync.binlog.event.mariadb.BinlogCheckPointLogEvent;
 import org.junit.Assert;
-import org.junit.Ignore;
 import org.junit.Test;
 
 import com.taobao.tddl.dbsync.binlog.event.*;
 import com.taobao.tddl.dbsync.binlog.event.mariadb.AnnotateRowsEvent;
 
-@Ignore
 public class DirectLogFetcherTest extends BaseLogFetcherTest {
 
     @Test

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

@@ -1,7 +1,5 @@
 package com.taobao.tddl.dbsync.binlog;
 
-import java.nio.charset.Charset;
-
 import org.junit.Assert;
 import org.junit.Rule;
 import org.junit.Test;
@@ -50,31 +48,7 @@ public class JsonConversion_Json_ValueTest {
         // Act
         thrown.expect(IllegalArgumentException.class);
         JsonConversion.parse_value(type, buffer, len, charsetName);
+
         // Method is not expected to return due to exception thrown
     }
-
-    // test for 5127
-    @Test
-    public void testJsonKeyContainsSpecialCharacter() {
-
-        // {"internal_uri_rewrite": {"(.*)(/[^/]+\\.(mp|MP)4)$": "$1/mp4$2"}}
-        String jsonData = "{\"internal_uri_rewrite\": {\"(.*)(/[^/]+\\\\.(mp|MP)4)$\": \"$1/mp4$2\"}}";
-        byte[] data = new byte[] { 1, 0, 74, 0, 11, 0, 20, 0, 0, 31, 0, 105, 110, 116, 101, 114, 110, 97, 108, 95, 117,
-                                   114, 105, 95, 114, 101, 119, 114, 105, 116, 101, 1, 0, 43, 0, 11, 0, 23, 0, 12, 34,
-                                   0, 40, 46, 42, 41, 40, 47, 91, 94, 47, 93, 43, 92, 46, 40, 109, 112, 124, 77, 80, 41,
-                                   52, 41, 36, 8, 36, 49, 47, 109, 112, 52, 36, 50 };
-        final LogBuffer buffer = new LogBuffer(data, 0, 74);
-
-        Charset charset = Charset.forName("UTF-8");
-        Json_Value jsonValue = JsonConversion.parse_value(0, buffer, 74, charset);
-        StringBuilder builder = new StringBuilder();
-        jsonValue.toJsonString(builder, charset);
-
-        Assert.assertEquals(builder.toString(), jsonData);
-
-        Assert.assertEquals(jsonValue.key(0, charset), "internal_uri_rewrite");
-
-        Json_Value element = jsonValue.element(0, charset);
-        Assert.assertEquals(element.key(0, charset), "(.*)(/[^/]+\\.(mp|MP)4)$");
-    }
 }

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

@@ -37,7 +37,7 @@ public class JsonDiffConversionTest {
         final long len = 71L;
         final String columnName = "foo";
         final int columnIndex = 0;
-        final String charsetName = "gbk";
+        final String charsetName = "foo";
         try {
 
             // Act
@@ -78,7 +78,7 @@ public class JsonDiffConversionTest {
         final long len = 0L;
         final String columnName = "foo";
         final int columnIndex = 0;
-        final String charsetName = "gbk";
+        final String charsetName = "foo";
         try {
 
             // Act
@@ -119,7 +119,7 @@ public class JsonDiffConversionTest {
         final long len = 0L;
         final String columnName = "foo";
         final int columnIndex = 0;
-        final String charsetName = "gbk";
+        final String charsetName = "foo";
         try {
 
             // Act
@@ -160,7 +160,7 @@ public class JsonDiffConversionTest {
         final long len = 0L;
         final String columnName = "foo";
         final int columnIndex = 0;
-        final String charsetName = "gbk";
+        final String charsetName = "foo";
         try {
 
             // Act
@@ -191,7 +191,7 @@ public class JsonDiffConversionTest {
         final long len = 0L;
         final String columnName = ",";
         final int columnIndex = 0;
-        final String charsetName = "gbk";
+        final String charsetName = "1a 2b 3c";
 
         // Act
         final StringBuilder actual = JsonDiffConversion.print_json_diff(buffer,

deployer/pom.xml

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

deployer/src/main/bin/startup.sh

@@ -81,14 +81,14 @@ esac
 
 JavaVersion=`$JAVA -version 2>&1 |awk 'NR==1{ gsub(/"/,""); print $3 }' | awk  -F '.' '{print $1}'`
 str=`file -L $JAVA | grep 64-bit`
-JAVA_OPTS="$JAVA_OPTS -Xss1m -XX:-UseBiasedLocking -XX:-OmitStackTraceInFastThrow -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=$base/logs"
+JAVA_OPTS="$JAVA_OPTS -Xss1m -XX:+AggressiveOpts -XX:-UseBiasedLocking -XX:-OmitStackTraceInFastThrow -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=$base/logs"
 
 if [ $JavaVersion -ge 11 ] ; then
   #JAVA_OPTS="$JAVA_OPTS -Xlog:gc*:$base_log/gc.log:time "
   JAVA_OPTS="$JAVA_OPTS"
 else
   #JAVA_OPTS="$JAVA_OPTS -Xloggc:$base/logs/canal/gc.log -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+PrintGCApplicationStoppedTime"
-  JAVA_OPTS="$JAVA_OPTS -XX:+AggressiveOpts -XX:+UseFastAccessorMethods -XX:+PrintAdaptiveSizePolicy -XX:+PrintTenuringDistribution"
+  JAVA_OPTS="$JAVA_OPTS -XX:+UseFastAccessorMethods -XX:+PrintAdaptiveSizePolicy -XX:+PrintTenuringDistribution"
 fi
 
 if [ -n "$str" ]; then

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

@@ -1,26 +1,5 @@
 package com.alibaba.otter.canal.deployer;
 
-import static com.alibaba.otter.canal.deployer.CanalConstants.CANAL_DESTINATIONS;
-import static com.alibaba.otter.canal.deployer.CanalConstants.CANAL_DESTINATIONS_EXPR;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-import java.util.stream.Collectors;
-
-import org.I0Itec.zkclient.IZkStateListener;
-import org.I0Itec.zkclient.exception.ZkNoNodeException;
-import org.I0Itec.zkclient.exception.ZkNodeExistsException;
-import org.apache.commons.lang.BooleanUtils;
-import org.apache.commons.lang.StringUtils;
-import org.apache.zookeeper.Watcher.Event.KeeperState;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.slf4j.MDC;
-
 import com.alibaba.otter.canal.common.utils.AddressUtils;
 import com.alibaba.otter.canal.common.zookeeper.ZkClientx;
 import com.alibaba.otter.canal.common.zookeeper.ZookeeperPathUtils;
@@ -44,6 +23,26 @@ import com.alibaba.otter.canal.server.netty.CanalServerWithNetty;
 import com.google.common.base.Function;
 import com.google.common.collect.MapMaker;
 import com.google.common.collect.MigrateMap;
+import org.I0Itec.zkclient.IZkStateListener;
+import org.I0Itec.zkclient.exception.ZkNoNodeException;
+import org.I0Itec.zkclient.exception.ZkNodeExistsException;
+import org.apache.commons.lang.BooleanUtils;
+import org.apache.commons.lang.StringUtils;
+import org.apache.zookeeper.Watcher.Event.KeeperState;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.slf4j.MDC;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import static com.alibaba.otter.canal.deployer.CanalConstants.CANAL_DESTINATIONS;
+import static com.alibaba.otter.canal.deployer.CanalConstants.CANAL_DESTINATIONS_EXPR;
 
 /**
  * canal调度控制器
@@ -117,14 +116,8 @@ public class CanalController {
 
         this.adminUser = getProperty(properties, CanalConstants.CANAL_ADMIN_USER);
         this.adminPasswd = getProperty(properties, CanalConstants.CANAL_ADMIN_PASSWD);
-        String user = getProperty(properties, CanalConstants.CANAL_USER);
+        embeddedCanalServer.setUser(getProperty(properties, CanalConstants.CANAL_USER));
-        String passwd = getProperty(properties, CanalConstants.CANAL_PASSWD);
+        embeddedCanalServer.setPasswd(getProperty(properties, CanalConstants.CANAL_PASSWD));
-        if (StringUtils.isNotEmpty(user) && StringUtils.isEmpty(passwd)) {
-            throw new IllegalArgumentException(
-                "canal.user = " + user + " , but canal.passwd is empty , pls check https://github.com/alibaba/canal/issues/4941");
-        }
-        embeddedCanalServer.setUser(user);
-        embeddedCanalServer.setPasswd(passwd);
 
         String canalWithoutNetty = getProperty(properties, CanalConstants.CANAL_WITHOUT_NETTY);
         if (canalWithoutNetty == null || "false".equals(canalWithoutNetty)) {
@@ -608,7 +601,7 @@ public class CanalController {
         ZkClientx.clearClients();
 
         // 需要释放 CanalServerWithEmbedded 否则主线程退出后，进程无法自动完整退出...
-        if (embeddedCanalServer != null && embeddedCanalServer.isStart()) {
+        if (embeddedCanalServer != null) {
             embeddedCanalServer.stop();
         }
     }

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

@@ -54,10 +54,6 @@ public class CanalLauncher {
             if (StringUtils.isNotEmpty(managerAddress)) {
                 String user = CanalController.getProperty(properties, CanalConstants.CANAL_ADMIN_USER);
                 String passwd = CanalController.getProperty(properties, CanalConstants.CANAL_ADMIN_PASSWD);
-                if (StringUtils.isEmpty(passwd)) {
-                    throw new IllegalArgumentException(
-                        "canal.admin.passwd is empty , pls check https://github.com/alibaba/canal/issues/4941");
-                }
                 String adminPort = CanalController.getProperty(properties, CanalConstants.CANAL_ADMIN_PORT, "11110");
                 boolean autoRegister = BooleanUtils.toBoolean(CanalController.getProperty(properties,
                     CanalConstants.CANAL_ADMIN_AUTO_REGISTER));

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

@@ -115,6 +115,7 @@ public class CanalStarter {
             CanalAdminController canalAdmin = new CanalAdminController(this);
             canalAdmin.setUser(user);
             canalAdmin.setPasswd(passwd);
+
             String ip = CanalController.getProperty(properties, CanalConstants.CANAL_IP);
 
             logger.debug("canal admin port:{}, canal admin user:{}, canal admin password: {}, canal ip:{}",

deployer/src/main/resources/canal.properties

@@ -9,13 +9,13 @@ canal.port = 11111
 canal.metrics.pull.port = 11112
 # canal instance user/passwd
 # canal.user = canal
-# canal.passwd =
+# canal.passwd = E3619321C1A937C46A0D8BD1DAC39F93B27D4458
 
 # canal admin config
 #canal.admin.manager = 127.0.0.1:8089
 canal.admin.port = 11110
 canal.admin.user = admin
-canal.admin.passwd =
+canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
 # admin auto register
 #canal.admin.register.auto = true
 #canal.admin.register.cluster =
@@ -176,9 +176,8 @@ rabbitmq.virtual.host =
 rabbitmq.exchange =
 rabbitmq.username =
 rabbitmq.password =
-rabbitmq.queue =
-rabbitmq.routingKey =
 rabbitmq.deliveryMode =
+rabbitmq.queue =
 
 
 ##################################################

deployer/src/main/resources/canal_local.properties

@@ -5,7 +5,7 @@ canal.register.ip =
 canal.admin.manager = 127.0.0.1:8089
 canal.admin.port = 11110
 canal.admin.user = admin
-canal.admin.passwd =
+canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
 # admin auto register
 canal.admin.register.auto = true
 canal.admin.register.cluster =

deployer/src/main/resources/logback.xml

@@ -23,7 +23,7 @@
 						<!-- or whenever the file size reaches 100MB -->
 						<maxFileSize>512MB</maxFileSize>
 					</timeBasedFileNamingAndTriggeringPolicy>
-					<maxHistory>14</maxHistory>
+					<maxHistory>60</maxHistory>
 				</rollingPolicy>
 				<encoder>
 					<pattern>
@@ -50,7 +50,7 @@
 						<!-- or whenever the file size reaches 100MB -->
 						<maxFileSize>32MB</maxFileSize>
 					</timeBasedFileNamingAndTriggeringPolicy>
-					<maxHistory>14</maxHistory>
+					<maxHistory>60</maxHistory>
 				</rollingPolicy>
 				<encoder>
 					<pattern>
@@ -71,7 +71,7 @@
 				<!-- or whenever the file size reaches 100MB -->
 				<maxFileSize>512MB</maxFileSize>
 			</timeBasedFileNamingAndTriggeringPolicy>
-			<maxHistory>14</maxHistory>
+			<maxHistory>60</maxHistory>
 		</rollingPolicy>
 		<encoder charset="UTF-8">
 			<pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{56} - %msg%n</pattern>

docker/base/amd64/Dockerfile

@@ -22,7 +22,7 @@ RUN \
     yum clean all && yum makecache
 
 RUN \
-    /usr/bin/ln -fs /usr/share/zoneinfo/Asia/Shanghai /etc/localtime && \
+    /bin/cp -f /usr/share/zoneinfo/Asia/Shanghai /etc/localtime && \
     echo 'root:Hello1234' | chpasswd && \
     groupadd -r admin && useradd -g admin admin && \
     yum install -y man && \