fix(server/admin): 修复 Canal Admin 认证绕过漏洞 (#2076021)

在 ClientAuthenticationHandler.messageReceived() 的认证失败分支补充
ctx.getChannel().close() 与 return，防止 fall-through 到成功 ACK 分支
导致认证 handler 被移除、后续命令被当作已认证执行。

新增 ClientAuthenticationHandlerTest 覆盖错误密码、seed 为空、
认证成功三条路径。

.github/workflows/maven.yml

@@ -0,0 +1,46 @@
+# This workflow will build a Java project with Maven, and cache/restore any dependencies to improve the workflow execution time
+# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-java-with-maven
+
+# This workflow uses actions that are not certified by GitHub.
+# They are provided by a third-party and are governed by
+# separate terms of service, privacy policy, and support
+# documentation.
+
+name: Java CI with Maven
+
+on:
+  push:
+    branches: [ "master" ]
+  pull_request:
+    branches: [ "master" ]
+
+permissions:
+  contents: read
+
+jobs:
+  test:
+    runs-on: ${{ matrix.os }}
+    permissions:
+      contents: read
+    strategy:
+      matrix:
+        os: [ ubuntu-latest ]
+        java: [ 8, 11, 17, 21 ]
+      fail-fast: false
+      max-parallel: 16
+    name: Test JDK ${{ matrix.java }}, ${{ matrix.os }}
+
+    steps:
+      - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
+      - name: Set up JDK
+        uses: actions/setup-java@c5195efecf7bdfc987ee8bae7a71cb8b11521c00 # v4.7.1
+        with:
+          distribution: 'temurin'
+          java-version: ${{ matrix.java }}
+          cache: 'maven'
+      - name: Build with Maven if test jdk8
+        if: ${{ matrix.java == '8' || matrix.java == '11'}}
+        run: ./mvnw -Pgen-javadoc clean package -B
+      - name: Build with Maven if test jdk17
+        if: ${{ matrix.java == '17' || matrix.java == '21' }}
+        run: ./mvnw -Penable-for-jdk17+,gen-code-cov clean package -B

.gitignore

@@ -7,6 +7,7 @@ test-output/
 .settings/
 tmp
 temp
+logs/
 *.log
 antx.properties
 otter.properties
@@ -18,6 +19,8 @@ jtester.properties
 *.rpm
 client-adapter/example/
 *.dat
+.qwen/
+.vscode/
 conf/
 canal-admin/canal-admin-ui/dist/
 canal-admin/canal-admin-ui/node/

README.md

@@ -57,7 +57,7 @@
 - [Introduction](https://github.com/alibaba/canal/wiki/Introduction)
 - [QuickStart](https://github.com/alibaba/canal/wiki/QuickStart)
   - [Docker QuickStart](https://github.com/alibaba/canal/wiki/Docker-QuickStart)
-  - [Canal Kafka/RocketMQ QuickStart](https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart")
+  - [Canal Kafka/RocketMQ QuickStart](https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart)
   - [Aliyun RDS for MySQL QuickStart](https://github.com/alibaba/canal/wiki/aliyun-RDS-QuickStart)
   - [Prometheus QuickStart](https://github.com/alibaba/canal/wiki/Prometheus-QuickStart)
 - Canal Admin

SECURITY.md

@@ -0,0 +1,118 @@
+# 漏洞奖励计划 
+## 报告
+如果您认为自己在本程序中发现了任何安全（技术）漏洞，欢迎您通过 https://security.alibaba.com 向我们提交漏洞报告。
+如果您报告任何安全漏洞，请注意您可能包含以下信息（合格报告）：
+* git程序URL地址，运行的环境
+* 包含必要屏幕截图的详细说明
+* 重现漏洞的步骤以及修复漏洞的建议。
+* 其他有用信息
+
+## 处理
+ASRC（Alibaba Security Response Center阿里安全响应中心）将尽快审核并回复您的提交内容，并在我们努力修复您提交的漏洞时随时通知您。如有必要，我们可能会与您联系以获取更多信息。
+
+
+## 条款和条件
+1. 仅接受技术漏洞并对其进行评级
+2. 出于安全原因，上报者同意与ASRC合作完成他/她提交的漏洞，不向任何第三方透露任何漏洞信息
+3. 如果不止一个人报告相同的安全漏洞，奖励将给予完成合格报告的第一个人
+4. 为了保护程序的用户，请在修复之前不要直接提交git的issue，也不要在社区讨论任何漏洞信息
+5. 所有奖励和声誉积分将提供给仅向ASRC提交其安全漏洞的上报者
+6. 安全漏洞奖励的解释权利归ASRC所有
+
+## 收集范围
+我们的主要收集漏洞类别是：
+* 服务器端请求伪造（SSRF）
+* SQL注入
+* 拒绝服务攻击
+* 远程执行代码（RCE）
+* XML外部实体攻击（XXE）
+* 访问控制问题（不安全的直接对象参考问题等）
+* 敏感目录遍历问题
+* 本地文件读取（LFD）
+* 敏感信息泄露（密钥，Cookie，Session等）
+
+## 奖励
+* 可直接导致严重问题的每个漏洞奖励7000元人民币
+* 存在限制及需要一定特殊环境下才能利用的问题将给予700-5600元人民币不等的奖励，比如需要用户主动点击才会触发的问题或需要admin权限
+* 只有在指定环境下才可以运行的利用将有可能被收纳但不给予奖励，或直接被忽略，比如只在fastjson+linux特定版本才会出现的问题
+
+## 不在收集范围的报告
+* 影响过时浏览器或平台用户的漏洞
+* Self-XSS
+* 会话固定
+* 内容欺骗
+* 缺少cookie标记
+* 混合内容警告
+* SSL / TLS问题
+* Clickjacking 
+* 基于Flash的漏洞
+* 反射文件下载攻击（RFD）
+* 物理或社会工程攻击
+* 未验证自动化工具或扫描仪的结果
+* 登录/注销/未认证/低影响CSRF
+* 需要MITM或物理访问用户设备的攻击
+* 与网络协议或行业标准相关的问题
+* 不能用于直接攻击的错误信息泄露
+* 缺少与安全相关的HTTP标头等
+
+
+
+
+
+# Vulnerability Reward Program
+## Reporting
+If you believe you have found any security (technical) vulnerability in the Program, you are welcomed to submit a vulnerability report to us at https://security.alibaba.com 
+In case of reporting any security vulnerability, please be noted that you may including following information (Qualified Reporting):
+* The git program URL and running version 
+* A detailed description with necessary screenshots
+* Steps to reappearance the vulnerability and your advice to fix it
+* Other useful information
+
+
+## Processing
+ASRC (Alibaba Security Response Center) will review and respond as quickly as possible to your submission, and keep you informed as we work to fix the vulnerability you submitted. We may contact you for further information if necessary.
+
+
+## Terms and Conditions
+1. ONLY technical vulnerabilities will be accepted and rated.
+2. With regarding to security reasons, reporters agree to cooperate with ASRC exclusively on the vulnerability he/she submitted and not disclose any information of vulnerability to any third-parties.
+3. In the case that more than one person report the same security vulnerability, the reward will be given to the first person who accomplish a Qualified Reporting.
+4. To protect users of the program, please do not directly submit issue on github or discuss anything with the community 
+5. All Rewards and Reputation Credits are given to the reporters who submit his/her security vulnerabilities ONLY to ASRC.
+6. All rights for the security vulnerability rewards are reserved by ASRC.
+
+## Scope of Collecting
+The main categories of vulnerabilities that we are sincerely looking for are:
+* Server-Side Request Forgery (SSRF)
+* SQL Injection
+* Denial of Service Attack
+* Remote Code Execution (RCE)
+* XML External Entity Attacks (XXE)
+* Access Control Issues (Insecure Direct Object Reference issues, etc.)
+* Directory Traversal Issues
+* Local File Disclosure (LFD)
+* Sensitive Information Leakage (Key, Cookie, Session etc.)
+
+## Reward
+* $1,000 for one valid report
+* $100-$800 for Vuls which is limited. For example, Vuls that need user interactions or administrator authority
+* Vuls which only work on the special version will be accepted but no reward, or directly rejected. For example, Vul runs only on a special linux version
+
+## Ineligible Reports
+* Vulnerabilities affecting users of outdated browsers or platforms
+* "Self" XSS
+* Session fixation
+* Content Spoofing
+* Missing cookie flags
+* Mixed content warnings
+* SSL/TLS best practices
+* Clickjacking/UI redressing
+* Flash-based vulnerabilities
+* Reflected file download attacks (RFD)
+* Physical or social engineering attacks
+* Unverified Results of automated tools or scanners
+* Login/logout/unauthenticated/low-impact CSRF
+* Attacks requiring MITM or physical access to a user's device
+* Issues related to networking protocols or industry standards
+* Error information disclosure that cannot be used to make a direct attack
+* Missing security-related HTTP headers which do not lead directly to a vulnerability

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.9-SNAPSHOT</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.9-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>canal.admin-web</artifactId>

admin/admin-web/src/main/bin/startup.bat

@@ -11,7 +11,7 @@ set CLASSPATH=%conf_dir%
 set CLASSPATH=%conf_dir%\..\lib\*;%CLASSPATH%
 
 set JAVA_MEM_OPTS= -Xms128m -Xmx512m
-set JAVA_OPTS_EXT= -Djava.awt.headless=true -Djava.net.preferIPv4Stack=true -Dapplication.codeset=UTF-8 -Dfile.encoding=UTF-8
+set JAVA_OPTS_EXT= -Djava.awt.headless=true -Djava.net.preferIPv4Stack=false -Dapplication.codeset=UTF-8 -Dfile.encoding=UTF-8
 set ADAPTER_OPTS= -DappName=canal-admin
 
 set JAVA_OPTS= %JAVA_MEM_OPTS% %JAVA_OPTS_EXT% %ADAPTER_OPTS%

admin/admin-web/src/main/bin/startup.sh

@@ -79,7 +79,7 @@ else
 	JAVA_OPTS="-server -Xms1024m -Xmx1024m -XX:NewSize=256m -XX:MaxNewSize=256m -XX:MaxPermSize=128m $JAVA_OPTS"
 fi
 
-JAVA_OPTS=" $JAVA_OPTS -Djava.awt.headless=true -Djava.net.preferIPv4Stack=true -Dfile.encoding=UTF-8"
+JAVA_OPTS=" $JAVA_OPTS -Djava.awt.headless=true -Djava.net.preferIPv4Stack=false -Dfile.encoding=UTF-8"
 CANAL_OPTS="-DappName=canal-admin"
 
 for i in $base/lib/*;

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

@@ -98,6 +98,26 @@ public class WebConfig implements WebMvcConfigurer {
                     return false;
                 }
 
+                // 检查是否使用默认密码，如果是则仅允许改密和登出操作
+                User validUser = (User) httpServletRequest.getAttribute("user");
+                if (validUser != null && validUser.isDefaultPassword()) {
+                    String uri = httpServletRequest.getRequestURI();
+                    String method = httpServletRequest.getMethod();
+                    boolean isPasswordChange = "PUT".equals(method) && uri.matches("/api/[^/]+/user");
+                    boolean isLogout = uri.contains("/user/logout");
+                    if (!isPasswordChange && !isLogout) {
+                        BaseModel baseModel = BaseModel.getInstance(null);
+                        baseModel.setCode(50015);
+                        baseModel.setMessage("Default password must be changed before proceeding");
+                        ObjectMapper mapper = new ObjectMapper();
+                        String json = mapper.writeValueAsString(baseModel);
+                        httpServletResponse.setContentType("application/json;charset=UTF-8");
+                        PrintWriter out = httpServletResponse.getWriter();
+                        out.print(json);
+                        return false;
+                    }
+                }
+
                 return true;
             }
         })

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

@@ -56,6 +56,9 @@ public class UserController {
             String token = UUID.randomUUID().toString();
             loginUsers.put(token, loginUser);
             tokenResp.put("token", token);
+            if (loginUser.isDefaultPassword()) {
+                tokenResp.put("needPasswordChange", "true");
+            }
             return BaseModel.getInstance(tokenResp);
         } else {
             BaseModel<Map<String, String>> model = BaseModel.getInstance(null);
@@ -98,6 +101,7 @@ public class UserController {
                                     HttpServletRequest httpServletRequest) {
         userService.update(user);
         String token = (String) httpServletRequest.getAttribute("token");
+        user.setDefaultPassword(false);
         loginUsers.put(token, user);
         return BaseModel.getInstance("success");
     }

admin/admin-web/src/main/java/com/alibaba/otter/canal/admin/model/User.java

@@ -44,6 +44,9 @@ public class User extends Model {
     @Transient
     private String oldPassword;
 
+    @Transient
+    private boolean defaultPassword;
+
     public Long getId() {
         return id;
     }
@@ -115,4 +118,12 @@ public class User extends Model {
     public void setOldPassword(String oldPassword) {
         this.oldPassword = oldPassword;
     }
+
+    public boolean isDefaultPassword() {
+        return defaultPassword;
+    }
+
+    public void setDefaultPassword(boolean defaultPassword) {
+        this.defaultPassword = defaultPassword;
+    }
 }

admin/admin-web/src/main/java/com/alibaba/otter/canal/admin/service/impl/UserServiceImpl.java

@@ -3,6 +3,8 @@ package com.alibaba.otter.canal.admin.service.impl;
 import java.security.NoSuchAlgorithmException;
 
 import org.apache.commons.lang.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
 import com.alibaba.otter.canal.admin.common.exception.ServiceException;
@@ -19,7 +21,11 @@ import com.alibaba.otter.canal.protocol.SecurityUtil;
 @Service
 public class UserServiceImpl implements UserService {
 
-    private static byte[] seeds = "canal is best!".getBytes();
+    private static final Logger  logger                = LoggerFactory.getLogger(UserServiceImpl.class);
+
+    private static final String  DEFAULT_PASSWORD_HASH = "6BB4837EB74329105EE4568DDA7DC67ED2CA2AD9";
+
+    private static byte[]        seeds                 = "canal is best!".getBytes();
 
     private static final Integer PASSWORD_LENGTH = 5;
 
@@ -40,6 +46,11 @@ public class UserServiceImpl implements UserService {
             throw new ServiceException("user:" + user.getName() + " auth failed!");
         }
 
+        if (DEFAULT_PASSWORD_HASH.equalsIgnoreCase(user.getPassword())) {
+            user.setDefaultPassword(true);
+            logger.warn("Default admin password detected for user: {}. Please change immediately.", username);
+        }
+
         user.setPassword("");
         return user;
     }

admin/admin-web/src/main/resources/canal_manager.sql

@@ -11,8 +11,8 @@ SET FOREIGN_KEY_CHECKS = 0;
 DROP TABLE IF EXISTS `canal_adapter_config`;
 CREATE TABLE `canal_adapter_config` (
   `id` bigint(20) NOT NULL AUTO_INCREMENT,
-  `category` varchar(45) NOT NULL,
+  `category` varchar(255) NOT NULL,
-  `name` varchar(45) NOT NULL,
+  `name` varchar(255) NOT NULL,
   `status` varchar(45) DEFAULT NULL,
   `content` text NOT NULL,
   `modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
@@ -25,7 +25,7 @@ CREATE TABLE `canal_adapter_config` (
 DROP TABLE IF EXISTS `canal_cluster`;
 CREATE TABLE `canal_cluster` (
   `id` bigint(20) NOT NULL AUTO_INCREMENT,
-  `name` varchar(63) NOT NULL,
+  `name` varchar(255) NOT NULL,
   `zk_hosts` varchar(255) NOT NULL,
   `modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
   PRIMARY KEY (`id`)
@@ -39,7 +39,7 @@ CREATE TABLE `canal_config` (
   `id` bigint(20) NOT NULL AUTO_INCREMENT,
   `cluster_id` bigint(20) DEFAULT NULL,
   `server_id` bigint(20) DEFAULT NULL,
-  `name` varchar(45) NOT NULL,
+  `name` varchar(255) NOT NULL,
   `status` varchar(45) DEFAULT NULL,
   `content` text NOT NULL,
   `content_md5` varchar(128) NOT NULL,
@@ -56,7 +56,7 @@ CREATE TABLE `canal_instance_config` (
   `id` bigint(20) NOT NULL AUTO_INCREMENT,
   `cluster_id` bigint(20) DEFAULT NULL,
   `server_id` bigint(20) DEFAULT NULL,
-  `name` varchar(45) NOT NULL,
+  `name` varchar(255) NOT NULL,
   `status` varchar(45) DEFAULT NULL,
   `content` text NOT NULL,
   `content_md5` varchar(128) DEFAULT NULL,
@@ -72,8 +72,8 @@ DROP TABLE IF EXISTS `canal_node_server`;
 CREATE TABLE `canal_node_server` (
   `id` bigint(20) NOT NULL AUTO_INCREMENT,
   `cluster_id` bigint(20) DEFAULT NULL,
-  `name` varchar(63) NOT NULL,
+  `name` varchar(255) NOT NULL,
-  `ip` varchar(63) NOT NULL,
+  `ip` varchar(255) NOT NULL,
   `admin_port` int(11) DEFAULT NULL,
   `tcp_port` int(11) DEFAULT NULL,
   `metric_port` int(11) DEFAULT NULL,
@@ -88,10 +88,10 @@ CREATE TABLE `canal_node_server` (
 DROP TABLE IF EXISTS `canal_user`;
 CREATE TABLE `canal_user` (
   `id` bigint(20) NOT NULL AUTO_INCREMENT,
-  `username` varchar(31) NOT NULL,
+  `username` varchar(255) NOT NULL,
-  `password` varchar(128) NOT NULL,
+  `password` varchar(255) NOT NULL,
-  `name` varchar(31) NOT NULL,
+  `name` varchar(255) NOT NULL,
-  `roles` varchar(31) NOT NULL,
+  `roles` varchar(255) NOT NULL,
   `introduction` varchar(255) DEFAULT NULL,
   `avatar` varchar(255) DEFAULT NULL,
   `creation_date` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,

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.9-SNAPSHOT</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.9-SNAPSHOT</version>
     <packaging>pom</packaging>
     <name>canal admin module for otter ${project.version}</name>
     <modules>
@@ -36,7 +36,7 @@
             <dependency>
                 <groupId>org.springframework.boot</groupId>
                 <artifactId>spring-boot-dependencies</artifactId>
-                <version>2.5.4</version>
+                <version>2.5.15</version>
                 <type>pom</type>
                 <scope>import</scope>
             </dependency>
@@ -48,7 +48,12 @@
             <dependency>
                 <groupId>io.ebean</groupId>
                 <artifactId>ebean</artifactId>
-                <version>11.41.1</version>
+                <version>11.45.1</version>
+            </dependency>
+            <dependency>
+                <groupId>org.yaml</groupId>
+                <artifactId>snakeyaml</artifactId>
+                <version>2.0</version>
             </dependency>
         </dependencies>
     </dependencyManagement>

charts/README.md

@@ -0,0 +1,160 @@
+# helm charts 使用说明
+
+包含`canal-admin`和`canal-server`两个chart，基本配置保存在k8s的`secret`里，`canal.properties`和`instance.properties`保存在mysql数据库中，使用zookeeper注册中心实现高可用。
+
+- [helm charts 使用说明](#helm-charts-使用说明)
+  - [创建zk集群](#创建zk集群)
+  - [初始化数据库](#初始化数据库)
+  - [部署canal-admin chart](#部署canal-admin-chart)
+    - [配置values.yaml](#配置valuesyaml)
+    - [配置集群信息](#配置集群信息)
+  - [部署canal-server chart](#部署canal-server-chart)
+    - [配置values.yaml](#配置valuesyaml-1)
+  - [instance 实例创建](#instance-实例创建)
+
+## 创建zk集群
+
+```sh
+helm install canal-zookeeper oci://registry-1.docker.io/bitnamicharts/zookeeper
+```
+
+如需挂载磁盘参考 [bitnami/zookeeper](`https://github.com/bitnami/charts/tree/main/bitnami/zookeeper`)
+
+
+## 初始化数据库
+
+[/admin/admin-web/src/main/resources/canal_manager.sql](/admin/admin-web/src/main/resources/canal_manager.sql)
+
+[/deployer/src/main/resources/spring/tsdb/sql/create_table.sql](/deployer/src/main/resources/spring/tsdb/sql/create_table.sql)
+
+## 部署canal-admin chart
+
+### 配置values.yaml
+
+```yaml
+# 主要配置
+admin:
+  config: |
+    server:
+      port: 8089
+    spring:
+      jackson:
+        date-format: yyyy-MM-dd HH:mm:ss
+        time-zone: GMT+8
+    spring.datasource:
+      address: your_db_host:3306
+      database: canal_manager
+      username: ****
+      password: ****
+      driver-class-name: com.mysql.jdbc.Driver
+      url: jdbc:mysql://${spring.datasource.address}/${spring.datasource.database}?useUnicode=true&characterEncoding=UTF-8&useSSL=false&allowPublicKeyRetrieval=true
+      hikari:
+        maximum-pool-size: 30
+        minimum-idle: 1
+    canal:
+      adminUser: admin
+      adminPasswd: admin
+    # debug: true
+```
+
+全量参数见：[canal-admin/values.yaml](./canal-admin/values.yaml)，测试用例见：[tests/admin-values.yaml](./tests/admin-values.yaml)
+
+```sh
+helm install canal-admin -f ./admin-values.yaml ./canal-admin
+```
+
+### 配置集群信息
+
+![cluster](./images/cluster.png)
+
+等待pod启动后，使用默认账户`admin`密码`123456`登录
+
+创建集群`default`并配置`zk地址`为`canal-zookeeper.default.svc.cluster.local:2181`
+
+配置完成后，点击`配置`-`主配置`，修改以下配置项，不是覆盖整个文件
+
+```yaml
+canal.user = admin
+canal.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
+
+canal.zkServers = canal-zookeeper.default.svc.cluster.local:2181
+# tcp, kafka, rocketMQ, rabbitMQ, pulsarMQ 看实际需求，此处改为了rabbitMQ
+canal.serverMode = rabbitMQ
+
+#canal.instance.tsdb.dir = ${canal.file.data.dir:../conf}/${canal.instance.destination:}
+canal.instance.tsdb.url = jdbc:mysql://you_db_host:3306/canal_tsdb?useUnicode=true&characterEncoding=UTF-8&useSSL=false
+canal.instance.tsdb.dbUsername = ****
+canal.instance.tsdb.dbPassword = ****
+
+#canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml
+canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml
+
+#canal.instance.global.spring.xml = classpath:spring/file-instance.xml
+canal.instance.global.spring.xml = classpath:spring/default-instance.xml
+
+# 如果用的是阿里云polardb需要配置
+canal.aliyun.accessKey = ****
+canal.aliyun.secretKey = ****
+canal.aliyun.uid = 主账号id
+
+# 我们是推送到rabbitmq所以需要配置
+rabbitmq.host = ****:5672
+rabbitmq.virtual.host = ****
+rabbitmq.exchange = ****
+# canal.aliyun.accessKey
+rabbitmq.username = ****
+# canal.aliyun.secretKey
+rabbitmq.password = ****
+rabbitmq.deliveryMode = direct
+```
+
+## 部署canal-server chart
+
+### 配置values.yaml
+
+```yaml
+# 主要配置
+server:
+  config: |
+    canal.port = 11111
+    canal.metrics.pull.port = 11112
+
+    # register ip
+    canal.register.ip =
+
+    # canal admin config
+    canal.admin.manager = canal-admin.default:8089
+    canal.admin.port = 11110
+    canal.admin.user = admin
+    canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
+    # admin auto register
+    canal.admin.register.auto = true
+    canal.admin.register.cluster = default
+```
+
+```sh
+helm install canal-server -f ./server-values.yaml ./canal-server
+```
+
+全量参数见：[canal-server/values.yaml](./canal-server/values.yaml)，测试用例见：[tests/server-values.yaml](./tests/server-values.yaml)
+
+![server](./images/server.png)
+
+至此，canal-server已自动创建，并自动注册。
+
+## instance 实例创建
+
+主要配置如下
+
+```yaml
+# 配置你要监听的数据库
+canal.instance.master.address=127.0.0.1:3306
+canal.instance.dbUsername=****
+canal.instance.dbPassword=****
+
+# 过滤监听哪些表
+canal.instance.filter.regex=.*\\..*
+
+# 如果是推送到rabbitmq，需要配置 Routing Key
+canal.mq.topic=你的Routing Key
+```

charts/canal-admin/.helmignore

@@ -0,0 +1,23 @@
+# Patterns to ignore when building packages.
+# This supports shell glob matching, relative path matching, and
+# negation (prefixed with !). Only one pattern per line.
+.DS_Store
+# Common VCS dirs
+.git/
+.gitignore
+.bzr/
+.bzrignore
+.hg/
+.hgignore
+.svn/
+# Common backup files
+*.swp
+*.bak
+*.tmp
+*.orig
+*~
+# Various IDEs
+.project
+.idea/
+*.tmproj
+.vscode/

charts/canal-admin/Chart.yaml

@@ -0,0 +1,24 @@
+apiVersion: v2
+name: canal-admin
+description: A Helm chart for Kubernetes
+
+# A chart can be either an 'application' or a 'library' chart.
+#
+# Application charts are a collection of templates that can be packaged into versioned archives
+# to be deployed.
+#
+# Library charts provide useful utilities or functions for the chart developer. They're included as
+# a dependency of application charts to inject those utilities and functions into the rendering
+# pipeline. Library charts do not define any templates and therefore cannot be deployed.
+type: application
+
+# This is the chart version. This version number should be incremented each time you make changes
+# to the chart and its templates, including the app version.
+# Versions are expected to follow Semantic Versioning (https://semver.org/)
+version: 1.1.8-alpha-3
+
+# This is the version number of the application being deployed. This version number should be
+# incremented each time you make changes to the application. Versions are not expected to
+# follow Semantic Versioning. They should reflect the version the application is using.
+# It is recommended to use it with quotes.
+appVersion: "1.16.0"

charts/canal-admin/templates/NOTES.txt

@@ -0,0 +1,22 @@
+1. Get the application URL by running these commands:
+{{- if .Values.ingress.enabled }}
+{{- range $host := .Values.ingress.hosts }}
+  {{- range .paths }}
+  http{{ if $.Values.ingress.tls }}s{{ end }}://{{ $host.host }}{{ .path }}
+  {{- end }}
+{{- end }}
+{{- else if contains "NodePort" .Values.service.type }}
+  export NODE_PORT=$(kubectl get --namespace {{ .Release.Namespace }} -o jsonpath="{.spec.ports[0].nodePort}" services {{ include "canal-admin.fullname" . }})
+  export NODE_IP=$(kubectl get nodes --namespace {{ .Release.Namespace }} -o jsonpath="{.items[0].status.addresses[0].address}")
+  echo http://$NODE_IP:$NODE_PORT
+{{- else if contains "LoadBalancer" .Values.service.type }}
+     NOTE: It may take a few minutes for the LoadBalancer IP to be available.
+           You can watch its status by running 'kubectl get --namespace {{ .Release.Namespace }} svc -w {{ include "canal-admin.fullname" . }}'
+  export SERVICE_IP=$(kubectl get svc --namespace {{ .Release.Namespace }} {{ include "canal-admin.fullname" . }} --template "{{"{{ range (index .status.loadBalancer.ingress 0) }}{{.}}{{ end }}"}}")
+  echo http://$SERVICE_IP:{{ .Values.service.port }}
+{{- else if contains "ClusterIP" .Values.service.type }}
+  export POD_NAME=$(kubectl get pods --namespace {{ .Release.Namespace }} -l "app.kubernetes.io/name={{ include "canal-admin.name" . }},app.kubernetes.io/instance={{ .Release.Name }}" -o jsonpath="{.items[0].metadata.name}")
+  export CONTAINER_PORT=$(kubectl get pod --namespace {{ .Release.Namespace }} $POD_NAME -o jsonpath="{.spec.containers[0].ports[0].containerPort}")
+  echo "Visit http://127.0.0.1:8080 to use your application"
+  kubectl --namespace {{ .Release.Namespace }} port-forward $POD_NAME 8080:$CONTAINER_PORT
+{{- end }}

charts/canal-admin/templates/_helpers.tpl

@@ -0,0 +1,62 @@
+{{/*
+Expand the name of the chart.
+*/}}
+{{- define "canal-admin.name" -}}
+{{- default .Chart.Name .Values.nameOverride | trunc 63 | trimSuffix "-" }}
+{{- end }}
+
+{{/*
+Create a default fully qualified app name.
+We truncate at 63 chars because some Kubernetes name fields are limited to this (by the DNS naming spec).
+If release name contains chart name it will be used as a full name.
+*/}}
+{{- define "canal-admin.fullname" -}}
+{{- if .Values.fullnameOverride }}
+{{- .Values.fullnameOverride | trunc 63 | trimSuffix "-" }}
+{{- else }}
+{{- $name := default .Chart.Name .Values.nameOverride }}
+{{- if contains $name .Release.Name }}
+{{- .Release.Name | trunc 63 | trimSuffix "-" }}
+{{- else }}
+{{- printf "%s-%s" .Release.Name $name | trunc 63 | trimSuffix "-" }}
+{{- end }}
+{{- end }}
+{{- end }}
+
+{{/*
+Create chart name and version as used by the chart label.
+*/}}
+{{- define "canal-admin.chart" -}}
+{{- printf "%s-%s" .Chart.Name .Chart.Version | replace "+" "_" | trunc 63 | trimSuffix "-" }}
+{{- end }}
+
+{{/*
+Common labels
+*/}}
+{{- define "canal-admin.labels" -}}
+helm.sh/chart: {{ include "canal-admin.chart" . }}
+{{ include "canal-admin.selectorLabels" . }}
+{{- if .Chart.AppVersion }}
+app.kubernetes.io/version: {{ .Chart.AppVersion | quote }}
+{{- end }}
+app.kubernetes.io/managed-by: {{ .Release.Service }}
+{{- end }}
+
+{{/*
+Selector labels
+*/}}
+{{- define "canal-admin.selectorLabels" -}}
+app.kubernetes.io/name: {{ include "canal-admin.name" . }}
+app.kubernetes.io/instance: {{ .Release.Name }}
+{{- end }}
+
+{{/*
+Create the name of the service account to use
+*/}}
+{{- define "canal-admin.serviceAccountName" -}}
+{{- if .Values.serviceAccount.create }}
+{{- default (include "canal-admin.fullname" .) .Values.serviceAccount.name }}
+{{- else }}
+{{- default "default" .Values.serviceAccount.name }}
+{{- end }}
+{{- end }}

charts/canal-admin/templates/deployment.yaml

@@ -0,0 +1,75 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: {{ include "canal-admin.fullname" . }}
+  labels:
+    {{- include "canal-admin.labels" . | nindent 4 }}
+spec:
+  {{- if not .Values.autoscaling.enabled }}
+  replicas: {{ .Values.replicaCount }}
+  {{- end }}
+  selector:
+    matchLabels:
+      {{- include "canal-admin.selectorLabels" . | nindent 6 }}
+  template:
+    metadata:
+      {{- with .Values.podAnnotations }}
+      annotations:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      labels:
+        {{- include "canal-admin.labels" . | nindent 8 }}
+        {{- with .Values.podLabels }}
+        {{- toYaml . | nindent 8 }}
+        {{- end }}
+    spec:
+      {{- with .Values.imagePullSecrets }}
+      imagePullSecrets:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      serviceAccountName: {{ include "canal-admin.serviceAccountName" . }}
+      securityContext:
+        {{- toYaml .Values.podSecurityContext | nindent 8 }}
+      containers:
+        - name: {{ .Chart.Name }}
+          securityContext:
+            {{- toYaml .Values.securityContext | nindent 12 }}
+          image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}"
+          imagePullPolicy: {{ .Values.image.pullPolicy }}
+          ports:
+            - name: http
+              containerPort: {{ .Values.service.port }}
+              protocol: TCP
+          livenessProbe:
+            {{- toYaml .Values.livenessProbe | nindent 12 }}
+          readinessProbe:
+            {{- toYaml .Values.readinessProbe | nindent 12 }}
+          resources:
+            {{- toYaml .Values.resources | nindent 12 }}
+          volumeMounts:
+            - mountPath: /home/admin/canal-admin/conf/application.yml
+              name: secret
+              subPath: application.yml
+            {{- with .Values.volumeMounts }}
+            {{- toYaml . | nindent 12 }}
+            {{- end }}
+      volumes:
+        - name: secret
+          secret:
+            defaultMode: 420
+            secretName: {{ include "canal-admin.fullname" . }}
+        {{- with .Values.volumes }}
+        {{- toYaml . | nindent 8 }}
+        {{- end }}
+      {{- with .Values.nodeSelector }}
+      nodeSelector:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      {{- with .Values.affinity }}
+      affinity:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      {{- with .Values.tolerations }}
+      tolerations:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}

charts/canal-admin/templates/hpa.yaml

@@ -0,0 +1,32 @@
+{{- if .Values.autoscaling.enabled }}
+apiVersion: autoscaling/v2
+kind: HorizontalPodAutoscaler
+metadata:
+  name: {{ include "canal-admin.fullname" . }}
+  labels:
+    {{- include "canal-admin.labels" . | nindent 4 }}
+spec:
+  scaleTargetRef:
+    apiVersion: apps/v1
+    kind: Deployment
+    name: {{ include "canal-admin.fullname" . }}
+  minReplicas: {{ .Values.autoscaling.minReplicas }}
+  maxReplicas: {{ .Values.autoscaling.maxReplicas }}
+  metrics:
+    {{- if .Values.autoscaling.targetCPUUtilizationPercentage }}
+    - type: Resource
+      resource:
+        name: cpu
+        target:
+          type: Utilization
+          averageUtilization: {{ .Values.autoscaling.targetCPUUtilizationPercentage }}
+    {{- end }}
+    {{- if .Values.autoscaling.targetMemoryUtilizationPercentage }}
+    - type: Resource
+      resource:
+        name: memory
+        target:
+          type: Utilization
+          averageUtilization: {{ .Values.autoscaling.targetMemoryUtilizationPercentage }}
+    {{- end }}
+{{- end }}

charts/canal-admin/templates/ingress.yaml

@@ -0,0 +1,43 @@
+{{- if .Values.ingress.enabled -}}
+apiVersion: networking.k8s.io/v1
+kind: Ingress
+metadata:
+  name: {{ include "canal-admin.fullname" . }}
+  labels:
+    {{- include "canal-admin.labels" . | nindent 4 }}
+  {{- with .Values.ingress.annotations }}
+  annotations:
+    {{- toYaml . | nindent 4 }}
+  {{- end }}
+spec:
+  {{- with .Values.ingress.className }}
+  ingressClassName: {{ . }}
+  {{- end }}
+  {{- if .Values.ingress.tls }}
+  tls:
+    {{- range .Values.ingress.tls }}
+    - hosts:
+        {{- range .hosts }}
+        - {{ . | quote }}
+        {{- end }}
+      secretName: {{ .secretName }}
+    {{- end }}
+  {{- end }}
+  rules:
+    {{- range .Values.ingress.hosts }}
+    - host: {{ .host | quote }}
+      http:
+        paths:
+          {{- range .paths }}
+          - path: {{ .path }}
+            {{- with .pathType }}
+            pathType: {{ . }}
+            {{- end }}
+            backend:
+              service:
+                name: {{ include "canal-admin.fullname" $ }}
+                port:
+                  number: {{ $.Values.service.port }}
+          {{- end }}
+    {{- end }}
+{{- end }}

charts/canal-admin/templates/secret.yaml

@@ -0,0 +1,9 @@
+apiVersion: v1
+kind: Secret
+metadata:
+  name: {{ include "canal-admin.fullname" . }}
+  labels:
+    {{- include "canal-admin.labels" . | nindent 4 }}
+type: Opaque
+data:
+  application.yml: {{tpl .Values.admin.config $ | b64enc | quote }}

charts/canal-admin/templates/service.yaml

@@ -0,0 +1,15 @@
+apiVersion: v1
+kind: Service
+metadata:
+  name: {{ include "canal-admin.fullname" . }}
+  labels:
+    {{- include "canal-admin.labels" . | nindent 4 }}
+spec:
+  type: {{ .Values.service.type }}
+  ports:
+    - port: {{ .Values.service.port }}
+      targetPort: http
+      protocol: TCP
+      name: http
+  selector:
+    {{- include "canal-admin.selectorLabels" . | nindent 4 }}

charts/canal-admin/templates/serviceaccount.yaml

@@ -0,0 +1,13 @@
+{{- if .Values.serviceAccount.create -}}
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: {{ include "canal-admin.serviceAccountName" . }}
+  labels:
+    {{- include "canal-admin.labels" . | nindent 4 }}
+  {{- with .Values.serviceAccount.annotations }}
+  annotations:
+    {{- toYaml . | nindent 4 }}
+  {{- end }}
+automountServiceAccountToken: {{ .Values.serviceAccount.automount }}
+{{- end }}

charts/canal-admin/templates/tests/test-connection.yaml

@@ -0,0 +1,15 @@
+apiVersion: v1
+kind: Pod
+metadata:
+  name: "{{ include "canal-admin.fullname" . }}-test-connection"
+  labels:
+    {{- include "canal-admin.labels" . | nindent 4 }}
+  annotations:
+    "helm.sh/hook": test
+spec:
+  containers:
+    - name: wget
+      image: busybox
+      command: ['wget']
+      args: ['{{ include "canal-admin.fullname" . }}:{{ .Values.service.port }}']
+  restartPolicy: Never

charts/canal-admin/values.yaml

@@ -0,0 +1,146 @@
+# Default values for canal-admin.
+# This is a YAML-formatted file.
+# Declare variables to be passed into your templates.
+
+# This will set the replicaset count more information can be found here: https://kubernetes.io/docs/concepts/workloads/controllers/replicaset/
+replicaCount: 1
+
+# This sets the container image more information can be found here: https://kubernetes.io/docs/concepts/containers/images/
+image:
+  repository: canal/canal-admin
+  # This sets the pull policy for images.
+  pullPolicy: IfNotPresent
+  # Overrides the image tag whose default is the chart appVersion.
+  tag: "latest"
+
+# This is for the secretes for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/
+imagePullSecrets: []
+# This is to override the chart name.
+nameOverride: ""
+fullnameOverride: ""
+
+# This section builds out the service account more information can be found here: https://kubernetes.io/docs/concepts/security/service-accounts/
+serviceAccount:
+  # Specifies whether a service account should be created
+  create: false
+  # Automatically mount a ServiceAccount's API credentials?
+  automount: false
+  # Annotations to add to the service account
+  annotations: {}
+  # The name of the service account to use.
+  # If not set and create is true, a name is generated using the fullname template
+  name: ""
+
+# This is for setting Kubernetes Annotations to a Pod.
+# For more information checkout: https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/
+podAnnotations: {}
+# This is for setting Kubernetes Labels to a Pod.
+# For more information checkout: https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/
+podLabels: {}
+
+podSecurityContext: {}
+  # fsGroup: 2000
+
+securityContext: {}
+  # capabilities:
+  #   drop:
+  #   - ALL
+  # readOnlyRootFilesystem: true
+  # runAsNonRoot: true
+  # runAsUser: 1000
+
+# This is for setting up a service more information can be found here: https://kubernetes.io/docs/concepts/services-networking/service/
+service:
+  # This sets the service type more information can be found here: https://kubernetes.io/docs/concepts/services-networking/service/#publishing-services-service-types
+  type: ClusterIP
+  # This sets the ports more information can be found here: https://kubernetes.io/docs/concepts/services-networking/service/#field-spec-ports
+  port: 8089
+
+# This block is for setting up the ingress for more information can be found here: https://kubernetes.io/docs/concepts/services-networking/ingress/
+ingress:
+  enabled: false
+  className: ""
+  annotations: {}
+    # kubernetes.io/ingress.class: nginx
+    # kubernetes.io/tls-acme: "true"
+  hosts:
+    - host: chart-example.local
+      paths:
+        - path: /
+          pathType: ImplementationSpecific
+  tls: []
+  #  - secretName: chart-example-tls
+  #    hosts:
+  #      - chart-example.local
+
+resources: {}
+  # We usually recommend not to specify default resources and to leave this as a conscious
+  # choice for the user. This also increases chances charts run on environments with little
+  # resources, such as Minikube. If you do want to specify resources, uncomment the following
+  # lines, adjust them as necessary, and remove the curly braces after 'resources:'.
+  # limits:
+  #   cpu: 100m
+  #   memory: 128Mi
+  # requests:
+  #   cpu: 100m
+  #   memory: 128Mi
+
+# This is to setup the liveness and readiness probes more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-startup-probes/
+livenessProbe:
+  httpGet:
+    path: /
+    port: http
+readinessProbe:
+  httpGet:
+    path: /
+    port: http
+
+# This section is for setting up autoscaling more information can be found here: https://kubernetes.io/docs/concepts/workloads/autoscaling/
+autoscaling:
+  enabled: false
+  minReplicas: 1
+  maxReplicas: 100
+  targetCPUUtilizationPercentage: 80
+  # targetMemoryUtilizationPercentage: 80
+
+# Additional volumes on the output Deployment definition.
+volumes: []
+# - name: foo
+#   secret:
+#     secretName: mysecret
+#     optional: false
+
+# Additional volumeMounts on the output Deployment definition.
+volumeMounts: []
+# - name: foo
+#   mountPath: "/etc/foo"
+#   readOnly: true
+
+nodeSelector: {}
+
+tolerations: []
+
+affinity: {}
+
+admin:
+  config: |
+    server:
+      port: 8089
+    spring:
+      jackson:
+        date-format: yyyy-MM-dd HH:mm:ss
+        time-zone: GMT+8
+    spring.datasource:
+      address: you_db_host:3306
+      database: canal_manager
+      username: ****
+      password: ****
+      driver-class-name: com.mysql.jdbc.Driver
+      url: jdbc:mysql://${spring.datasource.address}/${spring.datasource.database}?useUnicode=true&characterEncoding=UTF-8&useSSL=false&allowPublicKeyRetrieval=true
+      hikari:
+        maximum-pool-size: 30
+        minimum-idle: 1
+    canal:
+      adminUser: admin
+      adminPasswd: admin
+    # debug: true

charts/canal-server/.helmignore

@@ -0,0 +1,23 @@
+# Patterns to ignore when building packages.
+# This supports shell glob matching, relative path matching, and
+# negation (prefixed with !). Only one pattern per line.
+.DS_Store
+# Common VCS dirs
+.git/
+.gitignore
+.bzr/
+.bzrignore
+.hg/
+.hgignore
+.svn/
+# Common backup files
+*.swp
+*.bak
+*.tmp
+*.orig
+*~
+# Various IDEs
+.project
+.idea/
+*.tmproj
+.vscode/

charts/canal-server/Chart.yaml

@@ -0,0 +1,24 @@
+apiVersion: v2
+name: canal-server
+description: A Helm chart for Kubernetes
+
+# A chart can be either an 'application' or a 'library' chart.
+#
+# Application charts are a collection of templates that can be packaged into versioned archives
+# to be deployed.
+#
+# Library charts provide useful utilities or functions for the chart developer. They're included as
+# a dependency of application charts to inject those utilities and functions into the rendering
+# pipeline. Library charts do not define any templates and therefore cannot be deployed.
+type: application
+
+# This is the chart version. This version number should be incremented each time you make changes
+# to the chart and its templates, including the app version.
+# Versions are expected to follow Semantic Versioning (https://semver.org/)
+version: 1.1.8-alpha-3
+
+# This is the version number of the application being deployed. This version number should be
+# incremented each time you make changes to the application. Versions are not expected to
+# follow Semantic Versioning. They should reflect the version the application is using.
+# It is recommended to use it with quotes.
+appVersion: "1.16.0"

charts/canal-server/templates/NOTES.txt

@@ -0,0 +1,22 @@
+1. Get the application URL by running these commands:
+{{- if .Values.ingress.enabled }}
+{{- range $host := .Values.ingress.hosts }}
+  {{- range .paths }}
+  http{{ if $.Values.ingress.tls }}s{{ end }}://{{ $host.host }}{{ .path }}
+  {{- end }}
+{{- end }}
+{{- else if contains "NodePort" .Values.service.type }}
+  export NODE_PORT=$(kubectl get --namespace {{ .Release.Namespace }} -o jsonpath="{.spec.ports[0].nodePort}" services {{ include "canal-server.fullname" . }})
+  export NODE_IP=$(kubectl get nodes --namespace {{ .Release.Namespace }} -o jsonpath="{.items[0].status.addresses[0].address}")
+  echo http://$NODE_IP:$NODE_PORT
+{{- else if contains "LoadBalancer" .Values.service.type }}
+     NOTE: It may take a few minutes for the LoadBalancer IP to be available.
+           You can watch its status by running 'kubectl get --namespace {{ .Release.Namespace }} svc -w {{ include "canal-server.fullname" . }}'
+  export SERVICE_IP=$(kubectl get svc --namespace {{ .Release.Namespace }} {{ include "canal-server.fullname" . }} --template "{{"{{ range (index .status.loadBalancer.ingress 0) }}{{.}}{{ end }}"}}")
+  echo http://$SERVICE_IP:{{ .Values.service.port }}
+{{- else if contains "ClusterIP" .Values.service.type }}
+  export POD_NAME=$(kubectl get pods --namespace {{ .Release.Namespace }} -l "app.kubernetes.io/name={{ include "canal-server.name" . }},app.kubernetes.io/instance={{ .Release.Name }}" -o jsonpath="{.items[0].metadata.name}")
+  export CONTAINER_PORT=$(kubectl get pod --namespace {{ .Release.Namespace }} $POD_NAME -o jsonpath="{.spec.containers[0].ports[0].containerPort}")
+  echo "Visit http://127.0.0.1:8080 to use your application"
+  kubectl --namespace {{ .Release.Namespace }} port-forward $POD_NAME 8080:$CONTAINER_PORT
+{{- end }}

charts/canal-server/templates/_helpers.tpl

@@ -0,0 +1,62 @@
+{{/*
+Expand the name of the chart.
+*/}}
+{{- define "canal-server.name" -}}
+{{- default .Chart.Name .Values.nameOverride | trunc 63 | trimSuffix "-" }}
+{{- end }}
+
+{{/*
+Create a default fully qualified app name.
+We truncate at 63 chars because some Kubernetes name fields are limited to this (by the DNS naming spec).
+If release name contains chart name it will be used as a full name.
+*/}}
+{{- define "canal-server.fullname" -}}
+{{- if .Values.fullnameOverride }}
+{{- .Values.fullnameOverride | trunc 63 | trimSuffix "-" }}
+{{- else }}
+{{- $name := default .Chart.Name .Values.nameOverride }}
+{{- if contains $name .Release.Name }}
+{{- .Release.Name | trunc 63 | trimSuffix "-" }}
+{{- else }}
+{{- printf "%s-%s" .Release.Name $name | trunc 63 | trimSuffix "-" }}
+{{- end }}
+{{- end }}
+{{- end }}
+
+{{/*
+Create chart name and version as used by the chart label.
+*/}}
+{{- define "canal-server.chart" -}}
+{{- printf "%s-%s" .Chart.Name .Chart.Version | replace "+" "_" | trunc 63 | trimSuffix "-" }}
+{{- end }}
+
+{{/*
+Common labels
+*/}}
+{{- define "canal-server.labels" -}}
+helm.sh/chart: {{ include "canal-server.chart" . }}
+{{ include "canal-server.selectorLabels" . }}
+{{- if .Chart.AppVersion }}
+app.kubernetes.io/version: {{ .Chart.AppVersion | quote }}
+{{- end }}
+app.kubernetes.io/managed-by: {{ .Release.Service }}
+{{- end }}
+
+{{/*
+Selector labels
+*/}}
+{{- define "canal-server.selectorLabels" -}}
+app.kubernetes.io/name: {{ include "canal-server.name" . }}
+app.kubernetes.io/instance: {{ .Release.Name }}
+{{- end }}
+
+{{/*
+Create the name of the service account to use
+*/}}
+{{- define "canal-server.serviceAccountName" -}}
+{{- if .Values.serviceAccount.create }}
+{{- default (include "canal-server.fullname" .) .Values.serviceAccount.name }}
+{{- else }}
+{{- default "default" .Values.serviceAccount.name }}
+{{- end }}
+{{- end }}

charts/canal-server/templates/hpa.yaml

@@ -0,0 +1,32 @@
+{{- if .Values.autoscaling.enabled }}
+apiVersion: autoscaling/v2
+kind: HorizontalPodAutoscaler
+metadata:
+  name: {{ include "canal-server.fullname" . }}
+  labels:
+    {{- include "canal-server.labels" . | nindent 4 }}
+spec:
+  scaleTargetRef:
+    apiVersion: apps/v1
+    kind: Deployment
+    name: {{ include "canal-server.fullname" . }}
+  minReplicas: {{ .Values.autoscaling.minReplicas }}
+  maxReplicas: {{ .Values.autoscaling.maxReplicas }}
+  metrics:
+    {{- if .Values.autoscaling.targetCPUUtilizationPercentage }}
+    - type: Resource
+      resource:
+        name: cpu
+        target:
+          type: Utilization
+          averageUtilization: {{ .Values.autoscaling.targetCPUUtilizationPercentage }}
+    {{- end }}
+    {{- if .Values.autoscaling.targetMemoryUtilizationPercentage }}
+    - type: Resource
+      resource:
+        name: memory
+        target:
+          type: Utilization
+          averageUtilization: {{ .Values.autoscaling.targetMemoryUtilizationPercentage }}
+    {{- end }}
+{{- end }}

charts/canal-server/templates/secret.yaml

@@ -0,0 +1,9 @@
+apiVersion: v1
+kind: Secret
+metadata:
+  name: {{ include "canal-server.fullname" . }}
+  labels:
+    {{- include "canal-server.labels" . | nindent 4 }}
+type: Opaque
+data:
+  canal.properties: {{tpl .Values.server.config $ | b64enc | quote }}

charts/canal-server/templates/service.yaml

@@ -0,0 +1,23 @@
+apiVersion: v1
+kind: Service
+metadata:
+  name: {{ printf "%s-headless" (include "canal-server.fullname" .) | trunc 63 | trimSuffix "-" }}
+  labels:
+    {{- include "canal-server.labels" . | nindent 4 }}
+spec:
+  type: {{ .Values.service.type }}
+  ports:
+    - port: {{ .Values.service.ports.admin }}
+      targetPort: admin
+      protocol: TCP
+      name: tcp-admin
+    - port: {{ .Values.service.ports.tcp }}
+      targetPort: tcp
+      protocol: TCP
+      name: tcp-tcp
+    - port: {{ .Values.service.ports.metric }}
+      targetPort: metric
+      protocol: TCP
+      name: tcp-metric
+  selector:
+    {{- include "canal-server.selectorLabels" . | nindent 4 }}

charts/canal-server/templates/serviceaccount.yaml

@@ -0,0 +1,13 @@
+{{- if .Values.serviceAccount.create -}}
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: {{ include "canal-server.serviceAccountName" . }}
+  labels:
+    {{- include "canal-server.labels" . | nindent 4 }}
+  {{- with .Values.serviceAccount.annotations }}
+  annotations:
+    {{- toYaml . | nindent 4 }}
+  {{- end }}
+automountServiceAccountToken: {{ .Values.serviceAccount.automount }}
+{{- end }}

charts/canal-server/templates/statefulsets.yaml

@@ -0,0 +1,80 @@
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: {{ include "canal-server.fullname" . }}
+  labels:
+    {{- include "canal-server.labels" . | nindent 4 }}
+spec:
+  serviceName: {{ printf "%s-headless" (include "canal-server.fullname" .) | trunc 63 | trimSuffix "-" }}
+  replicas: {{ .Values.replicaCount }}
+  selector:
+    matchLabels:
+      {{- include "canal-server.selectorLabels" . | nindent 6 }}
+  template:
+    metadata:
+      {{- with .Values.podAnnotations }}
+      annotations:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      labels:
+        {{- include "canal-server.labels" . | nindent 8 }}
+        {{- with .Values.podLabels }}
+        {{- toYaml . | nindent 8 }}
+        {{- end }}
+    spec:
+      {{- with .Values.imagePullSecrets }}
+      imagePullSecrets:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      serviceAccountName: {{ include "canal-server.serviceAccountName" . }}
+      securityContext:
+        {{- toYaml .Values.podSecurityContext | nindent 8 }}
+      containers:
+        - name: {{ .Chart.Name }}
+          securityContext:
+            {{- toYaml .Values.securityContext | nindent 12 }}
+          image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}"
+          imagePullPolicy: {{ .Values.image.pullPolicy }}
+          ports:
+            - name: admin
+              containerPort: {{ .Values.service.ports.admin }}
+              protocol: TCP
+            - name: tcp
+              containerPort: {{ .Values.service.ports.tcp }}
+              protocol: TCP
+            - name: metric
+              containerPort: {{ .Values.service.ports.metric }}
+              protocol: TCP
+          livenessProbe:
+            {{- toYaml .Values.livenessProbe | nindent 12 }}
+          readinessProbe:
+            {{- toYaml .Values.readinessProbe | nindent 12 }}
+          resources:
+            {{- toYaml .Values.resources | nindent 12 }}
+          volumeMounts:
+            - mountPath: /home/admin/canal-server/conf/canal.properties
+              name: secret
+              subPath: canal.properties
+            {{- with .Values.volumeMounts }}
+            {{- toYaml . | nindent 12 }}
+            {{- end }}
+      volumes:
+        - name: secret
+          secret:
+            defaultMode: 420
+            secretName: {{ include "canal-server.fullname" . }}
+        {{- with .Values.volumes }}
+        {{- toYaml . | nindent 8 }}
+        {{- end }}
+      {{- with .Values.nodeSelector }}
+      nodeSelector:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      {{- with .Values.affinity }}
+      affinity:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      {{- with .Values.tolerations }}
+      tolerations:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}

charts/canal-server/templates/tests/test-connection.yaml

@@ -0,0 +1,15 @@
+apiVersion: v1
+kind: Pod
+metadata:
+  name: "{{ include "canal-server.fullname" . }}-test-connection"
+  labels:
+    {{- include "canal-server.labels" . | nindent 4 }}
+  annotations:
+    "helm.sh/hook": test
+spec:
+  containers:
+    - name: wget
+      image: busybox
+      command: ['wget']
+      args: ['{{ include "canal-server.fullname" . }}:{{ .Values.service.port }}']
+  restartPolicy: Never

charts/canal-server/values.yaml

@@ -0,0 +1,144 @@
+# Default values for canal-server.
+# This is a YAML-formatted file.
+# Declare variables to be passed into your templates.
+
+# This will set the replicaset count more information can be found here: https://kubernetes.io/docs/concepts/workloads/controllers/replicaset/
+replicaCount: 1
+
+# This sets the container image more information can be found here: https://kubernetes.io/docs/concepts/containers/images/
+image:
+  repository: canal/canal-server
+  # This sets the pull policy for images.
+  pullPolicy: IfNotPresent
+  # Overrides the image tag whose default is the chart appVersion.
+  tag: "1.1.8-alpha-3"
+
+# This is for the secretes for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/
+imagePullSecrets: []
+# This is to override the chart name.
+nameOverride: ""
+fullnameOverride: ""
+
+# This section builds out the service account more information can be found here: https://kubernetes.io/docs/concepts/security/service-accounts/
+serviceAccount:
+  # Specifies whether a service account should be created
+  create: false
+  # Automatically mount a ServiceAccount's API credentials?
+  automount: false
+  # Annotations to add to the service account
+  annotations: {}
+  # The name of the service account to use.
+  # If not set and create is true, a name is generated using the fullname template
+  name: ""
+
+# This is for setting Kubernetes Annotations to a Pod.
+# For more information checkout: https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/
+podAnnotations: {}
+# This is for setting Kubernetes Labels to a Pod.
+# For more information checkout: https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/
+podLabels: {}
+
+podSecurityContext: {}
+  # fsGroup: 2000
+
+securityContext: {}
+  # capabilities:
+  #   drop:
+  #   - ALL
+  # readOnlyRootFilesystem: true
+  # runAsNonRoot: true
+  # runAsUser: 1000
+
+# This is for setting up a service more information can be found here: https://kubernetes.io/docs/concepts/services-networking/service/
+service:
+  # This sets the service type more information can be found here: https://kubernetes.io/docs/concepts/services-networking/service/#publishing-services-service-types
+  type: ClusterIP
+  # This sets the ports more information can be found here: https://kubernetes.io/docs/concepts/services-networking/service/#field-spec-ports
+  # port: 80
+  ports:
+    admin: 11110
+    tcp: 11111
+    metric: 11112
+
+# This block is for setting up the ingress for more information can be found here: https://kubernetes.io/docs/concepts/services-networking/ingress/
+ingress:
+  enabled: false
+  className: ""
+  annotations: {}
+    # kubernetes.io/ingress.class: nginx
+    # kubernetes.io/tls-acme: "true"
+  hosts:
+    - host: chart-example.local
+      paths:
+        - path: /
+          pathType: ImplementationSpecific
+  tls: []
+  #  - secretName: chart-example-tls
+  #    hosts:
+  #      - chart-example.local
+
+resources: {}
+  # We usually recommend not to specify default resources and to leave this as a conscious
+  # choice for the user. This also increases chances charts run on environments with little
+  # resources, such as Minikube. If you do want to specify resources, uncomment the following
+  # lines, adjust them as necessary, and remove the curly braces after 'resources:'.
+  # limits:
+  #   cpu: 100m
+  #   memory: 128Mi
+  # requests:
+  #   cpu: 100m
+  #   memory: 128Mi
+
+# This is to setup the liveness and readiness probes more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-startup-probes/
+livenessProbe:
+  httpGet:
+    path: /
+    port: metric
+readinessProbe:
+  httpGet:
+    path: /
+    port: metric
+
+# This section is for setting up autoscaling more information can be found here: https://kubernetes.io/docs/concepts/workloads/autoscaling/
+autoscaling:
+  enabled: false
+  minReplicas: 1
+  maxReplicas: 100
+  targetCPUUtilizationPercentage: 80
+  # targetMemoryUtilizationPercentage: 80
+
+# Additional volumes on the output Deployment definition.
+volumes: []
+# - name: foo
+#   secret:
+#     secretName: mysecret
+#     optional: false
+
+# Additional volumeMounts on the output Deployment definition.
+volumeMounts: []
+# - name: foo
+#   mountPath: "/etc/foo"
+#   readOnly: true
+
+nodeSelector: {}
+
+tolerations: []
+
+affinity: {}
+
+server:
+  config: |
+    canal.port = 11111
+    canal.metrics.pull.port = 11112
+
+    # register ip
+    canal.register.ip =
+
+    # canal admin config
+    canal.admin.manager = canal-admin.default:8089
+    canal.admin.port = 11110
+    canal.admin.user = admin
+    canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
+    # admin auto register
+    canal.admin.register.auto = true
+    canal.admin.register.cluster = default

charts/tests/admin-values.yaml

@@ -0,0 +1,34 @@
+ingress:
+  enabled: true
+  hosts:
+    - host: admin.canal.com
+      paths:
+        - path: /
+          pathType: ImplementationSpecific
+
+admin:
+  config: |
+    server:
+      port: 8089
+    spring:
+      jackson:
+        date-format: yyyy-MM-dd HH:mm:ss
+        time-zone: GMT+8
+    spring.datasource:
+      address: you_db_host:3306
+      database: canal_manager
+      username: ****
+      password: ****
+      driver-class-name: com.mysql.jdbc.Driver
+      url: jdbc:mysql://${spring.datasource.address}/${spring.datasource.database}?useUnicode=true&characterEncoding=UTF-8&useSSL=false&allowPublicKeyRetrieval=true
+      hikari:
+        maximum-pool-size: 30
+        minimum-idle: 1
+    canal:
+      adminUser: admin
+      adminPasswd: admin
+    debug: true
+    
+image:
+  repository: registry.us-east-1.aliyuncs.com/sll0/canal-admin
+  tag: "latest"

charts/tests/server-values.yaml

@@ -0,0 +1,20 @@
+server:
+  config: |
+    canal.port = 11111
+    canal.metrics.pull.port = 11112
+
+    # register ip
+    canal.register.ip =
+
+    # canal admin config
+    canal.admin.manager = canal-admin.default:8089
+    canal.admin.port = 11110
+    canal.admin.user = admin
+    canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
+    # admin auto register
+    canal.admin.register.auto = true
+    canal.admin.register.cluster = default
+    
+image:
+  repository: registry.us-east-1.aliyuncs.com/sll0/canal-server
+  tag: "v1.1.8-alpha-3"

charts/tests/test.sh

@@ -0,0 +1,5 @@
+#! /bin/bash
+
+helm install canal-admin -f ./admin-values.yaml ../canal-admin --dry-run
+
+helm install canal-server -f ./server-values.yaml ../canal-server --dry-run

client-adapter/clickhouse/pom.xml

@@ -1,11 +1,9 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0"
+<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">
-         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>
+        <version>1.1.9-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>
@@ -60,7 +58,9 @@
                 </executions>
             </plugin>
             <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-antrun-plugin</artifactId>
+                <version>1.8</version>
                 <executions>
                     <execution>
                         <phase>package</phase>

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

@@ -40,7 +40,7 @@ public class ClickHouseEtlService extends AbstractEtlService {
     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());
+        String sql = "SELECT * FROM " + SyncUtil.getSourceDbTableName(dbMapping, dataSource.getDbType());
         return importData(sql, params);
     }
 
@@ -126,13 +126,13 @@ public class ClickHouseEtlService extends AbstractEtlService {
 
                             int i = 1;
                             for (Map.Entry<String, String> entry : columnsMap.entrySet()) {
-                                String targetClolumnName = entry.getKey();
+                                String targetColumnName = entry.getKey();
                                 String srcColumnName = entry.getValue();
                                 if (srcColumnName == null) {
-                                    srcColumnName = targetClolumnName;
+                                    srcColumnName = targetColumnName;
                                 }
 
-                                Integer type = columnType.get(targetClolumnName.toLowerCase());
+                                Integer type = columnType.get(targetColumnName.toLowerCase());
                                 Object value = rs.getObject(srcColumnName);
                                 if (value != null) {
                                     SyncUtil.setPStmt(type, pstmt, value, i);

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

@@ -268,6 +268,17 @@ public class SyncUtil {
         return result;
     }
 
+    public static String getSourceDbTableName(MappingConfig.DbMapping dbMapping, String dbType) {
+        String result = "";
+        String backtick = getBacktickByDbType(dbType);
+        if (StringUtils.isNotEmpty(dbMapping.getDatabase())) {
+            result += (backtick + dbMapping.getDatabase() + backtick + ".");
+        }
+
+        result += (backtick + dbMapping.getTable() + backtick);
+        return result;
+    }
+
     /**
      * 根据DbType返回反引号或空字符串
      *

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.9-SNAPSHOT</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

@@ -16,6 +16,7 @@ import java.util.function.Function;
 
 import javax.sql.DataSource;
 
+import com.alibaba.druid.pool.DruidDataSource;
 import org.apache.commons.lang.StringUtils;
 import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
@@ -38,7 +39,17 @@ public class Util {
     public static Object sqlRS(DataSource ds, String sql, Function<ResultSet, Object> fun) {
         try (Connection conn = ds.getConnection();
                 Statement stmt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY)) {
-            stmt.setFetchSize(Integer.MIN_VALUE);
+
+            DruidDataSource druidDataSource = (DruidDataSource) ds;
+            if ("postgresql".equals(druidDataSource.getDbType())) {
+                conn.setAutoCommit(false);
+                stmt.setFetchSize(1000);
+            } else if ("sqlserver".equals(druidDataSource.getDbType())) {
+                stmt.setFetchSize(100);
+            } else {
+                stmt.setFetchSize(Integer.MIN_VALUE);
+            }
+
             try (ResultSet rs = stmt.executeQuery(sql)) {
                 return fun.apply(rs);
             }
@@ -52,7 +63,13 @@ public class Util {
         try (Connection conn = ds.getConnection()) {
             try (PreparedStatement pstmt = conn
                 .prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY)) {
-                pstmt.setFetchSize(Integer.MIN_VALUE);
+                DruidDataSource druidDataSource = (DruidDataSource) ds;
+                if ("postgresql".equals(druidDataSource.getDbType())) {
+                    conn.setAutoCommit(false);
+                    pstmt.setFetchSize(1000);
+                } else {
+                    pstmt.setFetchSize(Integer.MIN_VALUE);
+                }
                 if (values != null) {
                     for (int i = 0; i < values.size(); i++) {
                         pstmt.setObject(i + 1, values.get(i));

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

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

client-adapter/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.9-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>
@@ -74,7 +74,9 @@
                 </executions>
             </plugin>
             <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-antrun-plugin</artifactId>
+                <version>1.8</version>
                 <executions>
                     <execution>
                         <phase>package</phase>

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.9-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>
@@ -74,7 +74,9 @@
                 </executions>
             </plugin>
             <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-antrun-plugin</artifactId>
+                <version>1.8</version>
                 <executions>
                     <execution>
                         <phase>package</phase>

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

@@ -39,13 +39,10 @@ public class ES7xTemplate implements ESTemplate {
         .getLogger(ESTemplate.class);
 
     private static final int                                  MAX_BATCH_SIZE = 1000;
-
-    private ESConnection                                      esConnection;
-
-    private ESBulkRequest                                     esBulkRequest;
-
     // es 字段类型本地缓存
     private static ConcurrentMap<String, Map<String, String>> esFieldTypes   = new ConcurrentHashMap<>();
+    private ESConnection                                      esConnection;
+    private ESBulkRequest                                     esBulkRequest;
 
     public ES7xTemplate(ESConnection esConnection){
         this.esConnection = esConnection;
@@ -65,8 +62,9 @@ public class ES7xTemplate implements ESTemplate {
         if (mapping.getId() != null) {
             String parentVal = (String) esFieldData.remove("$parent_routing");
             if (mapping.isUpsert()) {
-                ESUpdateRequest updateRequest = esConnection.new ES7xUpdateRequest(mapping.getIndex(),
+                ESUpdateRequest updateRequest = esConnection.new ES7xUpdateRequest(mapping.getIndex(), pkVal.toString())
-                    pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true);
+                    .setDoc(esFieldData)
+                    .setDocAsUpsert(true);
                 if (StringUtils.isNotEmpty(parentVal)) {
                     updateRequest.setRouting(parentVal);
                 }
@@ -283,7 +281,8 @@ public class ES7xTemplate implements ESTemplate {
     }
 
     @Override
-    public Object getESDataFromDmlData(ESMapping mapping, Map<String, Object> dmlData, Map<String, Object> esFieldData) {
+    public Object getESDataFromDmlData(ESMapping mapping, Map<String, Object> dmlData,
+                                       Map<String, Object> esFieldData) {
         SchemaItem schemaItem = mapping.getSchemaItem();
         String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
         Object resultIdVal = null;
@@ -307,8 +306,8 @@ public class ES7xTemplate implements ESTemplate {
     }
 
     @Override
-    public Object getESDataFromDmlData(ESMapping mapping,String owner, Map<String, Object> dmlData, Map<String, Object> dmlOld,
+    public Object getESDataFromDmlData(ESMapping mapping, String owner, Map<String, Object> dmlData,
-                                       Map<String, Object> esFieldData) {
+                                       Map<String, Object> dmlOld, Map<String, Object> esFieldData) {
         SchemaItem schemaItem = mapping.getSchemaItem();
         String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
         Object resultIdVal = null;
@@ -329,7 +328,7 @@ public class ES7xTemplate implements ESTemplate {
 
             if (dmlOld.containsKey(columnName) && !mapping.getSkips().contains(fieldItem.getFieldName())) {
                 esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()),
-                        getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName));
+                    getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName));
             }
         }
 

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

@@ -55,16 +55,9 @@ import com.alibaba.otter.canal.client.adapter.es.core.support.ESBulkRequest;
 public class ESConnection {
 
     private static final Logger logger = LoggerFactory.getLogger(ESConnection.class);
-
-    public enum ESClientMode {
-        TRANSPORT, REST
-    }
-
     private ESClientMode        mode;
-
     @SuppressWarnings("deprecation")
     private TransportClient     transportClient;
-
     private RestHighLevelClient restHighLevelClient;
 
     public ESConnection(String[] hosts, Map<String, String> properties, ESClientMode mode) throws UnknownHostException{
@@ -86,9 +79,10 @@ public class ESConnection {
             if (StringUtils.isNotEmpty(nameAndPwd) && nameAndPwd.contains(":")) {
                 String[] nameAndPwdArr = nameAndPwd.split(":");
                 final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
-                credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(nameAndPwdArr[0],
+                credentialsProvider.setCredentials(AuthScope.ANY,
-                    nameAndPwdArr[1]));
+                    new UsernamePasswordCredentials(nameAndPwdArr[0], nameAndPwdArr[1]));
-                restClientBuilder.setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider));
+                restClientBuilder.setHttpClientConfigCallback(
+                    httpClientBuilder -> httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider));
             }
             restHighLevelClient = new RestHighLevelClient(restClientBuilder);
         }
@@ -138,11 +132,94 @@ public class ESConnection {
                 logger.error(e.getMessage(), e);
                 return null;
             }
+
+            // 通过别名查询mapping返回的是真实索引名称，mappings.get(index)返回null，为兼容别名情况修改如下：
             mappingMetaData = mappings.get(index);
+            if (mappingMetaData == null && !mappings.isEmpty()) {
+                // 如果通过索引名找不到，可能是别名，取第一个映射
+                mappingMetaData = mappings.values().iterator().next();
+            }
+
+            // 如果还是null，说明索引不存在或没有mapping
+            if (mappingMetaData == null) {
+                throw new IllegalArgumentException("Not found the mapping info of index: " + index);
+            }
         }
         return mappingMetaData;
     }
 
+    // ------ get/set ------
+    public ESClientMode getMode() {
+        return mode;
+    }
+
+    public void setMode(ESClientMode mode) {
+        this.mode = mode;
+    }
+
+    public TransportClient getTransportClient() {
+        return transportClient;
+    }
+
+    public void setTransportClient(TransportClient transportClient) {
+        this.transportClient = transportClient;
+    }
+
+    public RestHighLevelClient getRestHighLevelClient() {
+        return restHighLevelClient;
+    }
+
+    public void setRestHighLevelClient(RestHighLevelClient restHighLevelClient) {
+        this.restHighLevelClient = restHighLevelClient;
+    }
+
+    private HttpHost createHttpHost(String uriStr) {
+        URI uri = URI.create(uriStr);
+        if (!org.springframework.util.StringUtils.hasLength(uri.getUserInfo())) {
+            return HttpHost.create(uri.toString());
+        }
+        try {
+            return HttpHost.create(new URI(uri
+                .getScheme(), null, uri.getHost(), uri.getPort(), uri.getPath(), uri.getQuery(), uri.getFragment())
+                    .toString());
+        } catch (URISyntaxException ex) {
+            throw new IllegalStateException(ex);
+        }
+    }
+
+    public enum ESClientMode {
+                              TRANSPORT, REST
+    }
+
+    public static class ES7xBulkResponse implements ESBulkRequest.ESBulkResponse {
+
+        private BulkResponse bulkResponse;
+
+        public ES7xBulkResponse(BulkResponse bulkResponse){
+            this.bulkResponse = bulkResponse;
+        }
+
+        @Override
+        public boolean hasFailures() {
+            return bulkResponse.hasFailures();
+        }
+
+        @Override
+        public void processFailBulkResponse(String errorMsg) {
+            for (BulkItemResponse itemResponse : bulkResponse.getItems()) {
+                if (!itemResponse.isFailed()) {
+                    continue;
+                }
+
+                if (itemResponse.getFailure().getStatus() == RestStatus.NOT_FOUND) {
+                    logger.error(itemResponse.getFailureMessage());
+                } else {
+                    throw new RuntimeException(errorMsg + itemResponse.getFailureMessage());
+                }
+            }
+        }
+    }
+
     public class ES7xIndexRequest implements ESBulkRequest.ESIndexRequest {
 
         private IndexRequestBuilder indexRequestBuilder;
@@ -443,71 +520,4 @@ public class ESConnection {
             this.bulkRequest = bulkRequest;
         }
     }
-
-    public static class ES7xBulkResponse implements ESBulkRequest.ESBulkResponse {
-
-        private BulkResponse bulkResponse;
-
-        public ES7xBulkResponse(BulkResponse bulkResponse){
-            this.bulkResponse = bulkResponse;
-        }
-
-        @Override
-        public boolean hasFailures() {
-            return bulkResponse.hasFailures();
-        }
-
-        @Override
-        public void processFailBulkResponse(String errorMsg) {
-            for (BulkItemResponse itemResponse : bulkResponse.getItems()) {
-                if (!itemResponse.isFailed()) {
-                    continue;
-                }
-
-                if (itemResponse.getFailure().getStatus() == RestStatus.NOT_FOUND) {
-                    logger.error(itemResponse.getFailureMessage());
-                } else {
-                    throw new RuntimeException(errorMsg + itemResponse.getFailureMessage());
-                }
-            }
-        }
-    }
-
-    // ------ get/set ------
-    public ESClientMode getMode() {
-        return mode;
-    }
-
-    public void setMode(ESClientMode mode) {
-        this.mode = mode;
-    }
-
-    public TransportClient getTransportClient() {
-        return transportClient;
-    }
-
-    public void setTransportClient(TransportClient transportClient) {
-        this.transportClient = transportClient;
-    }
-
-    public RestHighLevelClient getRestHighLevelClient() {
-        return restHighLevelClient;
-    }
-
-    public void setRestHighLevelClient(RestHighLevelClient restHighLevelClient) {
-        this.restHighLevelClient = restHighLevelClient;
-    }
-
-    private HttpHost createHttpHost(String uriStr) {
-        URI uri = URI.create(uriStr);
-        if (!org.springframework.util.StringUtils.hasLength(uri.getUserInfo())) {
-            return HttpHost.create(uri.toString());
-        }
-        try {
-            return HttpHost.create(new URI(uri.getScheme(), null, uri.getHost(), uri.getPort(), uri.getPath(),
-                                           uri.getQuery(), uri.getFragment()).toString());
-        } catch (URISyntaxException ex) {
-            throw new IllegalStateException(ex);
-        }
-    }
 }

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.9-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>
@@ -69,7 +69,9 @@
                 </executions>
             </plugin>
             <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-antrun-plugin</artifactId>
+                <version>1.8</version>
                 <executions>
                     <execution>
                         <phase>package</phase>

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

@@ -1,5 +1,14 @@
 package com.alibaba.otter.canal.client.adapter.es8x;
 
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import javax.sql.DataSource;
+
+import org.elasticsearch.action.search.SearchResponse;
+
 import com.alibaba.otter.canal.client.adapter.es.core.ESAdapter;
 import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig;
 import com.alibaba.otter.canal.client.adapter.es8x.etl.ESEtlService;
@@ -9,13 +18,6 @@ import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
 import com.alibaba.otter.canal.client.adapter.support.EtlResult;
 import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig;
 import com.alibaba.otter.canal.client.adapter.support.SPI;
-import org.elasticsearch.action.search.SearchResponse;
-
-import javax.sql.DataSource;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
 
 /**
  * ES 8.x 外部适配器

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

@@ -39,13 +39,10 @@ public class ES8xTemplate implements ESTemplate {
         .getLogger(ESTemplate.class);
 
     private static final int                                  MAX_BATCH_SIZE = 1000;
-
-    private ESConnection                                      esConnection;
-
-    private ESBulkRequest                                     esBulkRequest;
-
     // es 字段类型本地缓存
     private static ConcurrentMap<String, Map<String, String>> esFieldTypes   = new ConcurrentHashMap<>();
+    private ESConnection                                      esConnection;
+    private ESBulkRequest                                     esBulkRequest;
 
     public ES8xTemplate(ESConnection esConnection){
         this.esConnection = esConnection;

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

@@ -142,11 +142,73 @@ public class ESConnection {
             logger.error(e.getMessage(), e);
             return null;
         }
+
+        // 通过别名查询mapping返回的是真实索引名称，mappings.get(index)返回null，为兼容别名情况修改如下：
         mappingMetaData = mappings.get(index);
+        if (mappingMetaData == null && !mappings.isEmpty()) {
+            // 如果通过索引名找不到，可能是别名，取第一个映射
+            mappingMetaData = mappings.values().iterator().next();
+        }
+
+        // 如果还是null，说明索引不存在或没有mapping
+        if (mappingMetaData == null) {
+            throw new IllegalArgumentException("Not found the mapping info of index: " + index);
+        }
 
         return mappingMetaData;
     }
 
+    public RestHighLevelClient getRestHighLevelClient() {
+        return restHighLevelClient;
+    }
+
+    public void setRestHighLevelClient(RestHighLevelClient restHighLevelClient) {
+        this.restHighLevelClient = restHighLevelClient;
+    }
+
+    private HttpHost createHttpHost(String uriStr) {
+        URI uri = URI.create(uriStr);
+        if (!org.springframework.util.StringUtils.hasLength(uri.getUserInfo())) {
+            return HttpHost.create(uri.toString());
+        }
+        try {
+            return HttpHost.create(new URI(uri
+                .getScheme(), null, uri.getHost(), uri.getPort(), uri.getPath(), uri.getQuery(), uri.getFragment())
+                    .toString());
+        } catch (URISyntaxException ex) {
+            throw new IllegalStateException(ex);
+        }
+    }
+
+    public static class ES8xBulkResponse implements ESBulkRequest.ESBulkResponse {
+
+        private BulkResponse bulkResponse;
+
+        public ES8xBulkResponse(BulkResponse bulkResponse){
+            this.bulkResponse = bulkResponse;
+        }
+
+        @Override
+        public boolean hasFailures() {
+            return bulkResponse.hasFailures();
+        }
+
+        @Override
+        public void processFailBulkResponse(String errorMsg) {
+            for (BulkItemResponse itemResponse : bulkResponse.getItems()) {
+                if (!itemResponse.isFailed()) {
+                    continue;
+                }
+
+                if (itemResponse.getFailure().getStatus() == RestStatus.NOT_FOUND) {
+                    logger.error(itemResponse.getFailureMessage());
+                } else {
+                    throw new RuntimeException(errorMsg + itemResponse.getFailureMessage());
+                }
+            }
+        }
+    }
+
     public class ES8xIndexRequest implements ESBulkRequest.ESIndexRequest {
 
         private IndexRequestBuilder indexRequestBuilder;
@@ -395,55 +457,4 @@ public class ESConnection {
             this.bulkRequest = bulkRequest;
         }
     }
-
-    public static class ES8xBulkResponse implements ESBulkRequest.ESBulkResponse {
-
-        private BulkResponse bulkResponse;
-
-        public ES8xBulkResponse(BulkResponse bulkResponse){
-            this.bulkResponse = bulkResponse;
-        }
-
-        @Override
-        public boolean hasFailures() {
-            return bulkResponse.hasFailures();
-        }
-
-        @Override
-        public void processFailBulkResponse(String errorMsg) {
-            for (BulkItemResponse itemResponse : bulkResponse.getItems()) {
-                if (!itemResponse.isFailed()) {
-                    continue;
-                }
-
-                if (itemResponse.getFailure().getStatus() == RestStatus.NOT_FOUND) {
-                    logger.error(itemResponse.getFailureMessage());
-                } else {
-                    throw new RuntimeException(errorMsg + itemResponse.getFailureMessage());
-                }
-            }
-        }
-    }
-
-    public RestHighLevelClient getRestHighLevelClient() {
-        return restHighLevelClient;
-    }
-
-    public void setRestHighLevelClient(RestHighLevelClient restHighLevelClient) {
-        this.restHighLevelClient = restHighLevelClient;
-    }
-
-    private HttpHost createHttpHost(String uriStr) {
-        URI uri = URI.create(uriStr);
-        if (!org.springframework.util.StringUtils.hasLength(uri.getUserInfo())) {
-            return HttpHost.create(uri.toString());
-        }
-        try {
-            return HttpHost.create(new URI(uri
-                .getScheme(), null, uri.getHost(), uri.getPort(), uri.getPath(), uri.getQuery(), uri.getFragment())
-                    .toString());
-        } catch (URISyntaxException ex) {
-            throw new IllegalStateException(ex);
-        }
-    }
 }

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.9-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/support/ESSyncUtil.java

@@ -2,6 +2,7 @@ package com.alibaba.otter.canal.client.adapter.es.core.support;
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.math.BigDecimal;
 import java.nio.charset.StandardCharsets;
 import java.sql.Blob;
 import java.sql.SQLException;
@@ -101,13 +102,22 @@ public class ESSyncUtil {
                 break;
             case "float":
             case "half_float":
-            case "scaled_float":
                 if (val instanceof Number) {
                     res = ((Number) val).floatValue();
                 } else {
                     res = Float.parseFloat(val.toString());
                 }
                 break;
+            case "scaled_float":
+                // scaled_float 需要保持高精度，使用 BigDecimal 进行处理
+                if (val instanceof BigDecimal) {
+                    res = val; // 保持 BigDecimal 精度
+                } else if (val instanceof Number) {
+                    res = new BigDecimal(val.toString());
+                } else {
+                    res = new BigDecimal(val.toString());
+                }
+                break;
             case "boolean":
                 if (val instanceof Boolean) {
                     res = val;

client-adapter/escore/src/test/java/com/alibaba/otter/canal/client/adapter/es/core/support/ESSyncUtilTest.java

@@ -0,0 +1,76 @@
+package com.alibaba.otter.canal.client.adapter.es.core.support;
+
+import java.math.BigDecimal;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * ESSyncUtil 测试类
+ * 
+ * @author jianghang
+ */
+public class ESSyncUtilTest {
+
+    @Test
+    public void testScaledFloatPrecision() {
+        // 测试高精度 decimal 转换为 scaled_float
+        BigDecimal originalValue = new BigDecimal("1234567890.12345678901234567890");
+
+        // 转换为 scaled_float
+        Object result = ESSyncUtil.typeConvert(originalValue, "scaled_float");
+
+        // 验证结果仍然是 BigDecimal，保持精度
+        Assert.assertTrue("scaled_float 转换结果应该是 BigDecimal 类型", result instanceof BigDecimal);
+        Assert.assertEquals("scaled_float 转换应该保持原始精度", originalValue, result);
+
+        // 验证 toString 表示
+        Assert.assertEquals("scaled_float 精度应该保持完整", "1234567890.12345678901234567890", result.toString());
+    }
+
+    @Test
+    public void testScaledFloatFromString() {
+        // 测试字符串转换为 scaled_float
+        String originalString = "9876543210.98765432109876543210";
+
+        // 转换为 scaled_float
+        Object result = ESSyncUtil.typeConvert(originalString, "scaled_float");
+
+        // 验证结果是 BigDecimal
+        Assert.assertTrue("scaled_float 字符串转换结果应该是 BigDecimal 类型", result instanceof BigDecimal);
+
+        // 验证值正确
+        BigDecimal expected = new BigDecimal(originalString);
+        Assert.assertEquals("scaled_float 字符串转换应该正确解析", expected, result);
+    }
+
+    @Test
+    public void testScaledFloatFromDouble() {
+        // 测试 double 转换为 scaled_float（注意：double 本身有精度限制）
+        Double originalDouble = 1234567890.1234567890;
+
+        // 转换为 scaled_float
+        Object result = ESSyncUtil.typeConvert(originalDouble, "scaled_float");
+
+        // 验证结果是 BigDecimal
+        Assert.assertTrue("scaled_float double 转换结果应该是 BigDecimal 类型", result instanceof BigDecimal);
+
+        // 验证值正确（由于 double 精度限制，这里验证转换过程正确）
+        BigDecimal expected = new BigDecimal(originalDouble.toString());
+        Assert.assertEquals("scaled_float double 转换应该正确处理", expected, result);
+    }
+
+    @Test
+    public void testOtherFloatTypes() {
+        // 测试其他浮点类型仍然正常工作
+        BigDecimal value = new BigDecimal("123.45");
+
+        // 测试 float 类型
+        Object floatResult = ESSyncUtil.typeConvert(value, "float");
+        Assert.assertTrue("float 转换结果应该是 Float 类型", floatResult instanceof Float);
+
+        // 测试 half_float 类型
+        Object halfFloatResult = ESSyncUtil.typeConvert(value, "half_float");
+        Assert.assertTrue("half_float 转换结果应该是 Float 类型", halfFloatResult instanceof Float);
+    }
+}

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.9-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>
@@ -60,7 +60,9 @@
                 </executions>
             </plugin>
             <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-antrun-plugin</artifactId>
+                <version>1.8</version>
                 <executions>
                     <execution>
                         <phase>package</phase>

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.9-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>
@@ -57,7 +57,9 @@
                 </executions>
             </plugin>
             <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-antrun-plugin</artifactId>
+                <version>1.8</version>
                 <executions>
                     <execution>
                         <phase>package</phase>

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.9-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>
@@ -17,7 +17,7 @@
             <dependency>
                 <groupId>org.springframework.boot</groupId>
                 <artifactId>spring-boot-dependencies</artifactId>
-                <version>2.5.4</version>
+                <version>2.5.15</version>
                 <type>pom</type>
                 <scope>import</scope>
             </dependency>
@@ -68,7 +68,6 @@
             <groupId>com.microsoft.sqlserver</groupId>
             <artifactId>mssql-jdbc</artifactId>
         </dependency>
-
         <dependency>
             <groupId>org.apache.hbase</groupId>
             <artifactId>hbase-shaded-client</artifactId>
@@ -79,6 +78,18 @@
             <artifactId>clickhouse-jdbc</artifactId>
         </dependency>
 
+        <!-- prometheus hotspot exporter -->
+        <dependency>
+            <groupId>io.prometheus</groupId>
+            <artifactId>simpleclient_hotspot</artifactId>
+            <version>0.4.0</version>
+        </dependency>
+        <dependency>
+            <groupId>io.prometheus</groupId>
+            <artifactId>simpleclient_httpserver</artifactId>
+            <version>0.4.0</version>
+        </dependency>
+
         <!-- outer adapter jar with dependencies-->
         <dependency>
             <groupId>com.alibaba.otter</groupId>

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

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

client-adapter/launcher/src/main/bin/startup.bat

@@ -11,7 +11,7 @@ set CLASSPATH=%conf_dir%
 set CLASSPATH=%conf_dir%\..\lib\*;%CLASSPATH%
 
 set JAVA_MEM_OPTS= -Xms128m -Xmx512m -XX:PermSize=128m
-set JAVA_OPTS_EXT= -Djava.awt.headless=true -Djava.net.preferIPv4Stack=true -Dapplication.codeset=UTF-8 -Dfile.encoding=UTF-8
+set JAVA_OPTS_EXT= -Djava.awt.headless=true -Djava.net.preferIPv4Stack=false -Dapplication.codeset=UTF-8 -Dfile.encoding=UTF-8
 set ADAPTER_OPTS= -DappName=canal-adapter
 
 set JAVA_OPTS= %JAVA_MEM_OPTS% %JAVA_OPTS_EXT% %ADAPTER_OPTS%

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

@@ -9,7 +9,10 @@ case "`uname`" in
 		bin_abs_path=`cd $(dirname $0); pwd`
 		;;
 esac
+
 base=${bin_abs_path}/..
+jaas_conf=$base/conf/jaas.conf
+
 export LANG=en_US.UTF-8
 export BASE=$base
 
@@ -56,31 +59,44 @@ in
   exit;;
 esac
 
-JavaVersion=`$JAVA -version 2>&1 |awk 'NR==1{ gsub(/"/,""); print $3 }' | awk  -F '.' '{print $1}'`
+JAVA_VERSION=`$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:+AggressiveOpts -XX:-UseBiasedLocking -XX:-OmitStackTraceInFastThrow -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=$base/logs"
+JAVA_OPTS="$JAVA_OPTS -Xss1m -XX:+AggressiveOpts -XX:-UseBiasedLocking -XX:-OmitStackTraceInFastThrow"
-if [ $JavaVersion -ge 11 ] ; then
+JAVA_OPTS="$JAVA_OPTS -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=$base/logs"
+
+if [ $JAVA_VERSION -ge 11 ] ; then
   #JAVA_OPTS="$JAVA_OPTS -Xlog:gc*:$base_log/gc.log:time "
-  JAVA_OPTS="$JAVA_OPTS"
+  JAVA_OPTS="$JAVA_OPTS -Xlog:gc*:file=$base/logs/adapter/gc.log::filecount=5,filesize=32M"
 else
-  #JAVA_OPTS="$JAVA_OPTS -Xloggc:$base/logs/canal/gc.log -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+PrintGCApplicationStoppedTime"
+  JAVA_OPTS="$JAVA_OPTS -Xloggc:$base/logs/canal/gc.log -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=5 -XX:GCLogFileSize=32M"
-  JAVA_OPTS="$JAVA_OPTS -XX:+UseFastAccessorMethods -XX:+PrintAdaptiveSizePolicy -XX:+PrintTenuringDistribution"
+  JAVA_OPTS="$JAVA_OPTS -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:-PrintHeapAtGC -XX:+PrintGCApplicationStoppedTime"
+  JAVA_OPTS="$JAVA_OPTS -XX:+UseFastAccessorMethods -XX:-PrintReferenceGC -XX:+PrintAdaptiveSizePolicy -XX:+PrintTenuringDistribution"
+  JAVA_OPTS="$JAVA_OPTS -XX:+PrintSafepointStatistics -XX:PrintSafepointStatisticsCount=1 -XX:+SafepointTimeout -XX:SafepointTimeoutDelay=2000"
+  JAVA_OPTS="$JAVA_OPTS -XX:+UseCountedLoopSafepoints -XX:+DisableExplicitGC -XX:+ExplicitGCInvokesConcurrent -XX:+ExplicitGCInvokesConcurrentAndUnloadsClasses"
 fi
 
-if [ -n "$str" ]; then
+OS_ARCH=`file -L $JAVA | grep 64-bit`
-  if [ $JavaVersion -ge 11 ] ; then
+if [ -n "$OS_ARCH" ]; then
+  if [ $JAVA_VERSION -ge 11 ] ; then
     # For G1
     JAVA_OPTS="-server -Xms2g -Xmx3g -XX:+UseG1GC -XX:MaxGCPauseMillis=250 -XX:+UseGCOverheadLimit -XX:+ExplicitGCInvokesConcurrent $JAVA_OPTS"
   else
 	  JAVA_OPTS="-server -Xms2g -Xmx3g -Xmn1g -XX:SurvivorRatio=2 -XX:PermSize=96m -XX:MaxPermSize=256m -XX:MaxTenuringThreshold=15 -XX:+DisableExplicitGC $JAVA_OPTS"
+	  ## JDK 1.8 2c4g [-XX:ConcGCThreads=1 -XX:ParallelGCThreads=2], 4c8g [-XX:ConcGCThreads=2 -XX:ParallelGCThreads=4]
+    #JAVA_OPTS="$JAVA_OPTS -XX:InitialRAMPercentage=45.0 -XX:MinRAMPercentage=45.0 -XX:MaxRAMPercentage=45.0"
+    #JAVA_OPTS="$JAVA_OPTS -XX:-UseParallelGC -XX:+UseConcMarkSweepGC -XX:ConcGCThreads=2 -XX:ParallelGCThreads=4"
+    #JAVA_OPTS="$JAVA_OPTS -XX:+UseCMSInitiatingOccupancyOnly -XX:CMSInitiatingOccupancyFraction=70"
+    #JAVA_OPTS="-server -XX:NewRatio=1 -XX:SurvivorRatio=2 -XX:MetaspaceSize=96m -XX:MaxMetaspaceSize=128m "
 	fi
 else
 	JAVA_OPTS="-server -Xms1024m -Xmx1024m -XX:NewSize=256m -XX:MaxNewSize=256m -XX:MaxPermSize=128m $JAVA_OPTS"
 fi
 
-JAVA_OPTS=" $JAVA_OPTS -Djava.awt.headless=true -Djava.net.preferIPv4Stack=true -Dfile.encoding=UTF-8"
+JAVA_OPTS=" $JAVA_OPTS -Djava.awt.headless=true -Djava.net.preferIPv4Stack=false -Dfile.encoding=UTF-8"
 ADAPTER_OPTS="-DappName=canal-adapter"
+if [ -f "$jaas_conf" ]; then
+  ADAPTER_OPTS="$ADAPTER_OPTS -Djava.security.auth.login.config=$jaas_conf"
+fi
 
 for i in $base/lib/*;
     do CLASSPATH=$i:"$CLASSPATH";

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

@@ -13,6 +13,7 @@ import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
  */
 @SpringBootApplication(exclude= { DataSourceAutoConfiguration.class})
 public class CanalAdapterApplication {
+
     public static void main(String[] args) {
         // 支持rocketmq client 配置日志路径
         System.setProperty("rocketmq.client.logUseSlf4j","true");

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

@@ -8,6 +8,8 @@ import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
+import com.alibaba.otter.canal.protocol.exception.CanalClientException;
+import org.apache.commons.lang.exception.ExceptionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -32,27 +34,27 @@ import com.alibaba.otter.canal.connector.core.spi.ProxyCanalMsgConsumer;
  */
 public class AdapterProcessor {
 
-    private static final Logger logger = LoggerFactory.getLogger(AdapterProcessor.class);
+    private static final Logger             logger                    = LoggerFactory.getLogger(AdapterProcessor.class);
 
-    private static final String CONNECTOR_SPI_DIR = "/plugin";
+    private static final String             CONNECTOR_SPI_DIR         = "/plugin";
-    private static final String CONNECTOR_STANDBY_SPI_DIR = "/canal-adapter/plugin";
+    private static final String             CONNECTOR_STANDBY_SPI_DIR = "/canal-adapter/plugin";
 
-    private CanalMsgConsumer canalMsgConsumer;
+    private CanalMsgConsumer                canalMsgConsumer;
 
-    private String canalDestination;                                                           // canal实例
+    private String                          canalDestination;                                                           // canal实例
-    private String groupId = null;                                           // groupId
+    private String                          groupId                   = null;                                           // groupId
-    private List<List<OuterAdapter>> canalOuterAdapters;                                                         // 外部适配器
+    private List<List<OuterAdapter>>        canalOuterAdapters;                                                         // 外部适配器
-    private CanalClientConfig canalClientConfig;                                                          // 配置
+    private CanalClientConfig               canalClientConfig;                                                          // 配置
-    private ExecutorService groupInnerExecutorService;                                                  // 组内工作线程池
+    private ExecutorService                 groupInnerExecutorService;                                                  // 组内工作线程池
-    private volatile boolean running = false;                                          // 是否运行中
+    private volatile boolean                running                   = false;                                          // 是否运行中
-    private Thread thread = null;
+    private Thread                          thread                    = null;
-    private Thread.UncaughtExceptionHandler handler = (t, e) -> logger
+    private Thread.UncaughtExceptionHandler handler                   = (t, e) -> logger
         .error("parse events has an error", e);
 
-    private SyncSwitch syncSwitch;
+    private SyncSwitch                      syncSwitch;
 
     public AdapterProcessor(CanalClientConfig canalClientConfig, String destination, String groupId,
-        List<List<OuterAdapter>> canalOuterAdapters) {
+                            List<List<OuterAdapter>> canalOuterAdapters){
         this.canalClientConfig = canalClientConfig;
         this.canalDestination = destination;
         this.groupId = groupId;
@@ -63,9 +65,12 @@ public class AdapterProcessor {
 
         // load connector consumer
         ExtensionLoader<CanalMsgConsumer> loader = new ExtensionLoader<>(CanalMsgConsumer.class);
-        canalMsgConsumer = new ProxyCanalMsgConsumer(loader
+        // see https://github.com/alibaba/canal/pull/5175
-            .getExtension(canalClientConfig.getMode().toLowerCase(), destination, CONNECTOR_SPI_DIR,
+        String key = destination + "_" + groupId;
-                CONNECTOR_STANDBY_SPI_DIR));
+        canalMsgConsumer = new ProxyCanalMsgConsumer(loader.getExtension(canalClientConfig.getMode().toLowerCase(),
+            key,
+            CONNECTOR_SPI_DIR,
+            CONNECTOR_STANDBY_SPI_DIR));
 
         Properties properties = canalClientConfig.getConsumerProperties();
         properties.put(CanalConstants.CANAL_MQ_FLAT_MESSAGE, canalClientConfig.getFlatMessage());
@@ -170,7 +175,7 @@ public class AdapterProcessor {
         }
 
         int retry = canalClientConfig.getRetries() == null
-            || canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries();
+                    || canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries();
         if (retry == -1) {
             // 重试次数-1代表异常时一直阻塞重试
             retry = Integer.MAX_VALUE;
@@ -183,7 +188,7 @@ public class AdapterProcessor {
                 logger.info("=============> Start to connect destination: {} <=============", this.canalDestination);
                 canalMsgConsumer.connect();
                 logger.info("=============> Subscribe destination: {} succeed <=============", this.canalDestination);
-                while (running) {
+                out: while (running) {
                     try {
                         syncSwitch.get(canalDestination, 1L, TimeUnit.MINUTES);
                     } catch (TimeoutException e) {
@@ -213,6 +218,17 @@ public class AdapterProcessor {
                             }
                             break;
                         } catch (Exception e) {
+                            Throwable th = e.getCause();
+                            // Handle source error when getting message
+                            if (th instanceof CanalClientException) {
+                                String message = ExceptionUtils.getRootCauseMessage(th);
+                                if (message.contains("end of stream when reading header")
+                                    || message.contains("Connection reset by peer") || message.contains("Broken pipe")) {
+                                    logger.error("Sync failed, reconnect to canal instance. Error: {}", message);
+                                    break out;
+                                }
+                            }
+                            // Handle sink error
                             if (i != retry - 1) {
                                 canalMsgConsumer.rollback(); // 处理失败, 回滚数据
                                 logger.error(e.getMessage() + " Error sync and rollback, execute times: " + (i + 1));

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/ApplicationConfigMonitor.java

@@ -18,7 +18,9 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.cloud.context.refresh.ContextRefresher;
 import org.springframework.stereotype.Component;
+import org.yaml.snakeyaml.LoaderOptions;
 import org.yaml.snakeyaml.Yaml;
+import org.yaml.snakeyaml.constructor.SafeConstructor;
 
 import com.alibaba.otter.canal.adapter.launcher.loader.CanalAdapterService;
 import com.alibaba.otter.canal.client.adapter.support.Util;
@@ -70,7 +72,7 @@ public class ApplicationConfigMonitor {
             super.onFileChange(file);
             try {
                 // 检查yml格式
-                new Yaml().loadAs(new InputStreamReader(new FileInputStream(file), StandardCharsets.UTF_8), Map.class);
+                new Yaml(new SafeConstructor(new LoaderOptions())).loadAs(new InputStreamReader(new FileInputStream(file), StandardCharsets.UTF_8), Map.class);
 
                 canalAdapterService.destroy();
 

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/prometheus/CanalAdapterExports.java

@@ -0,0 +1,33 @@
+package com.alibaba.otter.canal.adapter.launcher.prometheus;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author sunxien
+ * @date 2025/5/19
+ * @since 1.0.0-SNAPSHOT
+ */
+public class CanalAdapterExports {
+
+    private static final Logger logger = LoggerFactory.getLogger(CanalAdapterExports.class);
+
+    private CanalAdapterExports() {
+    }
+
+    private static class SingletonHolder {
+        private static final CanalAdapterExports SINGLETON = new CanalAdapterExports();
+    }
+
+    public static CanalAdapterExports instance() {
+        return SingletonHolder.SINGLETON;
+    }
+
+    public void initialize() {
+        // TODO
+    }
+
+    public void terminate() {
+        // TODO
+    }
+}

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/prometheus/PrometheusService.java

@@ -0,0 +1,100 @@
+package com.alibaba.otter.canal.adapter.launcher.prometheus;
+
+import java.io.IOException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.stereotype.Component;
+
+import io.prometheus.client.exporter.HTTPServer;
+import io.prometheus.client.hotspot.DefaultExports;
+
+/**
+ * @author sunxien
+ * @date 2025/5/19
+ * @since 1.0.0-SNAPSHOT
+ */
+@Component
+public class PrometheusService implements InitializingBean, DisposableBean {
+
+    private static final Logger logger = LoggerFactory.getLogger(PrometheusService.class);
+
+    private final CanalAdapterExports adapterExports;
+    private volatile boolean running = false;
+
+    /**
+     * <pre>
+     * Canal Server admin port: 11110
+     * Canal Server tcp port: 11111
+     * Canal Server metric port: 11112
+     * Canal Adapter metric port: 11113
+     * </pre>
+     */
+    private int port = 11113;
+    private HTTPServer httpServer;
+
+    private PrometheusService() {
+        this.adapterExports = CanalAdapterExports.instance();
+    }
+
+    private static class SingletonHolder {
+        private static final PrometheusService SINGLETON = new PrometheusService();
+    }
+
+    public static PrometheusService getInstance() {
+        return SingletonHolder.SINGLETON;
+    }
+
+    @Override
+    public void afterPropertiesSet() throws Exception {
+        PrometheusService.getInstance().initialize();
+    }
+
+    @Override
+    public void destroy() throws Exception {
+        PrometheusService.getInstance().terminate();
+    }
+
+    public void initialize() {
+        try {
+            logger.info("Starting prometheus HTTPServer on port {}....", port);
+            // TODO 2.Https?
+            httpServer = new HTTPServer(port);
+            logger.info("Start prometheus HTTPServer on port {} success", port);
+        } catch (IOException e) {
+            logger.error("Unable to start prometheus HTTPServer on port {}.", port, e);
+            return;
+        }
+        try {
+            // JVM exports
+            DefaultExports.initialize();
+            // adapterExports.initialize();
+            this.running = true;
+        } catch (Throwable t) {
+            logger.error("Unable to initialize adapter exports. (Register the default Hotspot collectors)", t);
+        }
+    }
+
+    public void terminate() {
+        try {
+            this.running = false;
+            adapterExports.terminate();
+            if (httpServer != null) {
+                httpServer.stop();
+                logger.info("Stop prometheus HTTPServer on port {} success", port);
+            }
+        } catch (Throwable t) {
+            logger.error("Something happened while terminating prometheus HTTPServer.", t);
+        }
+    }
+
+    public boolean isRunning() {
+        return running;
+    }
+
+    public void setServerPort(int port) {
+        this.port = port;
+    }
+}

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/CommonRest.java

@@ -217,7 +217,7 @@ public class CommonRest {
             resStatus = "off";
         }
         Map<String, String> res = new LinkedHashMap<>();
-        res.put("stauts", resStatus);
+        res.put("status", resStatus);
         return res;
     }
 }

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.9-SNAPSHOT</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.9-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>
@@ -100,7 +100,9 @@
                 </executions>
             </plugin>
             <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-antrun-plugin</artifactId>
+                <version>1.8</version>
                 <executions>
                     <execution>
                         <phase>package</phase>

client-adapter/phoenix/src/main/java/com/alibaba/otter/canal/client/adapter/phoenix/service/PhoenixEtlService.java

@@ -172,11 +172,12 @@ public class PhoenixEtlService {
                     "SELECT * FROM " + dbMapping.getDatabase() + "." + dbMapping.getTable());
 
             // 拼接条件
-            appendCondition(params, dbMapping, srcDS, sql);
+            List<Object> values = new ArrayList<>();
+            appendCondition(params, dbMapping, srcDS, sql, values);
 
             // 获取总数
             String countSql = "SELECT COUNT(1) FROM ( " + sql + ") _CNT ";
-            long cnt = (Long) Util.sqlRS(srcDS, countSql, rs -> {
+            long cnt = (Long) Util.sqlRS(srcDS, countSql, values, rs -> {
                 Long count = null;
                 try {
                     if (rs.next()) {
@@ -206,14 +207,14 @@ public class PhoenixEtlService {
                         sqlFinal = sql + " LIMIT " + offset + "," + cnt;
                     }
                     executor
-                            .execute(() -> executeSqlImport(srcDS, targetDSConnection, sqlFinal, dbMapping, successCount, errMsg, debug));
+                            .execute(() -> executeSqlImport(srcDS, targetDSConnection, sqlFinal, values, dbMapping, successCount, errMsg, debug));
                 }
 
                 executor.shutdown();
                 //noinspection StatementWithEmptyBody
                 while (!executor.awaitTermination(3, TimeUnit.SECONDS)) ;
             } else {
-                executeSqlImport(srcDS, targetDSConnection, sql.toString(), dbMapping, successCount, errMsg, debug);
+                executeSqlImport(srcDS, targetDSConnection, sql.toString(), values, dbMapping, successCount, errMsg, debug);
             }
 
             logger.info(
@@ -235,7 +236,7 @@ public class PhoenixEtlService {
     }
 
     private static void appendCondition(List<String> params, DbMapping dbMapping, DataSource ds,
-                                        StringBuilder sql) {
+                                        StringBuilder sql, List<Object> values) {
         if (params != null && params.size() == 1 && dbMapping.getEtlCondition() == null) {
             AtomicBoolean stExists = new AtomicBoolean(false);
             // 验证是否有SYS_TIME字段
@@ -256,13 +257,27 @@ public class PhoenixEtlService {
                 return null;
             });
             if (stExists.get()) {
-                sql.append(" WHERE SYS_TIME >= '").append(params.get(0)).append("' ");
+                sql.append(" WHERE SYS_TIME >= ? ");
+                values.add(params.get(0));
             }
         } else if (dbMapping.getEtlCondition() != null && params != null && params.size() > 0) {
             String etlCondition = dbMapping.getEtlCondition();
             int size = params.size();
             for (int i = 0; i < size; i++) {
-                etlCondition = etlCondition.replace("{" + i + "}", params.get(i));
+                String placeholder = "{" + i + "}";
+                int idx = etlCondition.indexOf(placeholder);
+                if (idx >= 0) {
+                    int start = idx;
+                    int end = idx + placeholder.length();
+                    // 兼容配置中 '{i}' 这种被单引号包裹的写法，避免生成 '?' 语法错误
+                    if (start > 0 && etlCondition.charAt(start - 1) == '\''
+                        && end < etlCondition.length() && etlCondition.charAt(end) == '\'') {
+                        start--;
+                        end++;
+                    }
+                    etlCondition = etlCondition.substring(0, start) + "?" + etlCondition.substring(end);
+                    values.add(params.get(i));
+                }
             }
 
             sql.append(" ").append(etlCondition);
@@ -272,8 +287,8 @@ public class PhoenixEtlService {
     /**
      * 执行导入
      */
-    private static boolean executeSqlImport(DataSource srcDS, Connection targetDSConnection, String sql, DbMapping dbMapping,
+    private static boolean executeSqlImport(DataSource srcDS, Connection targetDSConnection, String sql, List<Object> values,
-                                            AtomicLong successCount, List<String> errMsg, boolean debug) {
+                                            DbMapping dbMapping, AtomicLong successCount, List<String> errMsg, boolean debug) {
         try {
             Map<String, String> columnsMap = new LinkedHashMap<>();
             Map<String, Integer> columnType = new LinkedHashMap<>();
@@ -300,7 +315,7 @@ public class PhoenixEtlService {
                     return false;
                 }
             });
-            Util.sqlRS(srcDS, sql, rs -> {
+            Util.sqlRS(srcDS, sql, values, rs -> {
                 int idx = 1;
 
                 try {
@@ -337,13 +352,13 @@ public class PhoenixEtlService {
                             pstmt.clearParameters();
 
                             // 删除数据
-                            Map<String, Object> values = new LinkedHashMap<>();
+                            Map<String, Object> deleteValues = new LinkedHashMap<>();
                             StringBuilder deleteSql = new StringBuilder(
                                     "DELETE FROM " + SyncUtil.getDbTableName(dbMapping) + " WHERE ");
-                            appendCondition(dbMapping, deleteSql, values, rs);
+                            appendCondition(dbMapping, deleteSql, deleteValues, rs);
                             try (PreparedStatement pstmt2 = connTarget.prepareStatement(deleteSql.toString())) {
                                 int k = 1;
-                                for (Object val : values.values()) {
+                                for (Object val : deleteValues.values()) {
                                     pstmt2.setObject(k++, val);
                                 }
                                 pstmt2.execute();
@@ -352,13 +367,13 @@ public class PhoenixEtlService {
                             Map<String, Object> insertValues = new HashMap<>();
                             int i = 1;
                             for (Map.Entry<String, String> entry : columnsMap.entrySet()) {
-                                String targetClolumnName = entry.getKey();
+                                String targetColumnName = entry.getKey();
                                 String srcColumnName = entry.getValue();
                                 if (srcColumnName == null) {
-                                    srcColumnName = targetClolumnName;
+                                    srcColumnName = targetColumnName;
                                 }
 
-                                Integer type = columnType.get(targetClolumnName.toLowerCase());
+                                Integer type = columnType.get(targetColumnName.toLowerCase());
 
                                 try {
                                     Object value = rs.getObject(srcColumnName);

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.9-SNAPSHOT</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.9-SNAPSHOT</version>
     <packaging>pom</packaging>
     <name>canal client adapter module for otter ${project.version}</name>
 
@@ -117,7 +117,7 @@
             <dependency>
                 <groupId>org.springframework.boot</groupId>
                 <artifactId>spring-boot</artifactId>
-                <version>2.5.4</version>
+                <version>2.5.15</version>
             </dependency>
             <dependency>
                 <groupId>com.h2database</groupId>
@@ -136,7 +136,7 @@
                 <artifactId>kafka-clients</artifactId>
                 <version>${kafka_version}</version>
             </dependency>
-            <!-- rabbotmq -->
+            <!-- rabbitmq -->
             <dependency>
                 <groupId>com.rabbitmq</groupId>
                 <artifactId>amqp-client</artifactId>
@@ -238,7 +238,7 @@
             <dependency>
                 <groupId>org.yaml</groupId>
                 <artifactId>snakeyaml</artifactId>
-                <version>1.29</version>
+                <version>2.0</version>
             </dependency>
             <!-- test -->
             <dependency>

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.9-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>
@@ -68,7 +68,9 @@
                 </executions>
             </plugin>
             <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-antrun-plugin</artifactId>
+                <version>1.8</version>
                 <executions>
                     <execution>
                         <phase>package</phase>

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

@@ -39,7 +39,7 @@ public class RdbEtlService extends AbstractEtlService {
         DbMapping dbMapping = config.getDbMapping();
         // 获取源数据源，根据数据库类型拼装数据库名和表名
         DruidDataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
-        String sql = "SELECT * FROM " + SyncUtil.getDbTableName(dbMapping, dataSource.getDbType());
+        String sql = "SELECT * FROM " + SyncUtil.getSourceDbTableName(dbMapping, dataSource.getDbType());
         return importData(sql, params);
     }
 
@@ -122,13 +122,13 @@ public class RdbEtlService extends AbstractEtlService {
 
                             int i = 1;
                             for (Map.Entry<String, String> entry : columnsMap.entrySet()) {
-                                String targetClolumnName = entry.getKey();
+                                String targetColumnName = entry.getKey();
                                 String srcColumnName = entry.getValue();
                                 if (srcColumnName == null) {
-                                    srcColumnName = targetClolumnName;
+                                    srcColumnName = targetColumnName;
                                 }
 
-                                Integer type = columnType.get(targetClolumnName.toLowerCase());
+                                Integer type = columnType.get(targetColumnName.toLowerCase());
                                 Object value = rs.getObject(srcColumnName);
                                 if (value != null) {
                                     SyncUtil.setPStmt(type, pstmt, value, i);

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

@@ -294,7 +294,7 @@ public class RdbSyncService {
             batchExecutor.execute(insertSql.toString(), values);
         } catch (SQLException e) {
             if (skipDupException
-                && (e.getMessage().contains("Duplicate entry") || e.getMessage().startsWith("ORA-00001:"))) {
+                && (e.getMessage().contains("Duplicate entry") || e.getMessage().contains("duplicate key") || e.getMessage().startsWith("ORA-00001:"))) {
                 // ignore
                 // TODO 增加更多关系数据库的主键冲突的错误码
             } else {

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

@@ -284,6 +284,17 @@ public class SyncUtil {
         return result;
     }
 
+    public static String getSourceDbTableName(MappingConfig.DbMapping dbMapping, String dbType) {
+        String result = "";
+        String backtick = getBacktickByDbType(dbType);
+        if (StringUtils.isNotEmpty(dbMapping.getDatabase())) {
+            result += (backtick + dbMapping.getDatabase() + backtick + ".");
+        }
+
+        result += (backtick + dbMapping.getTable() + backtick);
+        return result;
+    }
+
     /**
      * 根据DbType返回反引号或空字符串
      *
@@ -302,6 +313,9 @@ public class SyncUtil {
             case mariadb:
             case oceanbase:
                 return "`";
+        //  当dbType 为 postgresql时，返回双引号(避免表名为postgresql的关键字时，sql查询报错)
+            case postgresql:
+                return "\"";
             default:
                 return "";
         }

client-adapter/tablestore/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.9-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>
@@ -66,7 +66,9 @@
                 </executions>
             </plugin>
             <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-antrun-plugin</artifactId>
+                <version>1.8</version>
                 <executions>
                     <execution>
                         <phase>package</phase>

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.9-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>

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

@@ -1,15 +1,14 @@
 package com.alibaba.otter.canal.client.impl;
 
-import java.net.SocketAddress;
-import java.util.concurrent.TimeUnit;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import com.alibaba.otter.canal.client.CanalConnector;
 import com.alibaba.otter.canal.client.CanalNodeAccessStrategy;
 import com.alibaba.otter.canal.protocol.Message;
 import com.alibaba.otter.canal.protocol.exception.CanalClientException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.SocketAddress;
+import java.util.concurrent.TimeUnit;
 
 /**
  * 集群版本connector实现，自带了failover功能<br/>
@@ -65,7 +64,7 @@ public class ClusterCanalConnector implements CanalConnector {
                     currentConnector.connect();
                     break;
                 } catch (Exception e) {
-                    logger.warn("failed to connect to:{} after retry {} times", accessStrategy.currentNode(), times);
+                    logger.warn("failed to connect to canal server after retry {} times", times);
                     currentConnector.disconnect();
                     currentConnector = null;
                     // retry for #retryTimes for each node when trying to

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.9-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<artifactId>canal.common</artifactId>
@@ -90,7 +90,6 @@
 		<dependency>
 			<groupId>com.alibaba</groupId>
 			<artifactId>druid</artifactId>
-			<optional>true</optional>
 		</dependency>
 		<dependency>
 			<groupId>org.springframework</groupId>

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

@@ -1,5 +1,6 @@
 package com.alibaba.otter.canal.common.utils;
 
+import java.io.File;
 import java.io.IOException;
 import java.io.RandomAccessFile;
 import java.nio.charset.StandardCharsets;
@@ -88,8 +89,43 @@ public class FileUtils {
         return res.toString();
     }
 
-    public static void main(String[] args) {
+    /**
-        String res = readFileFromOffset("test2.txt", 2, "UTF-8");
+     * 校验自定义的文件名，是否在允许的基目录范围内，如何合法就返回全路径，否则就直接报错
-        System.out.println(res);
+     *
+     * @param baseDir
+     * @param destination
+     * @return
+     */
+    public static String validateFileName(String baseDir, String destination) {
+        try {
+            // 验证 destination 是否在允许的基目录范围内
+            String basePath = new File(baseDir).getCanonicalPath();
+            String fullPath = new File(basePath, destination).getCanonicalPath();
+
+            // 检查 fullPath 是否以 basePath 开头
+            if (!fullPath.startsWith(basePath + File.separator)) {
+                throw new IllegalArgumentException("Invalid destination path");
+            }
+
+            return fullPath;
+        } catch (IOException e) {
+            throw new RuntimeException("Failed to read file", e);
+        }
+    }
+
+    public static void main(String[] args) throws IOException {
+        String fullPath = validateFileName("/tmp/", "1.txt");
+        System.out.println(fullPath);
+        System.out.println(org.apache.commons.io.FileUtils.readLines(new File(fullPath)));
+
+        fullPath = validateFileName("/tmp/", "test");
+        fullPath = validateFileName(fullPath,"1.txt");
+        System.out.println(fullPath);
+        System.out.println(org.apache.commons.io.FileUtils.readLines(new File(fullPath)));
+
+
+        fullPath = validateFileName("/tmp/", "../etc/hosts");
+        System.out.println(fullPath);
+        System.out.println(org.apache.commons.io.FileUtils.readLines(new File(fullPath)));
     }
 }

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.9-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.alibaba.otter</groupId>

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

@@ -1,7 +1,6 @@
 package com.alibaba.otter.canal.connector.core.filter;
 
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
@@ -19,7 +18,6 @@ import com.googlecode.aviator.Expression;
  */
 public class AviaterRegexFilter {
 
-    private static final String             SPLIT             = ",";
     private static final String             PATTERN_SPLIT     = "|";
     private static final String             FILTER_EXPRESSION = "regex(pattern,target)";
     private static final RegexFunction      regexFunction     = new RegexFunction();
@@ -43,8 +41,9 @@ public class AviaterRegexFilter {
         if (StringUtils.isEmpty(pattern)) {
             list = new ArrayList<>();
         } else {
-            String[] ss = StringUtils.split(pattern, SPLIT);
+            // fix issue #5442: only split on commas outside regex grouping constructs,
-            list = Arrays.asList(ss);
+            // so quantifiers like \d{1,2} are kept intact.
+            list = splitPattern(pattern);
         }
 
         // 对pattern按照从长到短的排序
@@ -56,6 +55,60 @@ public class AviaterRegexFilter {
         this.pattern = StringUtils.join(list, PATTERN_SPLIT);
     }
 
+    /**
+     * 按顶层逗号拆分多个正则表达式，跳过位于 {}, [], () 内的逗号，避免破坏类似 \d{1,2} 的量词写法。
+     */
+    private static List<String> splitPattern(String pattern) {
+        List<String> result = new ArrayList<>();
+        int braceDepth = 0;
+        int bracketDepth = 0;
+        int parenDepth = 0;
+        int start = 0;
+        int len = pattern.length();
+        for (int i = 0; i < len; i++) {
+            char c = pattern.charAt(i);
+            // 反斜杠转义：跳过下一个字符，避免把 \{ \[ \( 等误判为分组开始
+            if (c == '\\' && i + 1 < len) {
+                i++;
+                continue;
+            }
+            switch (c) {
+                case '{':
+                    braceDepth++;
+                    break;
+                case '}':
+                    if (braceDepth > 0) braceDepth--;
+                    break;
+                case '[':
+                    bracketDepth++;
+                    break;
+                case ']':
+                    if (bracketDepth > 0) bracketDepth--;
+                    break;
+                case '(':
+                    parenDepth++;
+                    break;
+                case ')':
+                    if (parenDepth > 0) parenDepth--;
+                    break;
+                case ',':
+                    if (braceDepth == 0 && bracketDepth == 0 && parenDepth == 0) {
+                        if (i > start) {
+                            result.add(pattern.substring(start, i));
+                        }
+                        start = i + 1;
+                    }
+                    break;
+                default:
+                    break;
+            }
+        }
+        if (start < len) {
+            result.add(pattern.substring(start));
+        }
+        return result;
+    }
+
     public boolean filter(String filtered) {
         if (StringUtils.isEmpty(pattern)) {
             return defaultEmptyValue;

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.9-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

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.9-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 
     <groupId>com.alibaba.otter</groupId>
     <artifactId>canal.connector</artifactId>
-    <version>1.1.8-SNAPSHOT</version>
+    <version>1.1.9-SNAPSHOT</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.9-SNAPSHOT</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

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

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

@@ -18,7 +18,7 @@ public class PulsarMQProducerConfig extends MQProperties {
      * httpUrl: http://localhost:8080
      * </p>
      */
-    private String serverUrl;
+    private String  serverUrl;
     /**
      * pulsar topic前缀
      * <p>
@@ -27,20 +27,30 @@ public class PulsarMQProducerConfig extends MQProperties {
      * 在发送消息时会自动拼接上
      * </p>
      */
-    private String topicTenantPrefix;
+    private String  topicTenantPrefix;
     /**
      * 生产者角色权限，请确保该角色有canal使用的所有topic生产者权限（最低要求）
      */
-    private String roleToken;
+    private String  roleToken;
     /**
      * admin服务器地址
      */
-    private String adminServerUrl;
+    private String  adminServerUrl;
 
     /**
      * listener name
      */
-    private String listenerName;
+    private String  listenerName;
+
+    /**
+     * enableChunking
+     */
+    private boolean enableChunking;
+
+    /**
+     * compressionType
+     */
+    private String  compressionType;
 
     public String getServerUrl() {
         return serverUrl;
@@ -81,4 +91,20 @@ public class PulsarMQProducerConfig extends MQProperties {
     public void setListenerName(String listenerName) {
         this.listenerName = listenerName;
     }
+
+    public void setEnableChunking(boolean enableChunking) {
+        this.enableChunking = enableChunking;
+    }
+
+    public boolean getEnableChunking() {
+        return this.enableChunking;
+    }
+
+    public void setCompressionType(String compressionType) {
+        this.compressionType = compressionType;
+    }
+
+    public String getCompressionType() {
+        return this.compressionType;
+    }
 }

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

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

connector/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.9-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

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

@@ -210,13 +210,12 @@ public class CanalRabbitMQProducer extends AbstractMQProducer implements CanalMQ
         try {
             this.channel.close();
             this.connect.close();
-            super.stop();
         } catch (AlreadyClosedException ex) {
             logger.error("Connection is already closed", ex);
         } catch (IOException | TimeoutException ex) {
             throw new CanalException("Stop RabbitMQ producer error", ex);
+        } finally {
+            super.stop();
         }
-
-        super.stop();
     }
 }

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.9-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

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

@@ -1,5 +1,29 @@
 package com.alibaba.otter.canal.connector.rocketmq.producer;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.rocketmq.acl.common.AclClientRPCHook;
+import org.apache.rocketmq.acl.common.SessionCredentials;
+import org.apache.rocketmq.client.AccessChannel;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl;
+import org.apache.rocketmq.client.impl.producer.TopicPublishInfo;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.remoting.RPCHook;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import com.alibaba.fastjson2.JSON;
 import com.alibaba.fastjson2.JSONWriter;
 import com.alibaba.otter.canal.common.CanalException;
@@ -16,29 +40,6 @@ import com.alibaba.otter.canal.connector.core.util.CanalMessageSerializerUtil;
 import com.alibaba.otter.canal.connector.rocketmq.config.RocketMQConstants;
 import com.alibaba.otter.canal.connector.rocketmq.config.RocketMQProducerConfig;
 import com.alibaba.otter.canal.protocol.FlatMessage;
-import org.apache.commons.lang.StringUtils;
-import org.apache.rocketmq.acl.common.AclClientRPCHook;
-import org.apache.rocketmq.acl.common.SessionCredentials;
-import org.apache.rocketmq.client.AccessChannel;
-import org.apache.rocketmq.client.exception.MQClientException;
-import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl;
-import org.apache.rocketmq.client.impl.producer.TopicPublishInfo;
-import org.apache.rocketmq.client.producer.DefaultMQProducer;
-import org.apache.rocketmq.client.producer.SendResult;
-import org.apache.rocketmq.common.message.Message;
-import org.apache.rocketmq.common.message.MessageQueue;
-import org.apache.rocketmq.remoting.RPCHook;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
 
 /**
  * RocketMQ Producer SPI 实现
@@ -46,15 +47,18 @@ import java.util.stream.Collectors;
  * @author rewerma 2020-01-27
  * @version 1.0.0
  */
-@SPI("rocketmq") public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQProducer {
+@SPI("rocketmq")
+public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQProducer {
 
-    private static final Logger logger = LoggerFactory.getLogger(CanalRocketMQProducer.class);
+    private static final Logger  logger               = LoggerFactory.getLogger(CanalRocketMQProducer.class);
 
-    private              DefaultMQProducer  defaultMQProducer;
+    private DefaultMQProducer    defaultMQProducer;
-    private static final String             CLOUD_ACCESS_CHANNEL = "cloud";
+    private static final String  CLOUD_ACCESS_CHANNEL = "cloud";
-    protected            ThreadPoolExecutor sendPartitionExecutor;
+    private static final String  NAMESPACE_SEPARATOR  = "%";
+    protected ThreadPoolExecutor sendPartitionExecutor;
 
-    @Override public void init(Properties properties) {
+    @Override
+    public void init(Properties properties) {
         RocketMQProducerConfig rocketMQProperties = new RocketMQProducerConfig();
         this.mqProperties = rocketMQProperties;
         super.init(properties);
@@ -69,9 +73,9 @@ import java.util.stream.Collectors;
         }
 
         defaultMQProducer = new DefaultMQProducer(rocketMQProperties.getProducerGroup(),
-                rpcHook,
+            rpcHook,
-                rocketMQProperties.isEnableMessageTrace(),
+            rocketMQProperties.isEnableMessageTrace(),
-                rocketMQProperties.getCustomizedTraceTopic());
+            rocketMQProperties.getCustomizedTraceTopic());
         if (CLOUD_ACCESS_CHANNEL.equals(rocketMQProperties.getAccessChannel())) {
             defaultMQProducer.setAccessChannel(AccessChannel.CLOUD);
         }
@@ -90,12 +94,12 @@ import java.util.stream.Collectors;
 
         int parallelPartitionSendThreadSize = mqProperties.getParallelSendThreadSize();
         sendPartitionExecutor = new ThreadPoolExecutor(parallelPartitionSendThreadSize,
-                parallelPartitionSendThreadSize,
+            parallelPartitionSendThreadSize,
-                0,
+            0,
-                TimeUnit.SECONDS,
+            TimeUnit.SECONDS,
-                new ArrayBlockingQueue<>(parallelPartitionSendThreadSize * 2),
+            new ArrayBlockingQueue<>(parallelPartitionSendThreadSize * 2),
-                new NamedThreadFactory("MQ-Parallel-Sender-Partition"),
+            new NamedThreadFactory("MQ-Parallel-Sender-Partition"),
-                new ThreadPoolExecutor.CallerRunsPolicy());
+            new ThreadPoolExecutor.CallerRunsPolicy());
     }
 
     private void loadRocketMQProperties(Properties properties) {
@@ -111,12 +115,12 @@ import java.util.stream.Collectors;
             rocketMQProperties.setProducerGroup(producerGroup);
         }
         String enableMessageTrace = PropertiesUtils.getProperty(properties,
-                RocketMQConstants.ROCKETMQ_ENABLE_MESSAGE_TRACE);
+            RocketMQConstants.ROCKETMQ_ENABLE_MESSAGE_TRACE);
         if (!StringUtils.isEmpty(enableMessageTrace)) {
             rocketMQProperties.setEnableMessageTrace(Boolean.parseBoolean(enableMessageTrace));
         }
         String customizedTraceTopic = PropertiesUtils.getProperty(properties,
-                RocketMQConstants.ROCKETMQ_CUSTOMIZED_TRACE_TOPIC);
+            RocketMQConstants.ROCKETMQ_CUSTOMIZED_TRACE_TOPIC);
         if (!StringUtils.isEmpty(customizedTraceTopic)) {
             rocketMQProperties.setCustomizedTraceTopic(customizedTraceTopic);
         }
@@ -133,7 +137,7 @@ import java.util.stream.Collectors;
             rocketMQProperties.setRetryTimesWhenSendFailed(Integer.parseInt(retry));
         }
         String vipChannelEnabled = PropertiesUtils.getProperty(properties,
-                RocketMQConstants.ROCKETMQ_VIP_CHANNEL_ENABLED);
+            RocketMQConstants.ROCKETMQ_VIP_CHANNEL_ENABLED);
         if (!StringUtils.isEmpty(vipChannelEnabled)) {
             rocketMQProperties.setVipChannelEnabled(Boolean.parseBoolean(vipChannelEnabled));
         }
@@ -143,15 +147,14 @@ import java.util.stream.Collectors;
         }
     }
 
-    @Override public void send(MQDestination destination, com.alibaba.otter.canal.protocol.Message message,
+    @Override
-                               Callback callback) {
+    public void send(MQDestination destination, com.alibaba.otter.canal.protocol.Message message, Callback callback) {
         ExecutorTemplate template = new ExecutorTemplate(sendExecutor);
         try {
             if (!StringUtils.isEmpty(destination.getDynamicTopic())) {
                 // 动态topic
-                Map<String, com.alibaba.otter.canal.protocol.Message> messageMap = MQMessageUtils.messageTopics(message,
+                Map<String, com.alibaba.otter.canal.protocol.Message> messageMap = MQMessageUtils
-                        destination.getTopic(),
+                    .messageTopics(message, destination.getTopic(), destination.getDynamicTopic());
-                        destination.getDynamicTopic());
 
                 for (Map.Entry<String, com.alibaba.otter.canal.protocol.Message> entry : messageMap.entrySet()) {
                     String topicName = entry.getKey().replace('.', '_');
@@ -183,7 +186,7 @@ import java.util.stream.Collectors;
                      com.alibaba.otter.canal.protocol.Message message) {
         // 获取当前topic的分区数
         Integer partitionNum = MQMessageUtils.parseDynamicTopicPartition(topicName,
-                destination.getDynamicTopicPartitionNum());
+            destination.getDynamicTopicPartitionNum());
 
         // 获取topic的队列数为分区数
         if (partitionNum == null) {
@@ -199,10 +202,10 @@ import java.util.stream.Collectors;
                 MQMessageUtils.EntryRowData[] datas = MQMessageUtils.buildMessageData(message, buildExecutor);
                 // 串行分区
                 com.alibaba.otter.canal.protocol.Message[] messages = MQMessageUtils.messagePartition(datas,
-                        message.getId(),
+                    message.getId(),
-                        partitionNum,
+                    partitionNum,
-                        destination.getPartitionHash(),
+                    destination.getPartitionHash(),
-                        mqProperties.isDatabaseHash());
+                    mqProperties.isDatabaseHash());
                 int length = messages.length;
 
                 ExecutorTemplate template = new ExecutorTemplate(sendPartitionExecutor);
@@ -212,9 +215,9 @@ import java.util.stream.Collectors;
                         final int index = i;
                         template.submit(() -> {
                             Message data = new Message(topicName,
-                                    ((RocketMQProducerConfig) this.mqProperties).getTag(),
+                                ((RocketMQProducerConfig) this.mqProperties).getTag(),
-                                    CanalMessageSerializerUtil.serializer(dataPartition,
+                                CanalMessageSerializerUtil.serializer(dataPartition,
-                                            mqProperties.isFilterTransactionEntry()));
+                                    mqProperties.isFilterTransactionEntry()));
                             sendMessage(data, index);
                         });
                     }
@@ -224,8 +227,8 @@ import java.util.stream.Collectors;
             } else {
                 final int partition = destination.getPartition() != null ? destination.getPartition() : 0;
                 Message data = new Message(topicName,
-                        ((RocketMQProducerConfig) this.mqProperties).getTag(),
+                    ((RocketMQProducerConfig) this.mqProperties).getTag(),
-                        CanalMessageSerializerUtil.serializer(message, mqProperties.isFilterTransactionEntry()));
+                    CanalMessageSerializerUtil.serializer(message, mqProperties.isFilterTransactionEntry()));
                 sendMessage(data, partition);
             }
         } else {
@@ -242,9 +245,9 @@ import java.util.stream.Collectors;
 
                 for (FlatMessage flatMessage : flatMessages) {
                     FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage,
-                            partitionNum,
+                        partitionNum,
-                            destination.getPartitionHash(),
+                        destination.getPartitionHash(),
-                            mqProperties.isDatabaseHash());
+                        mqProperties.isDatabaseHash());
                     int length = partitionFlatMessage.length;
                     for (int i = 0; i < length; i++) {
                         // 增加null判断,issue #3267
@@ -261,10 +264,12 @@ import java.util.stream.Collectors;
                         final int index = i;
                         template.submit(() -> {
                             List<Message> messages = flatMessagePart.stream()
-                                    .map(flatMessage -> new Message(topicName,
+                                .map(flatMessage -> new Message(topicName,
-                                            ((RocketMQProducerConfig) this.mqProperties).getTag(),
+                                    ((RocketMQProducerConfig) this.mqProperties).getTag(),
-                                            JSON.toJSONBytes(flatMessage, JSONWriter.Feature.WriteNulls, JSONWriter.Feature.LargeObject)))
+                                    JSON.toJSONBytes(flatMessage,
-                                    .collect(Collectors.toList());
+                                        JSONWriter.Feature.WriteNulls,
+                                        JSONWriter.Feature.LargeObject)))
+                                .collect(Collectors.toList());
                             // 批量发送
                             sendMessage(messages, index);
                         });
@@ -276,10 +281,10 @@ import java.util.stream.Collectors;
             } else {
                 final int partition = destination.getPartition() != null ? destination.getPartition() : 0;
                 List<Message> messages = flatMessages.stream()
-                        .map(flatMessage -> new Message(topicName,
+                    .map(flatMessage -> new Message(topicName,
-                                ((RocketMQProducerConfig) this.mqProperties).getTag(),
+                        ((RocketMQProducerConfig) this.mqProperties).getTag(),
-                                JSON.toJSONBytes(flatMessage, JSONWriter.Feature.WriteNulls,JSONWriter.Feature.LargeObject)))
+                        JSON.toJSONBytes(flatMessage, JSONWriter.Feature.WriteNulls, JSONWriter.Feature.LargeObject)))
-                        .collect(Collectors.toList());
+                    .collect(Collectors.toList());
                 // 批量发送
                 sendMessage(messages, partition);
             }
@@ -304,14 +309,19 @@ import java.util.stream.Collectors;
         }
     }
 
-    @SuppressWarnings("deprecation") private void sendMessage(List<Message> messages, int partition) {
+    @SuppressWarnings("deprecation")
+    private void sendMessage(List<Message> messages, int partition) {
         if (messages.isEmpty()) {
             return;
         }
 
         // 获取一下messageQueue
         DefaultMQProducerImpl innerProducer = this.defaultMQProducer.getDefaultMQProducerImpl();
-        TopicPublishInfo topicInfo = innerProducer.getTopicPublishInfoTable().get(messages.get(0).getTopic());
+        String topic = messages.get(0).getTopic();
+        if (StringUtils.isNotBlank(this.defaultMQProducer.getNamespace())) {
+            topic = this.defaultMQProducer.getNamespace() + NAMESPACE_SEPARATOR + topic;
+        }
+        TopicPublishInfo topicInfo = innerProducer.getTopicPublishInfoTable().get(topic);
         if (topicInfo == null) {
             for (Message message : messages) {
                 sendMessage(message, partition);
@@ -346,7 +356,8 @@ import java.util.stream.Collectors;
         }
     }
 
-    @Override public void stop() {
+    @Override
+    public void stop() {
         logger.info("## Stop RocketMQ producer##");
         this.defaultMQProducer.shutdown();
         if (sendPartitionExecutor != null) {
@@ -361,6 +372,10 @@ import java.util.stream.Collectors;
             topicName = this.defaultMQProducer.withNamespace(topicName);
             DefaultMQProducerImpl innerProducer = this.defaultMQProducer.getDefaultMQProducerImpl();
             TopicPublishInfo topicInfo = innerProducer.getTopicPublishInfoTable().get(topicName);
+            if (topicInfo == null) {
+                innerProducer.getMqClientFactory().updateTopicRouteInfoFromNameServer(topicName);
+            }
+            topicInfo = innerProducer.getTopicPublishInfoTable().get(topicName);
             if (topicInfo == null) {
                 return null;
             } else {

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.9-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

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.9-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<groupId>com.alibaba.otter</groupId>

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

@@ -158,7 +158,7 @@ public class JsonConversion {
                     throw new IllegalArgumentException("illegal json data");
                 }
                 return new Json_Value(Json_enum_type.STRING,
-                    buffer.rewind().forward((int) n).getFixString((int) str_len, charset));
+                    buffer.rewind().forward((int) n).getFixLengthString((int) str_len, charset));
             case JSONB_TYPE_OPAQUE:
                 /*
                  * There should always be at least one byte, which tells the field type of the
@@ -257,7 +257,7 @@ public class JsonConversion {
             // entry, always two
             // bytes.
             long key_length = m_data.getUint16();
-            return m_data.rewind().forward((int) key_offset).getFixString((int) key_length, charset);
+            return m_data.rewind().forward((int) key_offset).getFixLengthString((int) key_length, charset);
         }
 
         public Json_Value element(int i, Charset charset) {
@@ -400,7 +400,7 @@ public class JsonConversion {
                                    }
                                    buf.append('"').append(text).append('"');
                                } else {
-                                   text = m_data.getFixString((int) m_length, charset);
+                                   text = m_data.getFixLengthString((int) m_length, charset);
                                    buf.append('"').append(escapse(text)).append('"');
                                }