[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,
-  `name` varchar(45) NOT NULL,
+  `category` varchar(255) 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,
-  `ip` varchar(63) NOT NULL,
+  `name` varchar(255) 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,
-  `password` varchar(128) NOT NULL,
-  `name` varchar(31) NOT NULL,
-  `roles` varchar(31) NOT NULL,
+  `username` varchar(255) NOT NULL,
+  `password` varchar(255) NOT NULL,
+  `name` varchar(255) 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>

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"
-         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">
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.8-SNAPSHOT</version>
+        <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/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>

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>
 
@@ -136,7 +136,7 @@
                 <artifactId>kafka-clients</artifactId>
                 <version>${kafka_version}</version>
             </dependency>
-            <!-- rabbotmq -->
+            <!-- rabbitmq -->
             <dependency>
                 <groupId>com.rabbitmq</groupId>
                 <artifactId>amqp-client</artifactId>

client-adapter/rdb/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>canal.client-adapter</artifactId>
         <groupId>com.alibaba.otter</groupId>
-        <version>1.1.8-SNAPSHOT</version>
+        <version>1.1.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>

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/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/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:
-                        // *start++ = thd->variables.binlog_ddl_skip_rewrite;
+                        break;
+                    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
@@ -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,7 +17,12 @@ canal.instance.master.position=
 canal.instance.master.timestamp=
 canal.instance.master.gtid=
 
+# multi stream for polardbx
+canal.instance.multi.stream.on=false
+
 # ssl
+#canal.instance.master.sslMode=DISABLED
+#canal.instance.master.tlsVersions=
 #canal.instance.master.trustCertificateKeyStoreType=
 #canal.instance.master.trustCertificateKeyStoreUrl=
 #canal.instance.master.trustCertificateKeyStorePassword=
@@ -20,11 +30,6 @@ canal.instance.master.gtid=
 #canal.instance.master.clientCertificateKeyStoreUrl=
 #canal.instance.master.clientCertificateKeyStorePassword=
 
-# rds oss binlog
-canal.instance.rds.accesskey=
-canal.instance.rds.secretkey=
-canal.instance.rds.instanceId=
-
 # table meta tsdb info
 canal.instance.tsdb.enable=true
 #canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb
@@ -64,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
 #################################################

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,
-  `name` varchar(45) NOT NULL,
+  `category` varchar(255) 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,
-  `ip` varchar(63) NOT NULL,
+  `name` varchar(255) 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,
-  `password` varchar(128) NOT NULL,
-  `name` varchar(31) NOT NULL,
-  `roles` varchar(31) NOT NULL,
+  `username` varchar(255) NOT NULL,
+  `password` varchar(255) NOT NULL,
+  `name` varchar(255) 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

@@ -107,8 +107,7 @@ public class MysqlConnector {
                 sslVersion,
                 sslCipher);
         } catch (Exception e) {
-            logger.info("Can't show SSL status, server may not standard MySQL server: {}", e.toString());
-            logger.debug("show SSL status exception", e);
+            logger.warn("Can't show SSL status, server may not standard MySQL server", e);
         }
     }
 
@@ -197,6 +196,7 @@ 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) {
@@ -224,9 +224,10 @@ public class MysqlConnector {
         serverVersion = handshakePacket.serverVersion; // 记录serverVersion
         logger.info("handshake initialization packet received, prepare the client authentication packet to send");
         // 某些老协议的 server 默认不返回 auth plugin，需要使用默认的 mysql_native_password
-        String authPluginName = (handshakePacket.authPluginName != null
-                                 && handshakePacket.authPluginName.length > 0) ? new String(
-                                     handshakePacket.authPluginName) : "mysql_native_password";
+        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;
@@ -254,9 +255,7 @@ 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];
@@ -268,7 +267,7 @@ public class MysqlConnector {
                 body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout);
             } else {
                 byte[] authData = null;
-                String pluginName = null;
+                String pluginName = authPluginName;
                 if (marker == 1) {
                     AuthSwitchRequestMoreData packet = new AuthSwitchRequestMoreData();
                     packet.fromBytes(body);
@@ -295,23 +294,30 @@ public class MysqlConnector {
                     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);
+                        // support full auth
+                        // clientAuth提前采用了sha2编码,会减少一次auth交互
+                        header = cachingSha2PasswordFullAuth(channel,
+                            header,
+                            getPassword().getBytes(),
+                            clientAuth.getScrumbleBuff());
                         body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout);
                     } else {
-                        header = PacketManager.readHeader(channel, 4);
+                        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);
@@ -348,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);

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);
-        gtidMap.put(mariaGtid.getDomainId(), mariaGtid);
+        if (StringUtils.isNotEmpty(str)) {
+            // 兼容 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

@@ -3,32 +3,30 @@ 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;
 
-import org.apache.commons.lang.StringUtils;
-
 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 clientCapability = 0;
-        clientCapability |= Capability.CLIENT_LONG_PASSWORD;
-        clientCapability |= Capability.CLIENT_LONG_FLAG;
-        clientCapability |= Capability.CLIENT_PROTOCOL_41;
-        clientCapability |= Capability.CLIENT_TRANSACTIONS;
-        clientCapability |= Capability.CLIENT_SECURE_CONNECTION;
-        clientCapability |= Capability.CLIENT_MULTI_STATEMENTS;
-        clientCapability |= Capability.CLIENT_PLUGIN_AUTH;
-        clientCapability |= Capability.CLIENT_PLUGIN_AUTH_LENENC_CLIENT_DATA;
+        int capability = clientCapability;
         if (getDatabaseName() != null) {
-            clientCapability |= Capability.CLIENT_CONNECT_WITH_DB;
+            capability |= Capability.CLIENT_CONNECT_WITH_DB;
         }
-        ByteHelper.writeUnsignedIntLittleEndian(clientCapability, out);
+        ByteHelper.writeUnsignedIntLittleEndian(capability, out);
 
         // 2. write max_packet_size
         ByteHelper.writeUnsignedIntLittleEndian(MSC.MAX_PACKET_LENGTH, out);

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

@@ -147,8 +147,7 @@ public abstract class BioSocketChannelPool {
                         ksIS.close();
                     } catch (IOException e) {
                         // can't close input stream, but keystore can be properly initialized so we
-                        // shouldn't throw
-                        // this exception
+                        // shouldn't throw this exception
                     }
                 }
             }
@@ -194,8 +193,7 @@ public abstract class BioSocketChannelPool {
                     trustStoreIS.close();
                 } catch (IOException e) {
                     // can't close input stream, but keystore can be properly initialized so we
-                    // shouldn't throw this
-                    // exception
+                    // shouldn't throw this exception
                 }
             }
         }

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

@@ -29,10 +29,9 @@ public abstract class SocketChannelPool {
 
     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);
+            throw new UnsupportedOperationException("canal socketChannel netty not support ssl mode: " + sslMode);
         } else {
             SocketAddress remoteSocketAddress = channel.getRemoteSocketAddress();
             try {

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/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/spring/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>

meta/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>

parse/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.parse</artifactId>

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

@@ -46,8 +46,7 @@ import com.taobao.tddl.dbsync.binlog.LogEvent;
 public class MysqlEventParser extends AbstractMysqlEventParser implements CanalEventParser, CanalHASwitchable {
 
     private CanalHAController    haController                      = null;
-
-    private int                  defaultConnectionTimeoutInSeconds = 30;       // sotimeout
+    private int                  defaultConnectionTimeoutInSeconds = 30;
     private int                  receiveBufferSize                 = 64 * 1024;
     private int                  sendBufferSize                    = 64 * 1024;
     // 数据库信息
@@ -69,10 +68,8 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
     private int                  dumpErrorCount                    = 0;        // binlogDump失败异常计数
     private int                  dumpErrorCountThreshold           = 2;        // binlogDump失败异常计数阀值
     private boolean              rdsOssMode                        = false;
-    private boolean              autoResetLatestPosMode            = false;    // true:
-                                                                                // binlog被删除之后，自动按最新的数据订阅
-
-    private boolean multiStreamEnable;//support for polardbx binlog-x
+    private boolean              autoResetLatestPosMode            = false;    // binlog被删除之后，自动按最新的数据订阅
+    private boolean              multiStreamEnable;                            // support for polardbx binlog-x
 
     protected ErosaConnection buildErosaConnection() {
         return buildMysqlConnection(this.runningInfo);
@@ -255,7 +252,6 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
                 reconnect = true;
                 logger.warn("connect failed by ", e);
             }
-
         }
 
         public MysqlConnection getMysqlConnection() {
@@ -706,8 +702,10 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
      */
     private EntryPosition findStartPosition(MysqlConnection mysqlConnection) {
         try {
-            String showSql = multiStreamEnable ?
-                    "show binlog events with " + destination + " limit 1" : "show binlog events limit 1";
+            String showSql = "show binlog events limit 1";
+            if (multiStreamEnable) {
+                showSql = "show binlog events with " + destination + " limit 1";
+            }
             ResultSetPacket packet = mysqlConnection.query(showSql);
             List<String> fields = packet.getFieldValues();
             if (CollectionUtils.isEmpty(fields)) {
@@ -728,7 +726,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
     @SuppressWarnings("unused")
     private SlaveEntryPosition findSlavePosition(MysqlConnection mysqlConnection) {
         try {
-            String showSql = "show slave stauts";
+            String showSql = "show slave status";
             if (mysqlConnection.atLeastMySQL84()) {
                 // 兼容mysql 8.4
                 showSql = "show replica status";

pom.xml

@@ -4,7 +4,7 @@
     <artifactId>canal</artifactId>
     <packaging>pom</packaging>
     <name>canal module for otter ${project.version}</name>
-    <version>1.1.8-SNAPSHOT</version>
+    <version>1.1.8</version>
     <url>https://github.com/alibaba/canal</url>
     <parent>
         <groupId>org.sonatype.oss</groupId>
@@ -98,7 +98,7 @@
         <java_source_version>1.8</java_source_version>
         <java_target_version>1.8</java_target_version>
         <file_encoding>UTF-8</file_encoding>
-        <javadoc_skip>true</javadoc_skip>
+        <javadoc_skip>false</javadoc_skip>
         <spring_version>5.3.26</spring_version>
         <log4j_version>2.17.0</log4j_version>
         <rocketmq_version>4.9.8</rocketmq_version>

prometheus/pom.xml

@@ -3,13 +3,13 @@
     <parent>
         <artifactId>canal</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>
     <artifactId>canal.prometheus</artifactId>
-    <version>1.1.8-SNAPSHOT</version>
+    <version>1.1.8</version>
     <name>canal prometheus module for otter ${project.version}</name>
     <dependencies>
         <dependency>

protocol/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>

server/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.server</artifactId>

server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java

@@ -1,5 +1,15 @@
 package com.alibaba.otter.canal.server;
 
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.commons.lang.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.slf4j.MDC;
+
 import com.alibaba.otter.canal.connector.core.config.MQProperties;
 import com.alibaba.otter.canal.connector.core.producer.MQDestination;
 import com.alibaba.otter.canal.connector.core.spi.CanalMQProducer;
@@ -9,17 +19,6 @@ import com.alibaba.otter.canal.instance.core.CanalMQConfig;
 import com.alibaba.otter.canal.protocol.ClientIdentity;
 import com.alibaba.otter.canal.protocol.Message;
 import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
-import org.apache.commons.lang.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.slf4j.MDC;
-
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
 
 public class CanalMQStarter {
 
@@ -107,24 +106,32 @@ public class CanalMQStarter {
     public synchronized void startDestination(String destination) {
         CanalInstance canalInstance = canalServer.getCanalInstances().get(destination);
         if (canalInstance != null) {
-            stopDestination(destination);
-            CanalMQRunnable canalMQRunnable = new CanalMQRunnable(destination);
-            canalMQWorks.put(canalInstance.getDestination(), canalMQRunnable);
-            executorService.execute(canalMQRunnable);
-            logger.info("## Start the MQ work of destination:" + destination);
+            CanalMQRunnable canalMQRunnable = canalMQWorks.get(destination);
+            if (Objects.isNull(canalMQRunnable)) {
+                canalMQRunnable = new CanalMQRunnable(destination);
+                canalMQWorks.put(canalInstance.getDestination(), canalMQRunnable);
+                // 触发一下任务启动
+                Future future = executorService.submit(canalMQRunnable);
+                canalMQRunnable.setFuture(future);
+                logger.info("## Start the MQ work of destination:" + destination);
+            } else {
+                // 主段时间内的zk出现session time out，会默认优先当前节点抢占成功
+                // 如果没有出发过stop动作，这里可以忽略start的启动
+                logger.info("## Start the MQ work of destination:" + destination + " , ignore stop");
+            }
         }
     }
 
     public synchronized void stopDestination(String destination) {
-        CanalMQRunnable canalMQRunable = canalMQWorks.get(destination);
-        if (canalMQRunable != null) {
-            canalMQRunable.stop();
+        CanalMQRunnable canalMQRunnable = canalMQWorks.get(destination);
+        if (canalMQRunnable != null) {
+            canalMQRunnable.stop(true);
             canalMQWorks.remove(destination);
             logger.info("## Stop the MQ work of destination:" + destination);
         }
     }
 
-    private void worker(String destination, AtomicBoolean destinationRunning) {
+    private void worker(String destination, AtomicBoolean destinationRunning, CountDownLatch latch) {
         while (!running || !destinationRunning.get()) {
             try {
                 Thread.sleep(100);
@@ -206,6 +213,9 @@ public class CanalMQStarter {
                 logger.error("process error!", e);
             }
         }
+
+        // 确保一下关闭
+        latch.countDown();
     }
 
     private class CanalMQRunnable implements Runnable {
@@ -218,13 +228,31 @@ public class CanalMQStarter {
 
         private AtomicBoolean running = new AtomicBoolean(true);
 
+        private CountDownLatch latch   = new CountDownLatch(1);
+
+        private Future         future;
+
         @Override
         public void run() {
-            worker(destination, running);
+            worker(destination, running, latch);
         }
 
-        public void stop() {
+        public void stop(boolean wait) {
             running.set(false);
+            if (wait) {
+                try {
+                    // 触发一下interrupt
+                    future.cancel(true);
+                    // 等待MQ发送线程的正常退出
+                    latch.await();
+                } catch (InterruptedException e) {
+                    // ignore
+                }
+            }
+        }
+
+        public void setFuture(Future future) {
+            this.future = future;
         }
     }
 }

sink/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>

store/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>