[maven-release-plugin] prepare release canal-1.1.8

Signed-off-by: BruceAko <chongzhi@hust.edu.cn>

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

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

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

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

client-adapter/es7x/pom.xml

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

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

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

client-adapter/hbase/pom.xml

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

client-adapter/kudu/pom.xml

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

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.8</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>

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/java/com/alibaba/otter/canal/adapter/launcher/loader/AdapterProcessor.java

@@ -32,27 +32,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 +63,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 +173,7 @@ public class AdapterProcessor {
         }
 
         int retry = canalClientConfig.getRetries() == null
-            || canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries();
+                    || canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries();
         if (retry == -1) {
             // 重试次数-1代表异常时一直阻塞重试
             retry = Integer.MAX_VALUE;

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/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.8</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.8</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

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

@@ -352,13 +352,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.8</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.8</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.8</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

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

@@ -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/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返回反引号或空字符串
      *

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

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

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

@@ -1,19 +1,18 @@
 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/>
- * 
+ *
  * @author jianghang 2012-10-29 下午08:04:06
  * @version 1.0.0
  */
@@ -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.8</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<artifactId>canal.common</artifactId>
@@ -87,6 +87,10 @@
 			<groupId>com.google.code.findbugs</groupId>
 			<artifactId>jsr305</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>com.alibaba</groupId>
+			<artifactId>druid</artifactId>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework</groupId>
 			<artifactId>spring-test</artifactId>

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

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

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

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

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.8</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) {

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

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

@@ -620,12 +620,30 @@ public class QueryLogEvent extends LogEvent {
     public static final int Q_DEFAULT_TABLE_ENCRYPTION        = 20;
 
     /**
-     * @since percona 8.0.31 Replicate ddl_skip_rewrite.
+     * Replicate PolarDB-X
+     * 
+     * @since PolarDB-X 8.0.32
+     */
+    public static final int Q_OPT_FLASHBACK_AREA              = 21;
+
+    /**
+     * Replicate PolarDB-X
+     * 
+     * @since PolarDB-X 8.0.32
+     */
+    public static final int Q_OPT_INDEX_FORMAT_GPP_ENABLED    = 22;
+
+    /**
+     * Replicate ddl_skip_rewrite.
+     *
+     * @since percona 8.0.31
      */
     public static final int Q_DDL_SKIP_REWRITE                = 21;
 
     /**
-     * @since percona 8.0.31 Replicate Q_WSREP_SKIP_READONLY_CHECKS.
+     * Replicate Q_WSREP_SKIP_READONLY_CHECKS.
+     *
+     * @since percona 8.0.31
      */
     public static final int Q_WSREP_SKIP_READONLY_CHECKS      = 128;
 
@@ -643,6 +661,15 @@ public class QueryLogEvent extends LogEvent {
 
     public static final int Q_CHARACTER_SET_COLLATIONS        = 131;
 
+    /**
+     * support PolarDB-X used for storing snapshot tso or commit tso snapshot tso is
+     * stored in XA End Event with Query_log commit tso is stored in XA Commit Event
+     * with Query_log
+     */
+    public static final int Q_LIZARD_COMMIT_GCN               = 200;
+
+    public static final int Q_LIZARD_PREPARE_GCN              = 201;
+
     private final void unpackVariables(LogBuffer buffer, final int end) throws IOException {
         int code = -1;
         try {
@@ -740,9 +767,22 @@ public class QueryLogEvent extends LogEvent {
                     case Q_DEFAULT_TABLE_ENCRYPTION:
                         // *start++ = thd->variables.default_table_encryption;
                         buffer.forward(1);
-                    case Q_DDL_SKIP_REWRITE:
+                        break;
-                        // *start++ = thd->variables.binlog_ddl_skip_rewrite;
+                    case Q_OPT_FLASHBACK_AREA:
+                        if (compatiablePercona) {
+                            // percona
+                            // *start++ = thd->variables.binlog_ddl_skip_rewrite;
+                            buffer.forward(1);
+                        }else {
+                            // PolarDB-X
+                            // *start++ = thd->variables.opt_flashback_area;
+                            buffer.forward(1);
+                        }
+                        break;
+                    case Q_OPT_INDEX_FORMAT_GPP_ENABLED :
+                        // *start++ = thd->variables.opt_index_format_gpp_enabled;
                         buffer.forward(1);
+                        break;
                     case Q_HRNOW:
                         // https://github.com/alibaba/canal/issues/4940
                         // percona 和 mariadb各自扩展mysql binlog的格式后有冲突
@@ -779,6 +819,15 @@ public class QueryLogEvent extends LogEvent {
                         int count = buffer.getUint8();
                         // character_set_collations= Lex_cstring((const char *) pos0 , (const char *) pos);
                         buffer.forward(count * 4);
+                        break;
+                    case Q_LIZARD_COMMIT_GCN:
+                        // commitGCN = buffer.getLong64();
+                        buffer.forward(8);
+                        break;
+                    case Q_LIZARD_PREPARE_GCN:
+                        // prepareGCN = buffer.getLong64();
+                        buffer.forward(8);
+                        break;
                     default:
                         /*
                          * That's why you must write status vars in growing
@@ -819,7 +868,7 @@ public class QueryLogEvent extends LogEvent {
             case Q_MASTER_DATA_WRITTEN_CODE:
                 return "Q_MASTER_DATA_WRITTEN_CODE";
             case Q_INVOKER:
-                return "case Q_INVOKER";
+                return "Q_INVOKER";
             case Q_MICROSECONDS:
                 return "Q_MICROSECONDS";
             case Q_UPDATED_DB_NAMES:
@@ -834,9 +883,13 @@ public class QueryLogEvent extends LogEvent {
                 return "Q_SQL_REQUIRE_PRIMARY_KEY";
             case Q_DEFAULT_TABLE_ENCRYPTION:
                 return "Q_DEFAULT_TABLE_ENCRYPTION";
-            case Q_DDL_SKIP_REWRITE:
+            case Q_OPT_FLASHBACK_AREA:
+                // or Q_DDL_SKIP_REWRITE
+                return "Q_DDL_SKIP_REWRITE";
+            case Q_OPT_INDEX_FORMAT_GPP_ENABLED:
                 return "Q_DDL_SKIP_REWRITE";
             case Q_HRNOW:
+                // or Q_WSREP_SKIP_READONLY_CHECKS
                 return "Q_HRNOW";
             case Q_XID:
                 return "Q_XID";

deployer/pom.xml

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

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

@@ -5,6 +5,11 @@
 # enable gtid use true/false
 canal.instance.gtidon=false
 
+# rds oss binlog
+canal.instance.rds.accesskey=
+canal.instance.rds.secretkey=
+canal.instance.rds.instanceId=
+
 # position info
 canal.instance.master.address=127.0.0.1:3306
 canal.instance.master.journal.name=
@@ -12,10 +17,18 @@ canal.instance.master.position=
 canal.instance.master.timestamp=
 canal.instance.master.gtid=
 
-# rds oss binlog
+# multi stream for polardbx
-canal.instance.rds.accesskey=
+canal.instance.multi.stream.on=false
-canal.instance.rds.secretkey=
+
-canal.instance.rds.instanceId=
+# ssl
+#canal.instance.master.sslMode=DISABLED
+#canal.instance.master.tlsVersions=
+#canal.instance.master.trustCertificateKeyStoreType=
+#canal.instance.master.trustCertificateKeyStoreUrl=
+#canal.instance.master.trustCertificateKeyStorePassword=
+#canal.instance.master.clientCertificateKeyStoreType=
+#canal.instance.master.clientCertificateKeyStoreUrl=
+#canal.instance.master.clientCertificateKeyStorePassword=
 
 # table meta tsdb info
 canal.instance.tsdb.enable=true
@@ -56,7 +69,4 @@ canal.mq.partition=0
 #canal.mq.partitionsNum=3
 #canal.mq.dynamicTopicPartitionNum=test.*:4,mycanal:6
 #canal.mq.partitionHash=test.table:id^name,.*\\..*
-#
-# multi stream for polardbx
-canal.instance.multi.stream.on=false
 #################################################

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

@@ -36,4 +36,4 @@
 		<property name="secretkey" value="${canal.aliyun.secretkey:}" />
 		<property name="instanceId" value="${canal.instance.rds.instanceId:}" />
 	</bean>
-</beans>
+</beans>

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

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

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

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

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

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

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

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

docker/image/admin/health.sh

@@ -1,6 +1,6 @@
 #!/bin/sh
 
-if [ -d "/home/admin/canal-server" ]; then
+if [ -d "/home/admin/canal-admin" ]; then
 	serverPort=`perl -le 'print $ENV{"server.port"}'`
     if [ -z "$serverPort" ] ; then
         serverPort=8089
@@ -27,4 +27,4 @@ else
     echo "[  OK  ]"
     status=1
 	error=0
-fi
+fi

docker/image/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,

docker/run.sh

@@ -97,6 +97,6 @@ fi
 
 MEMORY="-m 4096m"
 LOCALHOST=`getMyIp`
-cmd="docker run -d --privileged=true -it -h $LOCALHOST $CONFIG --name=canal-server $VOLUMNS $NET_MODE $PORTS $MEMORY canal/canal-server"
+cmd="docker run -d -it -h $LOCALHOST $CONFIG --name=canal-server $VOLUMNS $NET_MODE $PORTS $MEMORY canal/canal-server"
 echo $cmd
 eval $cmd

docker/run_admin.sh

@@ -86,6 +86,6 @@ fi
 
 MEMORY="-m 1024m"
 LOCALHOST=`getMyIp`
-cmd="docker run -d --privileged=true -it -h $LOCALHOST $CONFIG --name=canal-admin $VOLUMNS $NET_MODE $PORTS $MEMORY canal/canal-admin"
+cmd="docker run -d -it -h $LOCALHOST $CONFIG --name=canal-admin $VOLUMNS $NET_MODE $PORTS $MEMORY canal/canal-admin"
 echo $cmd
 eval $cmd

driver/pom.xml

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

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

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

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

@@ -24,8 +24,11 @@ public class MariaGTIDSet implements GTIDSet {
 
     @Override
     public void update(String str) {
-        MariaGtid mariaGtid = MariaGtid.parse(str);
+        if (StringUtils.isNotEmpty(str)) {
-        gtidMap.put(mariaGtid.getDomainId(), mariaGtid);
+            // 兼容 GTID 为空的情况，例如 mysql-bin.000001 的 GTID 是空的
+            MariaGtid mariaGtid = MariaGtid.parse(str);
+            gtidMap.put(mariaGtid.getDomainId(), mariaGtid);
+        }
     }
 
     public void add(MariaGtid mariaGtid) {

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

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

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

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

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

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

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

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

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

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

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

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

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

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

example/pom.xml

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

filter/pom.xml

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

instance/core/pom.xml

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

instance/manager/pom.xml

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

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

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

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

@@ -2,6 +2,7 @@ package com.alibaba.otter.canal.instance.manager;
 
 import java.util.Properties;
 
+import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.BeanFactory;
@@ -17,7 +18,7 @@ import com.alibaba.otter.canal.parse.CanalEventParser;
 
 /**
  * 基于manager生成对应的{@linkplain CanalInstance}
- * 
+ *
  * @author jianghang 2012-7-12 下午05:37:09
  * @version 1.0.0
  */
@@ -69,6 +70,9 @@ public class PlainCanalInstanceGenerator implements CanalInstanceGenerator {
     // ================ setter / getter ================
 
     private BeanFactory getBeanFactory(String springXml) {
+        if (!StringUtils.startsWithIgnoreCase(springXml, "classpath:")) {
+            springXml = "classpath:" + springXml;
+        }
         ApplicationContext applicationContext = new ClassPathXmlApplicationContext(springXml);
         return applicationContext;
     }