Compare commits

..

57 Commits

Author SHA1 Message Date
jianghang.loujh
0741ccda1e [maven-release-plugin] prepare release canal-1.1.8 2025-01-16 16:07:00 +08:00
jianghang.loujh
02fffc9e74 compatiable polardb-x 2025-01-13 22:07:12 +08:00
jianghang.loujh
5ca4c39313 compatiable percona & polardb-x 2025-01-13 22:02:09 +08:00
jianghang.loujh
3162d7093d fixed polardb-x variables(flashback & gpp) 2025-01-13 21:35:33 +08:00
jianghang.loujh
f350fbdc0a fixed mysql 8.0.30+ , support new variable 2025-01-08 10:10:33 +08:00
jianghang.loujh
2d0dcf0e5d fixed issue #5301, fixed caching_sha2_password full auth & client SHA2 2024-12-24 17:51:31 +08:00
jianghang.loujh
515c05c873 fixed issue #5301, fixed caching_sha2_password full auth 2024-12-24 16:39:21 +08:00
jianghang.loujh
7ee551cbf7 fixed issue #5301, fixed caching_sha2_password full auth 2024-12-24 16:33:41 +08:00
jianghang.loujh
923675aa49 fixed issue #5334 , adjust canal_manager default schema 2024-12-17 16:06:57 +08:00
jianghang.loujh
ebb541001c fixed issue #5337 , remove docker privileged 2024-12-17 15:52:01 +08:00
Chongzhi Deng
44ed76ebb8 fix typo (#5299)
Signed-off-by: BruceAko <chongzhi@hust.edu.cn>
2024-12-17 15:32:03 +08:00
jianghang.loujh
fe836f7515 fixed issue #5336 #5332 , etl bugfix 2024-12-17 15:29:51 +08:00
quanllong
b30544c750 bugfix: 兼容MariaDB GTID 为空的情况 (#5352) 2024-12-17 15:09:41 +08:00
lol
66483ef83f Feature/helm charts (#5285)
* canal-admin canal-server k8s helm chart

* 拼写错误
2024-10-14 16:48:05 +08:00
jianghang.loujh
b391188baf code format & fine-tune 2024-10-14 16:44:59 +08:00
lol
2a7a507987 fixed health.sh failed (#5281)
fixed health.sh failed
2024-10-14 16:44:15 +08:00
lanicc
f53749bfc5 fixed issue #5289 (#5290)
Co-authored-by: lanicc <liaonan@souche.com>
2024-10-14 16:18:21 +08:00
jianghang.loujh
80ca4366db fixed #5270 , add CanalMQStarter stop latch 2024-09-13 18:04:56 +08:00
lwd-coding
be0d945d7d 当zk出现session time out可能会导致instance出现多个消费者,从而出现乱序 (#5270) 2024-09-13 17:02:19 +08:00
jianghang.loujh
5a968b3715 fixed typo 2024-09-11 15:05:36 +08:00
jianghang.loujh
1e5b8a2021 fixed schema blackFilter & mysql 8.4 2024-08-29 13:07:26 +08:00
hbzheng8
ff82fd6532 直接close socket,解决TLS half-close policy报错问题 (#5240) 2024-08-29 12:47:50 +08:00
Yifan Xia
b060f46ccf Update AbstractRequest.java (#5208) 2024-08-29 12:47:09 +08:00
TK337
7e31b40531 support mysql 8.4 (#5231) 2024-08-29 12:43:06 +08:00
dingxiaobo
b2b3223d1d fix mysql 5.1 negotiate NPE. (#5255) 2024-08-29 12:42:31 +08:00
jianghang.loujh
fea4d8210a fixed pull #5191 , reformat 2024-07-16 15:02:10 +08:00
JZ
fb3d2c6811 add compressionType,enableChunking for pulsar client , upgrade dependencies for security (#5191)
* add support for listener name which is described in https://pulsar.apache.org/docs/next/concepts-multiple-advertised-listeners/

* upgrade for CVE-2023-20883  CVE-2022-1471  CVE-2023-20860

* add   enableChunking

* downgrade  pulsar client version

* fix

* 增加压缩算法
2024-07-16 14:59:29 +08:00
jianghang.loujh
3439c48948 fixed pull #5175 , reformat 2024-07-16 14:57:53 +08:00
iamssx
2b8a309ec7 fixed Rocketmq can not batch send messages when producer set namespace (#5202) 2024-07-16 14:51:56 +08:00
Yazid Lee
9373429015 fix: resource loading failure caused by incorrect parameter order in StringUtils.startsWithIgnoreCase (#5173) 2024-06-26 15:53:24 +08:00
lwd-coding
9493ff733d fix: 同个instance配置多个group避免共享CanalMsgConsumer (#5175) 2024-06-26 15:49:48 +08:00
Tass
31530dc9af fix(adapter): add args into adapter/restart.sh (#5149)
Signed-off-by: Tass <tasszz2k@gmail.com>
2024-06-26 15:34:30 +08:00
jianghang.loujh
b0f3ff8dcc fixed pull #5147, reformat & add more configs 2024-06-26 15:33:14 +08:00
dingxiaobo
5c61f2fb56 Feat, support SSL(TLS), improve SHA2 password. (#5147)
* feat #4489, mysql SSL

* feat #4489, mysql SHA2 and SSL

* merge master, fix conflict.

* improve SHA2 auth plugin, fix SSL status print.

* SSL key store, use url rather than path, keep same as JDBC driver.
2024-06-26 15:06:16 +08:00
jianghang.loujh
7bb619cf12 fixed #5154,avoid spring RCE 2024-06-06 15:29:17 +08:00
lulu2panpan
d66a15fe35 migrate sql parser from druid to polardbx-parser (#5097)
* migrate parser from druid to polardbx-parser

* migrate parser from druid to polardbx-parser(with version 5.4.19)

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

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

---------

Co-authored-by: agapple <jianghang.loujh@alibaba-inc.com>
2024-05-25 22:05:06 +08:00
jianghang.loujh
aa05d8e2a0 reformat 2024-05-22 09:46:13 +08:00
leoloel
5ff6449620 fix issue #5127 escape key of json data (#5130) 2024-05-22 09:41:50 +08:00
ziyou
bbea802914 feat(es8): es8 支持证书认证 (#5106) 2024-04-28 09:54:53 +08:00
JZ
e2219c88dd add support for listener name which is described in https://pulsar.apache.org/docs/next/concepts-multiple-advertised-listeners/ (#5118) 2024-04-28 09:53:39 +08:00
jianghang.loujh
a92e61f5a9 fixed issue #5073 , optimize docker build 2024-04-15 18:52:42 +08:00
jianghang.loujh
eaacc01b77 fixed issue #5094 , support function unique index 2024-04-15 15:34:50 +08:00
jianghang.loujh
caa46f3672 merge pull #5089 2024-04-03 10:45:29 +08:00
jianghang.loujh
c0ff058fe1 remove unnecessary bean copy 2024-03-18 15:58:53 +08:00
jianghang.loujh
059f21208d fixed issue #5066 , es7.x npe 2024-03-18 14:19:45 +08:00
jianghang.loujh
8ca900377a merge & reformat code 2024-03-08 11:58:46 +08:00
jackzlz
6dffcc8573 yml解析支持占位符 (#5045) 2024-03-08 11:56:30 +08:00
wbxlll
711971599f fix issue client-adapter.rdb模块在转换时间时丢失精度 #5052 (#5053)
* fix the issue that client-adapter.rdb module loses precision when converting microsencond time

* fix the issue that client-adapter.rdb module loses precision when converting to java.sql.Time or java.sql.Timestamp

* fix the issue that client-adapter.rdb module loses precision when converting to java.sql.Time or java.sql.Timestamp
2024-03-08 11:50:53 +08:00
Devin Zhang
87c7a95523 startup.sh支持JDK17启动 (#5060) 2024-03-08 11:48:16 +08:00
ChengDaqi2023
e255baff11 update com.rabbitmq:amqp-client 5.5.0 to 5.18.0 (#5050) 2024-02-26 19:25:48 +08:00
jianghang.loujh
d457cd835e fixed issue #4819 , optmize code 2024-02-05 17:01:33 +08:00
jianghang.loujh
cf9c64629e optimize split IPAndPort 2024-02-05 16:49:05 +08:00
jianghang.loujh
307547746c optimize test & connected mistake 2024-02-04 13:59:29 +08:00
jianghang.loujh
c7d5ed4ba1 fixed issue #5036 , connected mistake 2024-02-04 13:58:52 +08:00
jianghang.loujh
b9ed5caf05 format 2024-02-04 13:35:10 +08:00
jianghang.loujh
348d218120 fixed issue #5029 , add deserializer message 2024-02-04 13:28:27 +08:00
zoemak
e2a64daad8 fixed issue #5037 ,使用client-adapter.rdb模块进行ETL同步,源库为MySQL,数据库名有短横线,读取源数据时报SQL异常 (#5044) 2024-02-04 13:07:51 +08:00
164 changed files with 3528 additions and 755 deletions

View File

@@ -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>

View File

@@ -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>

View File

@@ -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,

View File

@@ -3,13 +3,13 @@
<parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId>
<version>1.1.8-SNAPSHOT</version>
<version>1.1.8</version>
<relativePath>../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.admin</artifactId>
<version>1.1.8-SNAPSHOT</version>
<version>1.1.8</version>
<packaging>pom</packaging>
<name>canal admin module for otter ${project.version}</name>
<modules>
@@ -36,7 +36,7 @@
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-dependencies</artifactId>
<version>2.5.4</version>
<version>2.5.15</version>
<type>pom</type>
<scope>import</scope>
</dependency>
@@ -48,7 +48,12 @@
<dependency>
<groupId>io.ebean</groupId>
<artifactId>ebean</artifactId>
<version>11.41.1</version>
<version>11.45.1</version>
</dependency>
<dependency>
<groupId>org.yaml</groupId>
<artifactId>snakeyaml</artifactId>
<version>2.0</version>
</dependency>
</dependencies>
</dependencyManagement>

160
charts/README.md Normal file
View File

@@ -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
```

View File

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

View File

@@ -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"

View File

@@ -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 }}

View File

@@ -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 }}

View File

@@ -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 }}

View File

@@ -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 }}

View File

@@ -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 }}

View File

@@ -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 }}

View File

@@ -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 }}

View File

@@ -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 }}

View File

@@ -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

View File

@@ -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

View File

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

View File

@@ -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"

View File

@@ -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 }}

View File

@@ -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 }}

View File

@@ -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 }}

View File

@@ -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 }}

View File

@@ -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 }}

View File

@@ -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 }}

View File

@@ -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 }}

View File

@@ -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

View File

@@ -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

BIN
charts/images/cluster.png Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 100 KiB

BIN
charts/images/server.png Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 157 KiB

View File

@@ -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"

View File

@@ -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"

5
charts/tests/test.sh Executable file
View File

@@ -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

View File

@@ -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>

View File

@@ -13,10 +13,7 @@ import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.otter.canal.client.adapter.clickhouse.config.MappingConfig;
import com.alibaba.otter.canal.client.adapter.clickhouse.config.MappingConfig.DbMapping;
import com.alibaba.otter.canal.client.adapter.clickhouse.support.SyncUtil;
import com.alibaba.otter.canal.client.adapter.support.AbstractEtlService;
import com.alibaba.otter.canal.client.adapter.support.AdapterConfig;
import com.alibaba.otter.canal.client.adapter.support.EtlResult;
import com.alibaba.otter.canal.client.adapter.support.Util;
import com.alibaba.otter.canal.client.adapter.support.*;
/**
* ClickHouse ETL 操作业务类
@@ -42,7 +39,8 @@ public class ClickHouseEtlService extends AbstractEtlService {
*/
public EtlResult importData(List<String> params) {
DbMapping dbMapping = config.getDbMapping();
String sql = "SELECT * FROM " + dbMapping.getDatabase() + "." + dbMapping.getTable();
DruidDataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
String sql = "SELECT * FROM " + SyncUtil.getSourceDbTableName(dbMapping, dataSource.getDbType());
return importData(sql, params);
}
@@ -128,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);

View File

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

View File

@@ -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>

View File

@@ -28,9 +28,9 @@ import com.google.common.cache.LoadingCache;
public class Util {
private static final Logger logger = LoggerFactory.getLogger(Util.class);
private static final Logger logger = LoggerFactory.getLogger(Util.class);
public static final String AUTO_GENERATED_PREFIX = "AUTO_GENERATED_";
public static final String AUTO_GENERATED_PREFIX = "AUTO_GENERATED_";
/**
* 通过DS执行sql
@@ -165,39 +165,38 @@ public class Util {
public static ThreadPoolExecutor newFixedDaemonThreadPool(int nThreads, long keepAliveTime) {
return new ThreadPoolExecutor(nThreads,
nThreads,
keepAliveTime,
TimeUnit.MILLISECONDS,
new SynchronousQueue<>(),
DaemonThreadFactory.daemonThreadFactory,
(r, exe) -> {
if (!exe.isShutdown()) {
try {
exe.getQueue().put(r);
} catch (InterruptedException e) {
// ignore
}
nThreads,
keepAliveTime,
TimeUnit.MILLISECONDS,
new SynchronousQueue<>(),
DaemonThreadFactory.daemonThreadFactory,
(r, exe) -> {
if (!exe.isShutdown()) {
try {
exe.getQueue().put(r);
} catch (InterruptedException e) {
// ignore
}
}
);
});
}
public static ThreadPoolExecutor newSingleDaemonThreadExecutor(long keepAliveTime) {
return new ThreadPoolExecutor(1,
1,
keepAliveTime,
TimeUnit.MILLISECONDS,
new SynchronousQueue<>(),
DaemonThreadFactory.daemonThreadFactory,
(r, exe) -> {
if (!exe.isShutdown()) {
try {
exe.getQueue().put(r);
} catch (InterruptedException e) {
// ignore
}
1,
keepAliveTime,
TimeUnit.MILLISECONDS,
new SynchronousQueue<>(),
DaemonThreadFactory.daemonThreadFactory,
(r, exe) -> {
if (!exe.isShutdown()) {
try {
exe.getQueue().put(r);
} catch (InterruptedException e) {
// ignore
}
});
}
});
}
public final static String timeZone; // 当前时区
@@ -327,4 +326,56 @@ public class Util {
return null;
}
/**
* Check if the time string has millisecond or microsecond units
*
* @param timeStr time string
* @return boolean
*/
public static boolean isAccuracyOverSecond(String timeStr) {
if (StringUtils.isEmpty(timeStr)) {
return false;
}
String[] times = StringUtils.split(timeStr, ".");
return times.length > 1 && !times[times.length - 1].isEmpty();
}
/**
* Check if the datetime string has microsecond or nanosecond units
*
* @param datetimeStr datetime string
* @return boolean
*/
public static boolean isAccuracyOverMillisecond(String datetimeStr) {
if (StringUtils.isEmpty(datetimeStr)) {
return false;
}
String[] times = StringUtils.split(datetimeStr, ".");
return times.length > 1 && times[times.length - 1].length() > 3;
}
/**
* MySQL has fractional seconds support for TIME, DATETIME, and TIMESTAMP
* values, with up to microseconds (6 digits) precision. ISO-8601 standard
* format is with up to nanoseconds (9 digits) precision, which is sufficient
* for storing MySQL time-related data.
*
* @param datetimeStr datetime string
* @return LocalDateTime
*/
public static LocalDateTime parseISOLocalDateTime(String datetimeStr) {
if (StringUtils.isEmpty(datetimeStr)) {
return null;
}
datetimeStr = datetimeStr.trim();
try {
// Replace SPACE in middle with `T` to comply with ISO-8601 standard format
return LocalDateTime.parse(datetimeStr.replace(" ", "T"));
} catch (Exception e) {
logger.error("Convert datetime string to ISOLocalDateTime fail:{}", datetimeStr, e);
}
return null;
}
}

View File

@@ -4,9 +4,11 @@ import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import org.apache.commons.lang.StringUtils;
import org.springframework.boot.context.properties.bind.Bindable;
import org.springframework.boot.context.properties.bind.Binder;
@@ -18,10 +20,14 @@ import org.springframework.core.env.PropertySource;
import org.springframework.core.io.ByteArrayResource;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.Resource;
import org.springframework.util.PropertyPlaceholderHelper;
public class YamlUtils {
public static <T> T resourceYmlToObj(String resource, String prefix, Class<T> clazz) {
if (!StringUtils.startsWithIgnoreCase(resource, "classpath:")) {
resource = "classpath:" + resource;
}
ClassPathResource classPathResource = new ClassPathResource(resource);
String content;
@@ -85,9 +91,15 @@ public class YamlUtils {
entry.setValue(((OriginTrackedValue) value).getValue());
}
}
ConfigurationPropertySource sources = new MapConfigurationPropertySource(properties);
Binder binder = new Binder(sources);
PropertyPlaceholderHelper propertyPlaceholderHelper = new PropertyPlaceholderHelper("${", "}");
Binder binder = new Binder(Arrays.asList(sources), value -> {
if (value instanceof String) {
return propertyPlaceholderHelper.replacePlaceholders((String) value, baseProperties);
}
return value;
});
return binder.bind(prefix, Bindable.of(clazz)).get();
} catch (Exception e) {
throw new RuntimeException(e);

View File

@@ -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>

View File

@@ -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>

View File

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

View File

@@ -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>

View File

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

View File

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

View File

@@ -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>

View File

@@ -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>

View File

@@ -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>

View File

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

View File

@@ -3,7 +3,7 @@
<parent>
<artifactId>canal.client-adapter</artifactId>
<groupId>com.alibaba.otter</groupId>
<version>1.1.8-SNAPSHOT</version>
<version>1.1.8</version>
<relativePath>../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
@@ -17,7 +17,7 @@
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-dependencies</artifactId>
<version>2.5.4</version>
<version>2.5.15</version>
<type>pom</type>
<scope>import</scope>
</dependency>

View File

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

View File

@@ -32,27 +32,27 @@ import com.alibaba.otter.canal.connector.core.spi.ProxyCanalMsgConsumer;
*/
public class AdapterProcessor {
private static final Logger logger = LoggerFactory.getLogger(AdapterProcessor.class);
private static final Logger logger = LoggerFactory.getLogger(AdapterProcessor.class);
private static final String CONNECTOR_SPI_DIR = "/plugin";
private static final String CONNECTOR_STANDBY_SPI_DIR = "/canal-adapter/plugin";
private static final String CONNECTOR_SPI_DIR = "/plugin";
private static final String CONNECTOR_STANDBY_SPI_DIR = "/canal-adapter/plugin";
private CanalMsgConsumer canalMsgConsumer;
private CanalMsgConsumer canalMsgConsumer;
private String canalDestination; // canal实例
private String groupId = null; // groupId
private List<List<OuterAdapter>> canalOuterAdapters; // 外部适配器
private CanalClientConfig canalClientConfig; // 配置
private ExecutorService groupInnerExecutorService; // 组内工作线程池
private volatile boolean running = false; // 是否运行中
private Thread thread = null;
private Thread.UncaughtExceptionHandler handler = (t, e) -> logger
private String canalDestination; // canal实例
private String groupId = null; // groupId
private List<List<OuterAdapter>> canalOuterAdapters; // 外部适配器
private CanalClientConfig canalClientConfig; // 配置
private ExecutorService groupInnerExecutorService; // 组内工作线程池
private volatile boolean running = false; // 是否运行中
private Thread thread = null;
private Thread.UncaughtExceptionHandler handler = (t, e) -> logger
.error("parse events has an error", e);
private SyncSwitch syncSwitch;
private SyncSwitch syncSwitch;
public AdapterProcessor(CanalClientConfig canalClientConfig, String destination, String groupId,
List<List<OuterAdapter>> canalOuterAdapters) {
List<List<OuterAdapter>> canalOuterAdapters){
this.canalClientConfig = canalClientConfig;
this.canalDestination = destination;
this.groupId = groupId;
@@ -63,9 +63,12 @@ public class AdapterProcessor {
// load connector consumer
ExtensionLoader<CanalMsgConsumer> loader = new ExtensionLoader<>(CanalMsgConsumer.class);
canalMsgConsumer = new ProxyCanalMsgConsumer(loader
.getExtension(canalClientConfig.getMode().toLowerCase(), destination, CONNECTOR_SPI_DIR,
CONNECTOR_STANDBY_SPI_DIR));
// see https://github.com/alibaba/canal/pull/5175
String key = destination + "_" + groupId;
canalMsgConsumer = new ProxyCanalMsgConsumer(loader.getExtension(canalClientConfig.getMode().toLowerCase(),
key,
CONNECTOR_SPI_DIR,
CONNECTOR_STANDBY_SPI_DIR));
Properties properties = canalClientConfig.getConsumerProperties();
properties.put(CanalConstants.CANAL_MQ_FLAT_MESSAGE, canalClientConfig.getFlatMessage());
@@ -170,7 +173,7 @@ public class AdapterProcessor {
}
int retry = canalClientConfig.getRetries() == null
|| canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries();
|| canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries();
if (retry == -1) {
// 重试次数-1代表异常时一直阻塞重试
retry = Integer.MAX_VALUE;
@@ -224,7 +227,7 @@ public class AdapterProcessor {
logger.error("finish turn off switch of destination:" + canalDestination);
} else {
canalMsgConsumer.ack();
logger.error(e.getMessage() + " Error sync but ACK!");
logger.error(e.getMessage() + " Error sync but ACK!", e);
}
}
Thread.sleep(500);

View File

@@ -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;
}
}

View File

@@ -91,6 +91,7 @@ canal.conf:
# hosts: 127.0.0.1:9300 # 127.0.0.1:9200 for rest mode
# properties:
# mode: transport # or rest
# # security.ca.path: /etc/es8/ca.crt
# # security.auth: test:123456 # only used for rest mode
# cluster.name: elasticsearch
# - name: kudu
@@ -113,4 +114,4 @@ canal.conf:
# jdbc.password: 123456
# batchSize: 3000
# scheduleTime: 600 # second unit
# threads: 3 # parallel threads
# threads: 3 # parallel threads

View File

@@ -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>

View File

@@ -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>

View File

@@ -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);

View File

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

View File

@@ -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>

View File

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

View File

@@ -1,21 +1,23 @@
package com.alibaba.otter.canal.client.adapter.rdb.support;
import com.alibaba.druid.DbType;
import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig;
import com.alibaba.otter.canal.client.adapter.support.Util;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.Reader;
import java.io.StringReader;
import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
import java.sql.*;
import java.time.LocalDateTime;
import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.Map;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.druid.DbType;
import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig;
import com.alibaba.otter.canal.client.adapter.support.Util;
public class SyncUtil {
private static final Logger logger = LoggerFactory.getLogger(SyncUtil.class);
@@ -220,11 +222,16 @@ public class SyncUtil {
pstmt.setTime(i, new java.sql.Time(((java.util.Date) value).getTime()));
} else if (value instanceof String) {
String v = (String) value;
java.util.Date date = Util.parseDate(v);
if (date != null) {
pstmt.setTime(i, new Time(date.getTime()));
} else {
pstmt.setNull(i, type);
if(Util.isAccuracyOverSecond(v)) {
//the java.sql.time doesn't support for even millisecond, only setObject works here.
pstmt.setObject(i, v);
}else {
java.util.Date date = Util.parseDate(v);
if (date != null) {
pstmt.setTime(i, new Time(date.getTime()));
} else {
pstmt.setNull(i, type);
}
}
} else {
pstmt.setNull(i, type);
@@ -238,11 +245,22 @@ public class SyncUtil {
} else if (value instanceof String) {
String v = (String) value;
if (!v.startsWith("0000-00-00")) {
java.util.Date date = Util.parseDate(v);
if (date != null) {
pstmt.setTimestamp(i, new Timestamp(date.getTime()));
} else {
pstmt.setNull(i, type);
if(Util.isAccuracyOverMillisecond(v)){
//convert to ISO-8601 standard format, with up to nanoseconds (9 digits) precision
LocalDateTime isoDatetime = Util.parseISOLocalDateTime(v);
if (isoDatetime != null) {
pstmt.setTimestamp(i, Timestamp.valueOf(isoDatetime));
} else {
//if can't convert, set to null
pstmt.setNull(i, type);
}
}else {
java.util.Date date = Util.parseDate(v);
if (date != null) {
pstmt.setTimestamp(i, new Timestamp(date.getTime()));
} else {
pstmt.setNull(i, type);
}
}
} else {
pstmt.setObject(i, value);
@@ -266,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返回反引号或空字符串
*

View File

@@ -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>

View File

@@ -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>

View File

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

View File

@@ -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

View File

@@ -78,14 +78,15 @@ public class KafkaCanalConnector implements CanalMQConnector {
return;
}
connected = true;
if (kafkaConsumer == null && !flatMessage) {
kafkaConsumer = new KafkaConsumer<>(properties);
}
if (kafkaConsumer2 == null && flatMessage) {
kafkaConsumer2 = new KafkaConsumer<>(properties);
}
connected = true;
}
/**

View File

@@ -1,5 +1,13 @@
package com.alibaba.otter.canal.client.pulsarmq;
import java.util.List;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang.StringUtils;
import org.apache.pulsar.client.api.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.fastjson2.JSON;
import com.alibaba.otter.canal.client.CanalMQConnector;
import com.alibaba.otter.canal.client.CanalMessageDeserializer;
@@ -9,13 +17,6 @@ import com.alibaba.otter.canal.protocol.FlatMessage;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.exception.CanalClientException;
import com.google.common.collect.Lists;
import org.apache.commons.lang.StringUtils;
import org.apache.pulsar.client.api.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List;
import java.util.concurrent.TimeUnit;
/**
* PulsarMQ的连接
@@ -24,91 +25,98 @@ import java.util.concurrent.TimeUnit;
* 注意点:
* 1. 相比于canal {@linkplain SimpleCanalConnector}, 这里get和ack操作不能有并发, 必须是一个线程执行get后,内存里执行完毕ack后再取下一个get
* </pre>
*
* todo 重复消费的概率相当高。一次批处理中,只要有一个消息处理失败,则该批次全部重试
*
* @since 1.1.1
*/
public class PulsarMQCanalConnector implements CanalMQConnector {
private static final Logger logger = LoggerFactory.getLogger(PulsarMQCanalConnector.class);
private static final Logger logger = LoggerFactory
.getLogger(PulsarMQCanalConnector.class);
private volatile Messages<byte[]> lastGetBatchMessage;
/**
* 连接pulsar客户端
*/
private PulsarClient pulsarClient;
private PulsarClient pulsarClient;
/**
* 消费者
*/
private Consumer<byte[]> consumer;
private Consumer<byte[]> consumer;
/**
* 是否扁平化Canal消息内容
*/
private boolean isFlatMessage = false;
private boolean isFlatMessage = false;
/**
* 主题名称
*/
private String topic;
private String topic;
/**
* 环境连接URL
*/
private String serviceUrl;
private String serviceUrl;
/**
* 角色认证token
*/
private String roleToken;
private String roleToken;
/**
* listener name
*/
private String listenerName;
/**
* 订阅客户端名称
*/
private String subscriptName;
private String subscriptName;
/**
* 每次批量获取数据的最大条目数,默认30
*/
private int batchSize = 30;
private int batchSize = 30;
/**
* 与{@code batchSize}一起决定批量获取的数据大小
* 当:
* 与{@code batchSize}一起决定批量获取的数据大小 当:
* <p>
* 1. {@code batchSize} 条消息未消费时<br/>
* 2. 距上一次批量消费时间达到{@code batchTimeoutSeconds}秒时
* </p>
* 任一条件满足,即执行批量消费
*/
private int getBatchTimeoutSeconds = 30;
private int getBatchTimeoutSeconds = 30;
/**
* 批量处理消息时,一次批量处理的超时时间秒数
* <p>
* 该时间应该根据{@code batchSize}和{@code batchTimeoutSeconds}合理设置
* </p>
*/
private int batchProcessTimeoutSeconds = 60;
private int batchProcessTimeoutSeconds = 60;
/**
* 消费失败后的重试秒数,默认60秒
*/
private int redeliveryDelaySeconds = 60;
private int redeliveryDelaySeconds = 60;
/**
* 当客户端接收到消息,30秒还没有返回ack给服务端时,ack超时,会重新消费该消息
*/
private int ackTimeoutSeconds = 30;
private int ackTimeoutSeconds = 30;
/**
* 是否开启消息失败重试功能,默认开启
*/
private boolean isRetry = true;
private boolean isRetry = true;
/**
* <p>
* true重试(-RETRY)和死信队列(-DLQ)后缀为大写,有些地方创建的为小写,需确保正确
* </p>
*/
private boolean isRetryDLQUpperCase = false;
private boolean isRetryDLQUpperCase = false;
/**
* 最大重试次数
*/
private int maxRedeliveryCount = 128;
private int maxRedeliveryCount = 128;
/**
* 连接标识位,在连接或关闭连接后改变值
*/
private boolean connected = false;
private boolean connected = false;
/**
* 除必要参数外,其他参数使用默认值
@@ -118,16 +126,21 @@ public class PulsarMQCanalConnector implements CanalMQConnector {
* </p>
*
* @param isFlatMessage true使用扁平消息
* @param serviceUrl pulsar服务连接地址,通常为:pulsar:host:ip或http://host:ip
* @param roleToken 有对应topic的消费者权限的角色token
* @param topic 订阅主题
* @param serviceUrl pulsar服务连接地址,通常为:pulsar:host:ip或http://host:ip
* @param roleToken 有对应topic的消费者权限的角色token
* @param topic 订阅主题
* @param subscriptName 订阅和客户端名称,同一个订阅名视为同一个消费实例
* @date 2021/9/18 08:54
* @author chad
* @since 1 by chad at 2021/9/18 完善
*/
public PulsarMQCanalConnector(boolean isFlatMessage, String serviceUrl, String roleToken, String topic
, String subscriptName) {
public PulsarMQCanalConnector(boolean isFlatMessage, String serviceUrl, String roleToken, String topic,
String subscriptName){
this(isFlatMessage, serviceUrl, roleToken, topic, subscriptName, null);
}
public PulsarMQCanalConnector(boolean isFlatMessage, String serviceUrl, String roleToken, String topic,
String subscriptName, String listenerName){
this.isFlatMessage = isFlatMessage;
this.serviceUrl = serviceUrl;
this.roleToken = roleToken;
@@ -136,6 +149,7 @@ public class PulsarMQCanalConnector implements CanalMQConnector {
if (StringUtils.isEmpty(this.subscriptName)) {
throw new RuntimeException("Pulsar Consumer subscriptName required");
}
this.listenerName = listenerName;
}
/**
@@ -145,10 +159,31 @@ public class PulsarMQCanalConnector implements CanalMQConnector {
* @author chad
* @since 1 by chad at 2021/9/18 完善
*/
public PulsarMQCanalConnector(boolean isFlatMessage, String serviceUrl, String roleToken, String topic
, String subscriptName, int batchSize, int getBatchTimeoutSeconds, int batchProcessTimeoutSeconds
, int redeliveryDelaySeconds, int ackTimeoutSeconds, boolean isRetry, boolean isRetryDLQUpperCase
, int maxRedeliveryCount) {
public PulsarMQCanalConnector(boolean isFlatMessage, String serviceUrl, String roleToken, String topic,
String subscriptName, int batchSize, int getBatchTimeoutSeconds,
int batchProcessTimeoutSeconds, int redeliveryDelaySeconds, int ackTimeoutSeconds,
boolean isRetry, boolean isRetryDLQUpperCase, int maxRedeliveryCount){
this(isFlatMessage,
serviceUrl,
roleToken,
topic,
subscriptName,
batchSize,
getBatchTimeoutSeconds,
batchProcessTimeoutSeconds,
redeliveryDelaySeconds,
ackTimeoutSeconds,
isRetry,
isRetryDLQUpperCase,
maxRedeliveryCount,
null);
}
public PulsarMQCanalConnector(boolean isFlatMessage, String serviceUrl, String roleToken, String topic,
String subscriptName, int batchSize, int getBatchTimeoutSeconds,
int batchProcessTimeoutSeconds, int redeliveryDelaySeconds, int ackTimeoutSeconds,
boolean isRetry, boolean isRetryDLQUpperCase, int maxRedeliveryCount,
String listenerName){
this.isFlatMessage = isFlatMessage;
this.serviceUrl = serviceUrl;
this.roleToken = roleToken;
@@ -165,16 +200,20 @@ public class PulsarMQCanalConnector implements CanalMQConnector {
this.isRetry = isRetry;
this.isRetryDLQUpperCase = isRetryDLQUpperCase;
this.maxRedeliveryCount = maxRedeliveryCount;
this.listenerName = listenerName;
}
@Override
public void connect() throws CanalClientException {
// 连接创建客户端
try {
pulsarClient = PulsarClient.builder()
.serviceUrl(serviceUrl)
.authentication(AuthenticationFactory.token(roleToken))
.build();
ClientBuilder builder = PulsarClient.builder()
.serviceUrl(serviceUrl)
.authentication(AuthenticationFactory.token(roleToken));
if (StringUtils.isNotEmpty(listenerName)) {
builder.listenerName(listenerName);
}
pulsarClient = builder.build();
} catch (PulsarClientException e) {
throw new RuntimeException(e);
}
@@ -223,15 +262,14 @@ public class PulsarMQCanalConnector implements CanalMQConnector {
builder.subscriptionType(SubscriptionType.Failover);
builder
// 调用consumer.negativeAcknowledge(message) (即nack)来表示消费失败的消息
// 在指定的时间进行重新消费,默认是1分钟。
.negativeAckRedeliveryDelay(this.redeliveryDelaySeconds, TimeUnit.SECONDS)
.subscriptionName(this.subscriptName)
;
// 调用consumer.negativeAcknowledge(message) (即nack)来表示消费失败的消息
// 在指定的时间进行重新消费,默认是1分钟。
.negativeAckRedeliveryDelay(this.redeliveryDelaySeconds, TimeUnit.SECONDS)
.subscriptionName(this.subscriptName);
if (this.isRetry) {
DeadLetterPolicy.DeadLetterPolicyBuilder dlqBuilder = DeadLetterPolicy.builder()
// 最大重试次数
.maxRedeliverCount(this.maxRedeliveryCount);
// 最大重试次数
.maxRedeliverCount(this.maxRedeliveryCount);
// 指定重试队列,不是多个或通配符topic才能判断重试队列
if (!MQUtil.isPatternTag(this.topic)) {
String retryTopic = this.topic + (this.isRetryDLQUpperCase ? "-RETRY" : "-retry");
@@ -240,29 +278,26 @@ public class PulsarMQCanalConnector implements CanalMQConnector {
dlqBuilder.deadLetterTopic(dlqTopic);
}
//默认关闭,如果需要重试则开启
builder.enableRetry(true)
.deadLetterPolicy(dlqBuilder.build());
// 默认关闭,如果需要重试则开启
builder.enableRetry(true).deadLetterPolicy(dlqBuilder.build());
}
// ack超时
builder.ackTimeout(this.ackTimeoutSeconds, TimeUnit.SECONDS);
// pulsar批量获取消息设置
builder.batchReceivePolicy(new BatchReceivePolicy.Builder()
.maxNumMessages(this.batchSize)
.timeout(this.getBatchTimeoutSeconds, TimeUnit.SECONDS)
.build());
builder.batchReceivePolicy(new BatchReceivePolicy.Builder().maxNumMessages(this.batchSize)
.timeout(this.getBatchTimeoutSeconds, TimeUnit.SECONDS)
.build());
try {
this.consumer = builder.subscribe();
connected = true;
} catch (PulsarClientException e) {
throw new RuntimeException(e);
}
connected = true;
}
@Override
public void subscribe() throws CanalClientException {
this.subscribe(null);
@@ -283,7 +318,7 @@ public class PulsarMQCanalConnector implements CanalMQConnector {
* 不关注业务执行结果,只要收到消息即认识消费成功,自动ack
*
* @param timeout 阻塞获取消息的超时时间
* @param unit 时间单位
* @param unit 时间单位
* @return java.util.List<com.alibaba.otter.canal.protocol.Message>
* @date 2021/9/13 22:24
* @author chad
@@ -299,13 +334,14 @@ public class PulsarMQCanalConnector implements CanalMQConnector {
}
/**
* 关心业务执行结果,业务侧根据执行结果调用 {@link PulsarMQCanalConnector#ack()}或{@link PulsarMQCanalConnector#rollback()}
* 关心业务执行结果,业务侧根据执行结果调用
* {@link PulsarMQCanalConnector#ack()}或{@link PulsarMQCanalConnector#rollback()}
* <p>
* 本方法示支持多线程,在MQ保障顺序的前提下,也无法提供单Topic多线程
* </p>
*
* @param timeout 阻塞获取消息的超时时间
* @param unit 时间单位
* @param unit 时间单位
* @return java.util.List<com.alibaba.otter.canal.protocol.Message>
* @date 2021/9/13 22:26
* @author chad
@@ -339,7 +375,8 @@ public class PulsarMQCanalConnector implements CanalMQConnector {
* @return java.util.List<T>
* @date 2021/9/14 15:20
* @author chad
* @since 1 by chad at 2021/9/14 供{@link PulsarMQCanalConnector#getListWithoutAck(Long, TimeUnit)}
* @since 1 by chad at 2021/9/14
* 供{@link PulsarMQCanalConnector#getListWithoutAck(Long, TimeUnit)}
* 和{@link PulsarMQCanalConnector#getFlatListWithoutAck(Long, TimeUnit)}调用
*/
private <T> List<T> getListWithoutAck() {

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@@ -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>

View File

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

View File

@@ -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>

View File

@@ -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>

View File

@@ -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>

View File

@@ -61,4 +61,18 @@ public class PulsarMQConstants {
*/
public static final String PULSARMQ_ADMIN_SERVER_URL = ROOT + "." + "adminServerUrl";
/**
* Pulsar 监听器名字
*/
public static final String PULSARMQ_LISTENER_NAME = ROOT + "." + "listenerName";
/**
* Pulsar 开启chunking
*/
public static final String PULSARMQ_ENABLE_CHUNKING = ROOT + "." + "enableChunking";
/**
* Pulsar 压缩算法
*/
public static final String PULSARMQ_COMPRESSION_TYPE = ROOT + "." + "compressionType";
}

View File

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

View File

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

View File

@@ -73,6 +73,11 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
// 角色权限认证的token
builder.authentication(AuthenticationFactory.token(pulsarMQProducerConfig.getRoleToken()));
}
if (StringUtils.isNotEmpty(pulsarMQProducerConfig.getListenerName())) {
// listener name
builder.listenerName(pulsarMQProducerConfig.getListenerName());
}
client = builder.build();
} catch (PulsarClientException e) {
throw new RuntimeException(e);
@@ -127,6 +132,21 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
if (!StringUtils.isEmpty(adminServerUrl)) {
tmpProperties.setAdminServerUrl(adminServerUrl);
}
String listenerName = PropertiesUtils.getProperty(properties, PulsarMQConstants.PULSARMQ_LISTENER_NAME);
if (!StringUtils.isEmpty(listenerName)) {
tmpProperties.setListenerName(listenerName);
}
String enableChunkingStr = PropertiesUtils.getProperty(properties, PulsarMQConstants.PULSARMQ_ENABLE_CHUNKING);
if (!StringUtils.isEmpty(enableChunkingStr)) {
tmpProperties.setEnableChunking(Boolean.parseBoolean(enableChunkingStr));
}
String compressionType = PropertiesUtils.getProperty(properties, PulsarMQConstants.PULSARMQ_COMPRESSION_TYPE);
if (!StringUtils.isEmpty(compressionType)) {
tmpProperties.setCompressionType(compressionType);
}
if (logger.isDebugEnabled()) {
logger.debug("Load pulsar properties ==> {}", JSON.toJSON(this.mqProperties));
}
@@ -398,11 +418,34 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
}
// 创建指定topic的生产者
producer = client.newProducer()
.topic(fullTopic)
ProducerBuilder producerBuilder = client.newProducer();
if (pulsarMQProperties.getEnableChunking()) {
producerBuilder.enableChunking(true);
producerBuilder.enableBatching(false);
}
if (!StringUtils.isEmpty(pulsarMQProperties.getCompressionType())) {
switch (pulsarMQProperties.getCompressionType().toLowerCase()) {
case "lz4":
producerBuilder.compressionType(CompressionType.LZ4);
break;
case "zlib":
producerBuilder.compressionType(CompressionType.ZLIB);
break;
case "zstd":
producerBuilder.compressionType(CompressionType.ZSTD);
break;
case "snappy":
producerBuilder.compressionType(CompressionType.SNAPPY);
break;
}
}
producer = producerBuilder.topic(fullTopic)
// 指定路由器
.messageRouter(new MessageRouterImpl(topic))
.create();
// 放入缓存
PRODUCERS.put(topic, producer);
}

View File

@@ -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>

View File

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

View File

@@ -16,6 +16,7 @@ import org.slf4j.LoggerFactory;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONWriter;
import com.alibaba.otter.canal.common.CanalException;
import com.alibaba.otter.canal.common.utils.AddressUtils;
import com.alibaba.otter.canal.common.utils.ExecutorTemplate;
import com.alibaba.otter.canal.common.utils.PropertiesUtils;
import com.alibaba.otter.canal.connector.core.producer.AbstractMQProducer;
@@ -61,7 +62,7 @@ public class CanalRabbitMQProducer extends AbstractMQProducer implements CanalMQ
throw new CanalException("failed to parse host", ex);
}
} else if (servers.contains(":")) {
String[] serverHostAndPort = servers.split(":");
String[] serverHostAndPort = AddressUtils.splitIPAndPort(servers);
factory.setHost(serverHostAndPort[0]);
factory.setPort(Integer.parseInt(serverHostAndPort[1]));
} else {
@@ -81,13 +82,18 @@ public class CanalRabbitMQProducer extends AbstractMQProducer implements CanalMQ
try {
connect = factory.newConnection();
channel = connect.createChannel();
channel.queueDeclare(rabbitMQProperties.getQueue(), true, false, false, null);
channel.exchangeDeclare(rabbitMQProperties
.getExchange(), rabbitMQProperties.getDeliveryMode(), true, false, false, null);
channel.queueBind(rabbitMQProperties.getQueue(),
rabbitMQProperties.getExchange(),
rabbitMQProperties.getRoutingKey());
String queue = rabbitMQProperties.getQueue();
String exchange = rabbitMQProperties.getExchange();
String deliveryMode = rabbitMQProperties.getDeliveryMode();
String routingKey = rabbitMQProperties.getRoutingKey();
if (!StringUtils.isEmpty(queue)) {
channel.queueDeclare(queue, true, false, false, null);
}
if (!StringUtils.isEmpty(queue) && !StringUtils.isEmpty(exchange) && !StringUtils.isEmpty(deliveryMode)
&& !StringUtils.isEmpty(routingKey)) {
channel.exchangeDeclare(exchange, deliveryMode, true, false, false, null);
channel.queueBind(queue, exchange, routingKey);
}
} catch (IOException | TimeoutException ex) {
throw new CanalException("Start RabbitMQ producer error", ex);
}

View File

@@ -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>

View File

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

View File

@@ -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>

View File

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

View File

@@ -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>

View File

@@ -1,10 +1,8 @@
package com.taobao.tddl.dbsync.binlog;
import java.nio.charset.Charset;
import static com.taobao.tddl.dbsync.binlog.event.RowsLogBuffer.*;
import static com.taobao.tddl.dbsync.binlog.event.RowsLogBuffer.appendNumber2;
import static com.taobao.tddl.dbsync.binlog.event.RowsLogBuffer.appendNumber4;
import static com.taobao.tddl.dbsync.binlog.event.RowsLogBuffer.usecondsToStr;
import java.nio.charset.Charset;
/**
* 处理下MySQL json二进制转化为可读的字符串
@@ -34,26 +32,25 @@ public class JsonConversion {
public static final char JSONB_FALSE_LITERAL = '\2';
/*
* The size of offset or size fields in the small and the large storage
* format for JSON objects and JSON arrays.
* The size of offset or size fields in the small and the large storage format
* for JSON objects and JSON arrays.
*/
public static final int SMALL_OFFSET_SIZE = 2;
public static final int LARGE_OFFSET_SIZE = 4;
/*
* The size of key entries for objects when using the small storage format
* or the large storage format. In the small format it is 4 bytes (2 bytes
* for key length and 2 bytes for key offset). In the large format it is 6
* (2 bytes for length, 4 bytes for offset).
* The size of key entries for objects when using the small storage format or
* the large storage format. In the small format it is 4 bytes (2 bytes for key
* length and 2 bytes for key offset). In the large format it is 6 (2 bytes for
* length, 4 bytes for offset).
*/
public static final int KEY_ENTRY_SIZE_SMALL = (2 + SMALL_OFFSET_SIZE);
public static final int KEY_ENTRY_SIZE_LARGE = (2 + LARGE_OFFSET_SIZE);
/*
* The size of value entries for objects or arrays. When using the small
* storage format, the entry size is 3 (1 byte for type, 2 bytes for
* offset). When using the large storage format, it is 5 (1 byte for type, 4
* bytes for offset).
* The size of value entries for objects or arrays. When using the small storage
* format, the entry size is 3 (1 byte for type, 2 bytes for offset). When using
* the large storage format, it is 5 (1 byte for type, 4 bytes for offset).
*/
public static final int VALUE_ENTRY_SIZE_SMALL = (1 + SMALL_OFFSET_SIZE);
public static final int VALUE_ENTRY_SIZE_LARGE = (1 + LARGE_OFFSET_SIZE);
@@ -160,13 +157,12 @@ public class JsonConversion {
if (len < n + str_len) {
throw new IllegalArgumentException("illegal json data");
}
return new Json_Value(Json_enum_type.STRING, buffer.rewind()
.forward((int) n)
.getFixString((int) str_len, charset));
return new Json_Value(Json_enum_type.STRING,
buffer.rewind().forward((int) n).getFixString((int) str_len, charset));
case JSONB_TYPE_OPAQUE:
/*
* There should always be at least one byte, which tells the
* field type of the opaque value.
* There should always be at least one byte, which tells the field type of the
* opaque value.
*/
// The type is encoded as a uint8 that maps to an
// enum_field_types.
@@ -291,7 +287,7 @@ public class JsonConversion {
if (i > 0) {
buf.append(", ");
}
buf.append('"').append(key(i, charset)).append('"');
buf.append('"').append(escapse(key(i, charset))).append('"');
buf.append(": ");
element(i, charset).toJsonString(buf, charset);
}
@@ -368,45 +364,45 @@ public class JsonConversion {
buf.append('"').append(text).append('"');
} else if (m_field_type == LogEvent.MYSQL_TYPE_DATE || m_field_type == LogEvent.MYSQL_TYPE_DATETIME
|| m_field_type == LogEvent.MYSQL_TYPE_TIMESTAMP) {
long packed_value = m_data.getLong64();
if (packed_value == 0) {
text = "0000-00-00 00:00:00";
} else {
// 构造TimeStamp只处理到秒
long ultime = Math.abs(packed_value);
long intpart = ultime >> 24;
int frac = (int) (ultime % (1L << 24));
long ymd = intpart >> 17;
long ym = ymd >> 5;
long hms = intpart % (1 << 17);
// text =
// String.format("%04d-%02d-%02d %02d:%02d:%02d",
// (int) (ym / 13),
// (int) (ym % 13),
// (int) (ymd % (1 << 5)),
// (int) (hms >> 12),
// (int) ((hms >> 6) % (1 << 6)),
// (int) (hms % (1 << 6)));
StringBuilder builder = new StringBuilder(26);
appendNumber4(builder, (int) (ym / 13));
builder.append('-');
appendNumber2(builder, (int) (ym % 13));
builder.append('-');
appendNumber2(builder, (int) (ymd % (1 << 5)));
builder.append(' ');
appendNumber2(builder, (int) (hms >> 12));
builder.append(':');
appendNumber2(builder, (int) ((hms >> 6) % (1 << 6)));
builder.append(':');
appendNumber2(builder, (int) (hms % (1 << 6)));
builder.append('.').append(usecondsToStr(frac, 6));
text = builder.toString();
}
buf.append('"').append(text).append('"');
} else {
text = m_data.getFixString((int) m_length, charset);
buf.append('"').append(escapse(text)).append('"');
}
long packed_value = m_data.getLong64();
if (packed_value == 0) {
text = "0000-00-00 00:00:00";
} else {
// 构造TimeStamp只处理到秒
long ultime = Math.abs(packed_value);
long intpart = ultime >> 24;
int frac = (int) (ultime % (1L << 24));
long ymd = intpart >> 17;
long ym = ymd >> 5;
long hms = intpart % (1 << 17);
// text =
// String.format("%04d-%02d-%02d %02d:%02d:%02d",
// (int) (ym / 13),
// (int) (ym % 13),
// (int) (ymd % (1 << 5)),
// (int) (hms >> 12),
// (int) ((hms >> 6) % (1 << 6)),
// (int) (hms % (1 << 6)));
StringBuilder builder = new StringBuilder(26);
appendNumber4(builder, (int) (ym / 13));
builder.append('-');
appendNumber2(builder, (int) (ym % 13));
builder.append('-');
appendNumber2(builder, (int) (ymd % (1 << 5)));
builder.append(' ');
appendNumber2(builder, (int) (hms >> 12));
builder.append(':');
appendNumber2(builder, (int) ((hms >> 6) % (1 << 6)));
builder.append(':');
appendNumber2(builder, (int) (hms % (1 << 6)));
builder.append('.').append(usecondsToStr(frac, 6));
text = builder.toString();
}
buf.append('"').append(text).append('"');
} else {
text = m_data.getFixString((int) m_length, charset);
buf.append('"').append(escapse(text)).append('"');
}
break;
case STRING:
@@ -452,7 +448,8 @@ public class JsonConversion {
}
public static enum Json_enum_type {
OBJECT, ARRAY, STRING, INT, UINT, DOUBLE, LITERAL_NULL, LITERAL_TRUE, LITERAL_FALSE, OPAQUE, ERROR
OBJECT, ARRAY, STRING, INT, UINT, DOUBLE, LITERAL_NULL, LITERAL_TRUE,
LITERAL_FALSE, OPAQUE, ERROR
}
}

View File

@@ -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
@@ -819,7 +868,7 @@ public class QueryLogEvent extends LogEvent {
case Q_MASTER_DATA_WRITTEN_CODE:
return "Q_MASTER_DATA_WRITTEN_CODE";
case Q_INVOKER:
return "case Q_INVOKER";
return "Q_INVOKER";
case Q_MICROSECONDS:
return "Q_MICROSECONDS";
case Q_UPDATED_DB_NAMES:
@@ -834,9 +883,13 @@ public class QueryLogEvent extends LogEvent {
return "Q_SQL_REQUIRE_PRIMARY_KEY";
case Q_DEFAULT_TABLE_ENCRYPTION:
return "Q_DEFAULT_TABLE_ENCRYPTION";
case Q_DDL_SKIP_REWRITE:
case Q_OPT_FLASHBACK_AREA:
// or Q_DDL_SKIP_REWRITE
return "Q_DDL_SKIP_REWRITE";
case Q_OPT_INDEX_FORMAT_GPP_ENABLED:
return "Q_DDL_SKIP_REWRITE";
case Q_HRNOW:
// or Q_WSREP_SKIP_READONLY_CHECKS
return "Q_HRNOW";
case Q_XID:
return "Q_XID";

View File

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

View File

@@ -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>

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

Some files were not shown because too many files have changed in this diff Show More