跳到主要内容
版本:Next

MySQL CDC

MySQL CDC source 连接器

支持这些引擎

SeaTunnel Zeta
Flink

描述

MySQL CDC连接器允许从MySQL数据库读取快照和增量数据. 本文档描述了如何配置MySQL CDC连接器以对MySQL数据库运行SQL查询.

主要功能

支持的数据源信息

数据源支持的版本DriverUrlMaven
MySQL
  • MySQL: 5.5, 5.6, 5.7, 8.0.x
  • RDS MySQL: 5.6, 5.7, 8.0.x
  • com.mysql.cj.jdbc.Driverjdbc:mysql://localhost:3306/testhttps://mvnrepository.com/artifact/mysql/mysql-connector-java/8.0.28

    依赖

    安装Jdbc驱动

    对于Flink引擎

    1. 你需要确保 jdbc 驱动 jar package 已经放在目录 ${SEATUNNEL_HOME}/plugins/.

    对于SeaTunnel Zeta引擎

    1. 你需要确保 jdbc 驱动 jar package 已经放在目录 ${SEATUNNEL_HOME}/lib/.

    创建MySQL用户

    你必须定义一个MySQL用户,该用户对Debezium MySQL连接器所监控的所有数据库拥有适当的权限.

    1. 创建MySQL用户:
    mysql> CREATE USER 'user'@'localhost' IDENTIFIED BY 'password';
    1. 给用户赋予所需权限:
    mysql> GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'user' IDENTIFIED BY 'password';
    1. 最终确定用户权限:
    mysql> FLUSH PRIVILEGES;

    启用MySQL Binlog

    一定要为MySQL复制启用binlog。binlog记录事务更新以供复制工具传播更改.

    1. 检查log-bin是否已经设置为on:
    mysql> show variables where variable_name in ('log_bin', 'binlog_format', 'binlog_row_image', 'gtid_mode', 'enforce_gtid_consistency');
    +--------------------------+----------------+
    | Variable_name | Value |
    +--------------------------+----------------+
    | binlog_format | ROW |
    | binlog_row_image | FULL |
    | enforce_gtid_consistency | ON |
    | gtid_mode | ON |
    | log_bin | ON |
    +--------------------------+----------------+
    1. 如果log_bin的值不是on, 配置你的MySQL server配置文件($MYSQL_HOME/mysql.cnf),配置文件中包含以下属性,这些属性在以下表格中有描述:
    # Enable binary replication log and set the prefix, expiration, and log format.
    # The prefix is arbitrary, expiration can be short for integration tests but would
    # be longer on a production system. Row-level info is required for ingest to work.
    # Server ID is required, but this will vary on production systems
    server-id = 223344
    log_bin = mysql-bin
    expire_logs_days = 10
    binlog_format = row
    # mysql 5.6+ requires binlog_row_image to be set to FULL
    binlog_row_image = FULL

    # optional enable gtid mode
    # mysql 5.6+ requires gtid_mode to be set to ON, but not required by mysql 8.0+
    gtid_mode = on
    enforce_gtid_consistency = on
    1. 重启MySQL Server
    /etc/inint.d/mysqld restart
    1. 修改之后再检查一次binlog的状态:

    MySQL 5.5:

    mysql> show variables where variable_name in ('log_bin', 'binlog_format', 'binlog_row_image', 'gtid_mode', 'enforce_gtid_consistency');
    +--------------------------+----------------+
    | Variable_name | Value |
    +--------------------------+----------------+
    | binlog_format | ROW |
    | log_bin | ON |
    +--------------------------+----------------+

    MySQL 5.6+:

    mysql> show variables where variable_name in ('log_bin', 'binlog_format', 'binlog_row_image', 'gtid_mode', 'enforce_gtid_consistency');
    +--------------------------+----------------+
    | Variable_name | Value |
    +--------------------------+----------------+
    | binlog_format | ROW |
    | binlog_row_image | FULL |
    | enforce_gtid_consistency | ON |
    | gtid_mode | ON |
    | log_bin | ON |
    +--------------------------+----------------+

    MySQL 8.0+:

    show variables where variable_name in ('log_bin', 'binlog_format', 'binlog_row_image', 'gtid_mode', 'enforce_gtid_consistency')
    +--------------------------+----------------+
    | Variable_name | Value |
    +--------------------------+----------------+
    | binlog_format | ROW |
    | binlog_row_image | FULL |
    | enforce_gtid_consistency | OFF |
    | gtid_mode | OFF |
    | log_bin | ON |
    +--------------------------+----------------+

    提示

    配置MySQL session超时时长

    当为大型数据库初始一致快照时,已建立的连接可能在读取表时超时。可以通过在MySQL配置文件中配置interactive_timeout(交互超时时间)和wait_timeout(等待超时时间)来防止这种行为.

    • interactive_timeout: 服务器在关闭交互连接之前等待活动(交互操作)的秒数. 详见 MySQL’s documentation.
    • wait_timeout: 服务器在关闭非交互式连接之前等待其活动的秒数. 详见 MySQL’s documentation.

    更多的数据库配置,见 Debezium MySQL Connector

    数据类型映射

    Mysql数据类型SeaTunnel数据类型
    BIT(1)
    TINYINT(1)
    BOOLEAN
    TINYINTTINYINT
    TINYINT UNSIGNED
    SMALLINT
    SMALLINT
    SMALLINT UNSIGNED
    MEDIUMINT
    MEDIUMINT UNSIGNED
    INT
    INTEGER
    YEAR
    INT
    INT UNSIGNED
    INTEGER UNSIGNED
    BIGINT
    BIGINT
    BIGINT UNSIGNEDDECIMAL(20,0)
    DECIMAL(p, s)
    DECIMAL(p, s) UNSIGNED
    NUMERIC(p, s)
    NUMERIC(p, s) UNSIGNED
    DECIMAL(p,s)
    FLOAT
    FLOAT UNSIGNED
    FLOAT
    DOUBLE
    DOUBLE UNSIGNED
    REAL
    REAL UNSIGNED
    DOUBLE
    CHAR
    VARCHAR
    TINYTEXT
    MEDIUMTEXT
    TEXT
    LONGTEXT
    ENUM
    JSON
    ENUM
    STRING
    DATEDATE
    TIME(s)TIME(s)
    DATETIME
    TIMESTAMP(s)
    TIMESTAMP(s)
    BINARY
    VARBINAR
    BIT(p)
    TINYBLOB
    MEDIUMBLOB
    BLOB
    LONGBLOB
    GEOMETRY
    BYTES

    配置参数选项

    参数名称类型是否必须默认值描述
    base-urlString-JDBC连接的URL. 例如: jdbc:mysql://localhost:3306/test.
    usernameString-用来连接到数据库服务的数据库名称.
    passwordString-连接到数据库服务所使用的密码.
    database-namesList-要监控的数据库名称.
    database-patternString.*要捕获的数据库名称的正则表达式, 例如: database_prefix.*.
    table-namesList-要监控的表名. 表名需要包括库名, 例如: database_name.table_name
    table-patternString-要捕获的表名称的正则表达式. 表名需要包括库名, 例如: database.*\\.table_.*
    table-names-configList-表配置的列表集合. 例如: [{"table": "db1.schema1.table1","primaryKeys": ["key1"],"snapshotSplitColumn": "key2"}]
    startup.modeEnumINITIALMySQL CDC 消费者的可选启动模式, 有效枚举值为 initial, earliest, latestspecific.
    initial: 启动时同步历史数据, 然后同步增量数据.
    earliest: 从尽可能最早的偏移量开始启动.
    latest: 从最近的偏移量启动.
    specific: 从用户提供的特定偏移量开始启动.
    startup.specific-offset.fileString-从指定的binlog日志文件名开始. 注意, 当使用 startup.mode 选项为 specific 时,此选项为必填项.
    startup.specific-offset.posLong-从指定的binlog日志文件位置开始. 注意, 当使用 startup.mode 选项为 specific 时,此选项为必填项.
    stop.modeEnumNEVERMySQL CDC 消费者的可选停止模式, 有效枚举值为 never, latestspecific.
    never: 实时任务一直运行不停止.
    latest: 从最新的偏移量处停止.
    specific: 从用户提供的特定偏移量处停止.
    stop.specific-offset.fileString-从指定的binlog日志文件名停止. 注意, 当使用 stop.mode 选项为 specific 时,此选项为必填项.
    stop.specific-offset.posLong-从指定的binlog日志文件位置停止. 注意, 当使用 stop.mode 选项为 specific 时,此选项为必填项.
    snapshot.split.sizeInteger8096表快照的分割大小(行数),读取表的快照时,被捕获的表会被分割成多个分割块.
    snapshot.fetch.sizeInteger1024每次轮询读取表快照时的最大获取大小.
    server-idString-此数据库客户端的数字 ID 或数字 ID 范围, 数字 ID 的语法如 5400, 数字 ID 范围的语法如 '5400-5408'.
    每个 ID 在 MySQL 集群中所有当前正在运行的数据库进程里必须是唯一的. 此连接加入
    MySQL服务以另外一个服务的身份 (带有此唯一 ID) 以便于能够读取binlog.
    默认情况下, 会生成一个介于 6500 到 2,148,492,146 之间的数字, 然而我们建议设置一个明确的值.
    server-time-zoneStringUTC数据库服务中的会话时区. 如果没设置, 使用 ZoneId.systemDefault() 来确定服务的时区.
    connect.timeout.msDuration30000连接器在尝试连接数据库服务器后,在超时之前应等待的最长时间.
    connect.max-retriesInteger3连接器在构建数据库服务器连接时应重试的最大重试次数.
    connection.pool.sizeInteger20jdbc连接池大小.
    chunk-key.even-distribution.factor.upper-boundDouble100块键分布因子的上限. 该因子用于确定表数据是否分布均匀. 如果分布式因子计算结果小于或等于此上限 (即., (MAX(id) - MIN(id) + 1) / row count), 表的分块将被优化以实现均匀分布. 否则, 如果分布因子大于此上限, 该表将被视为分布不均, 并且如果估计的分片数量超过 sample-sharding.threshold 所指定的值, 则将使用基于采样的分片策略. 默认值是100.0.
    chunk-key.even-distribution.factor.lower-boundDouble0.05块键分布因子的下限. 该因子用于确定表数据是否分布均匀. 如果计算得出的分布因子大于或等于此下限 (即., (MAX(id) - MIN(id) + 1) / row count), 表的分块将被优化以实现均匀分布. 否则, 如果分布因子小于此下限, 该表将被视为分布不均, 并且如果预估的分片数量超过了 sample-sharding.threshold 所指定的值,则将使用基于采样的分片策略. 默认值是 0.05.
    sample-sharding.thresholdInteger1000此配置指定了触发采样分片策略的预估分片数量阈值. 当分配因子超出由 chunk-key.even-distribution.factor.upper-boundchunk-key.even-distribution.factor.lower-bound 所指定的范围时, 如果估计的分片数量 (按近似行数/块大小 计算) 超过此阈值, 则将使用样本分片策略. 这有助于更高效地处理大型数据集. 默认值为 1000 分片.
    inverse-sampling.rateInteger1000采样分片策略中使用的采样率的倒数. 例如, 如果该值设置为 1000, 则表示在采样过程中应用了 1/1000 的采样率. 此选项在控制采样的粒度方面提供了灵活性, 从而影响最终的分片数量. 在处理非常大的数据集时非常有用, 因为此时更倾向于使用较低的采样率. 默认值为 1000.
    exactly_onceBooleanfalse启用精确一次语义.
    formatEnumDEFAULTMySQL CDC 的可选输出格式, 有效的枚举值为 DEFAULTCOMPATIBLE_DEBEZIUM_JSON.
    schema-changes.enabledBooleanfalse模式演进默认是禁用的. 当前我们只支持 add columndrop columnrename columnmodify column.
    debeziumConfig-传递 Debezium的属性 给Debezium嵌入式引擎, 该引擎用于捕获 MySQL 服务的数据变更.
    common-options-Source插件通用参数, 详见 Source Common Options

    任务示例

    简单的示例

    支持多表读取

    env {
    parallelism = 1
    job.mode = "STREAMING"
    checkpoint.interval = 10000
    }

    source {
    MySQL-CDC {
    base-url = "jdbc:mysql://localhost:3306/testdb"
    username = "root"
    password = "root@123"
    table-names = ["testdb.table1", "testdb.table2"]

    startup.mode = "initial"
    }
    }

    sink {
    Console {
    }
    }

    支持向Kafka发送与Debezium兼容的格式

    一定是使用kafka作为sink, 详见 compatible debezium format

    支持表的自定义主键

    env {
    parallelism = 1
    job.mode = "STREAMING"
    checkpoint.interval = 10000
    }

    source {
    MySQL-CDC {
    base-url = "jdbc:mysql://localhost:3306/testdb"
    username = "root"
    password = "root@123"

    table-names = ["testdb.table1", "testdb.table2"]
    table-names-config = [
    {
    table = "testdb.table2"
    primaryKeys = ["id"]
    }
    ]
    }
    }

    sink {
    Console {
    }
    }

    支持模式演变(表结构变更)

    env {
    # You can set engine configuration here
    parallelism = 5
    job.mode = "STREAMING"
    checkpoint.interval = 5000
    read_limit.bytes_per_second=7000000
    read_limit.rows_per_second=400
    }

    source {
    MySQL-CDC {
    server-id = 5652-5657
    username = "st_user_source"
    password = "mysqlpw"
    table-names = ["shop.products"]
    base-url = "jdbc:mysql://mysql_cdc_e2e:3306/shop"

    schema-changes.enabled = true
    }
    }

    sink {
    jdbc {
    url = "jdbc:mysql://mysql_cdc_e2e:3306/shop"
    driver = "com.mysql.cj.jdbc.Driver"
    user = "st_user_sink"
    password = "mysqlpw"
    generate_sink_sql = true
    database = shop
    table = mysql_cdc_e2e_sink_table_with_schema_change_exactly_once
    primary_keys = ["id"]
    is_exactly_once = true
    xa_data_source_class_name = "com.mysql.cj.jdbc.MysqlXADataSource"
    }
    }

    表名支持正则以读取多个表

    table-patterntable-names 只能选择一个

    env {
    # You can set engine configuration here
    parallelism = 1
    job.mode = "STREAMING"
    checkpoint.interval = 5000
    read_limit.bytes_per_second=7000000
    read_limit.rows_per_second=400
    }

    source {
    MySQL-CDC {
    server-id = 5652
    username = "st_user_source"
    password = "mysqlpw"
    database-pattern = "source.*"
    table-pattern = "source.*\\..*"
    base-url = "jdbc:mysql://mysql_cdc_e2e:3306"
    }
    }

    sink {
    Console {
    }
    }

    更新日志

    Change Log
    ChangeCommitVersion
    [Improve][CDC] Filter ddl for snapshot phase (#8911)https://github.com/apache/seatunnel/commit/641cc72f2f2.3.10
    [Improve][CDC] Extract duplicate code (#8906)https://github.com/apache/seatunnel/commit/b922bb90e62.3.10
    [Improve] restruct connector common options (#8634)https://github.com/apache/seatunnel/commit/f3499a6eeb2.3.10
    [Fix][mysql-cdc] Fix GTIDs on startup to correctly recover from checkpoint (#8528)https://github.com/apache/seatunnel/commit/82e4096c082.3.10
    [Feature][MySQL-CDC] Support database/table wildcards scan read (#8323)https://github.com/apache/seatunnel/commit/2116843ce82.3.9
    [Feature][Jdbc] Support sink ddl for postgresql (#8276)https://github.com/apache/seatunnel/commit/353bbd21a12.3.9
    [Feature][CDC] Add 'schema-changes.enabled' options (#8285)https://github.com/apache/seatunnel/commit/8e29ecf54f2.3.9
    Revert "[Feature][Redis] Flush data when the time reaches checkpoint interval" and "[Feature][CDC] Add 'schema-changes.enabled' options" (#8278)https://github.com/apache/seatunnel/commit/fcb29382862.3.9
    [Feature][CDC] Add 'schema-changes.enabled' options (#8252)https://github.com/apache/seatunnel/commit/d783f9447c2.3.9
    [Improve][dist]add shade check rule (#8136)https://github.com/apache/seatunnel/commit/51ef8000162.3.9
    [Feature][Connector-V2]Jdbc chunk split add snapshotSplitColumn config #7794 (#7840)https://github.com/apache/seatunnel/commit/b6c6dc04382.3.9
    [Feature][Core] Support cdc task ddl restore for zeta (#7463)https://github.com/apache/seatunnel/commit/8e322281ed2.3.9
    [Feature][Connector-v2] Support schema evolution for Oracle connector (#7908)https://github.com/apache/seatunnel/commit/79406bcc2f2.3.9
    [Hotfix][CDC] Fix ddl duplicate execution error when config multi_table_sink_replica (#7634)https://github.com/apache/seatunnel/commit/23ab3edbbb2.3.8
    [Hotfix][CDC] Fix package name spelling mistake (#7415)https://github.com/apache/seatunnel/commit/469112fa642.3.8
    [Hotfix][MySQL-CDC] Fix ArrayIndexOutOfBoundsException in mysql binlog read (#7381)https://github.com/apache/seatunnel/commit/40c5f313eb2.3.7
    [Improve][Connector-V2] Support schema evolution for mysql-cdc and mysql-jdbc (#6929)https://github.com/apache/seatunnel/commit/cf91e51fc72.3.6
    [Hotfix][MySQL-CDC] Fix read gbk varchar chinese garbled characters (#7046)https://github.com/apache/seatunnel/commit/4e4d2b8ee52.3.6
    [Improve][CDC] Bump the version of debezium to 1.9.8.Final (#6740)https://github.com/apache/seatunnel/commit/c3ac9535242.3.6
    [Improve][CDC] Close idle subtasks gorup(reader/writer) in increment phase (#6526)https://github.com/apache/seatunnel/commit/454c339b9c2.3.6
    [Improve][JDBC Source] Fix Split can not be cancel (#6825)https://github.com/apache/seatunnel/commit/ee3b7c37232.3.6
    [Hotfix][Jdbc/CDC] Fix postgresql uuid type in jdbc read (#6684)https://github.com/apache/seatunnel/commit/868ba4d7c72.3.6
    [Improve][mysql-cdc] Support mysql 5.5 versions (#6710)https://github.com/apache/seatunnel/commit/058f5594a32.3.6
    [Improve][mysql-cdc] Fallback to desc table when show create table failed (#6701)https://github.com/apache/seatunnel/commit/6f74663c082.3.6
    [Improve][Jdbc] Add quote identifier for sql (#6669)https://github.com/apache/seatunnel/commit/849d748d3d2.3.5
    [Fix][Connector-V2] Fix connector support SPI but without no args constructor (#6551)https://github.com/apache/seatunnel/commit/5f3c9c36a52.3.5
    [Improve][CDC-Connector]Fix CDC option rule. (#6454)https://github.com/apache/seatunnel/commit/1ea27afa872.3.5
    [Improve][CDC] Optimize memory allocation for snapshot split reading (#6281)https://github.com/apache/seatunnel/commit/48566458372.3.5
    [Improve][API] Unify type system api(data & type) (#5872)https://github.com/apache/seatunnel/commit/b38c7edcc92.3.5
    [Feature][CDC] Support custom table primary key (#6106)https://github.com/apache/seatunnel/commit/1312a1dd272.3.4
    [Feature][CDC] Support read no primary key table (#6098)https://github.com/apache/seatunnel/commit/b42d78de3f2.3.4
    [Bug][CDC] Fix state recovery error when switching a single table to multiple tables (#5784)https://github.com/apache/seatunnel/commit/37fcff347e2.3.4
    [Feature][formats][ogg] Support read ogg format message #4201 (#4225)https://github.com/apache/seatunnel/commit/7728e241e82.3.4
    [Improve][CDC] Clean unused code (#5785)https://github.com/apache/seatunnel/commit/b5a66d3dbe2.3.4
    [Improve][Jdbc] Fix database identifier (#5756)https://github.com/apache/seatunnel/commit/dbfc8a670a2.3.4
    [improve][mysql-cdc] Optimize the default value range of mysql server-id to reduce conflicts. (#5550)https://github.com/apache/seatunnel/commit/51746394632.3.4
    [Improve] Remove catalog tag for config file (#5645)https://github.com/apache/seatunnel/commit/dc509aa0802.3.4
    [Improve][Pom] Add junit4 to the root pom (#5611)https://github.com/apache/seatunnel/commit/7b4f7db2a22.3.4
    [Improve] Refactor CatalogTable and add SeaTunnelSource::getProducedCatalogTables (#5562)https://github.com/apache/seatunnel/commit/41173357f82.3.4
    [Improve][connector-cdc-mysql] avoid listing tables under unnecessary databases (#5365)https://github.com/apache/seatunnel/commit/3e5d018b352.3.4
    [Improve][Docs] Refactor MySQL-CDC docs (#5302)https://github.com/apache/seatunnel/commit/74530a04612.3.4
    [Improve][CheckStyle] Remove useless 'SuppressWarnings' annotation of checkstyle. (#5260)https://github.com/apache/seatunnel/commit/51c0d709ba2.3.4
    [Hotfix] Fix com.google.common.base.Preconditions to seatunnel shade one (#5284)https://github.com/apache/seatunnel/commit/ed5eadcf732.3.3
    [Imporve][CDC Base] Add a fast sampling method that supports character types (#5179)https://github.com/apache/seatunnel/commit/c0422dbfeb2.3.3
    [improve][CDC Base] Add some split parameters to the optionRule (#5161)https://github.com/apache/seatunnel/commit/94fd6755e62.3.3
    [Improve][CDC] support exactly-once of cdc and fix the BinlogOffset comparing bug (#5057)https://github.com/apache/seatunnel/commit/0e4190ab2e2.3.3
    [Feature][Connector-V2][CDC] Support string type shard fields. (#5147)https://github.com/apache/seatunnel/commit/e1be9d7f8a2.3.3
    [Feature][CDC] Support tables without primary keys (with unique keys) (#163) (#5150)https://github.com/apache/seatunnel/commit/32b7f2b6902.3.3
    [Feature][Connector-V2][mysql cdc] Conversion of tinyint(1) to bool is supported (#5105)https://github.com/apache/seatunnel/commit/86b1b7e31a2.3.3
    [Feature][connector-v2][mongodbcdc]Support source mongodb cdc (#4923)https://github.com/apache/seatunnel/commit/d729fcba4c2.3.3
    [Bugfix][connector-cdc-mysql] Fix listener not released when BinlogClient reuse (#5011)https://github.com/apache/seatunnel/commit/3287b1d8522.3.3
    [BugFix][Connector-V2] [MySQL-CDC] serverId from int to long (#5033) (#5035)https://github.com/apache/seatunnel/commit/4abc80e1112.3.3
    [Hotfix][CDC] Fix jdbc connection leak for mysql (#5037)https://github.com/apache/seatunnel/commit/738925ba102.3.3
    [Feature][CDC] Support disable/enable exactly once for INITIAL (#4921)https://github.com/apache/seatunnel/commit/6d9a3e59572.3.3
    [Improve][CDC]change driver scope to provider (#5002)https://github.com/apache/seatunnel/commit/745c0b9e922.3.3
    [Improve][CDC]Remove driver for cdc connector (#4952)https://github.com/apache/seatunnel/commit/b65f40c3c92.3.3
    [improve][CDC base] Implement Sample-based Sharding Strategy with Configurable Sampling Rate (#4856)https://github.com/apache/seatunnel/commit/d827c700f02.3.2
    [Hotfix][CDC] Fix chunk start/end parameter type error (#4777)https://github.com/apache/seatunnel/commit/c13c0319952.3.2
    [feature][catalog] Support for multiplexing connections (#4550)https://github.com/apache/seatunnel/commit/41277d7f782.3.2
    [BugFix][Mysql-CDC] Fix Time data type is empty when reading from MySQL CDC (#4670)https://github.com/apache/seatunnel/commit/e4f973daf72.3.2
    [Improve][CDC] Optimize jdbc fetch-size options (#4352)https://github.com/apache/seatunnel/commit/fbb60ce1be2.3.1
    [Improve][CDC] Improve startup.mode/stop.mode options (#4360)https://github.com/apache/seatunnel/commit/b71d8739d52.3.1
    Update CDC StartupMode and StopMode option to SingleChoiceOption (#4357)https://github.com/apache/seatunnel/commit/f60ac1a5e92.3.1
    [bugfix][cdc-base] Fix cdc base shutdown thread not cleared (#4327)https://github.com/apache/seatunnel/commit/ac61409bd82.3.1
    [Feature][CDC] Support export debezium-json format to kafka (#4339)https://github.com/apache/seatunnel/commit/5817ec07bf2.3.1
    [Improve][CDC][MySQL] Ennable binlog watermark compare (#4293)https://github.com/apache/seatunnel/commit/b22fb259c82.3.1
    [Feature][CDC][Mysql] Support read database list (#4255)https://github.com/apache/seatunnel/commit/3ca60c6fed2.3.1
    Add redshift datatype convertor (#4245)https://github.com/apache/seatunnel/commit/b19011517f2.3.1
    [improve][zeta] fix zeta bugshttps://github.com/apache/seatunnel/commit/3a82e8b39f2.3.1
    [Improve] Support MySqlCatalog Use JDBC URL With Custom Suffixhttps://github.com/apache/seatunnel/commit/210d0ff1f82.3.1
    [chore] Code format with spotless plugin.https://github.com/apache/seatunnel/commit/291214ad6f2.3.1
    Merge branch 'dev' into merge/cdchttps://github.com/apache/seatunnel/commit/4324ee19122.3.1
    [Improve][Project] Code format with spotless plugin.https://github.com/apache/seatunnel/commit/423b5830382.3.1
    [improve][jdbc] Reduce jdbc options configuration (#4218)https://github.com/apache/seatunnel/commit/ddd8f808b52.3.1
    [improve][cdc] support sharding-tables (#4207)https://github.com/apache/seatunnel/commit/5c3f0c9b002.3.1
    [Hotfix][CDC] Fix multiple-table data read (#4200)https://github.com/apache/seatunnel/commit/7f5671d2ce2.3.1
    [Feature][Zeta] Support shuffle multiple rows by tableId (#4147)https://github.com/apache/seatunnel/commit/8348f1a1082.3.1
    [Improve][build] Give the maven module a human readable name (#4114)https://github.com/apache/seatunnel/commit/d7cd6010512.3.1
    [Feature][CDC] Support batch processing on multiple-table shuffle flow (#4116)https://github.com/apache/seatunnel/commit/919653d83e2.3.1
    [Improve][Project] Code format with spotless plugin. (#4101)https://github.com/apache/seatunnel/commit/a2ab1665612.3.1
    [Feature][CDC] MySQL CDC supports deserialization of multi-tables (#4067)https://github.com/apache/seatunnel/commit/21ef45fcca2.3.1
    fix cdc option rule error (#4018)https://github.com/apache/seatunnel/commit/ea160429df2.3.1
    [Improve][CDC][base] Guaranteed to be exactly-once in the process of switching from SnapshotTask to IncrementalTask (#3837)https://github.com/apache/seatunnel/commit/8379aaf8762.3.1
    [Feature][Connector] add get source method to all source connector (#3846)https://github.com/apache/seatunnel/commit/417178fb842.3.1
    [Feature][API & Connector & Doc] add parallelism and column projection interface (#3829)https://github.com/apache/seatunnel/commit/b9164b8ba12.3.1
    [Improve][CDC] Add mysql-cdc source factory (#3791)https://github.com/apache/seatunnel/commit/356538de8a2.3.1
    [feature][connector-v2] add sqlServer CDC (#3686)https://github.com/apache/seatunnel/commit/0f0afb58af2.3.0
    [feature][e2e][cdc] add mysql cdc container (#3667)https://github.com/apache/seatunnel/commit/7696ba15512.3.0
    [feature][cdc] Fixed error in mysql cdc under real-time job (#3666)https://github.com/apache/seatunnel/commit/2238fda3002.3.0
    [feature][connector][cdc] add SeaTunnelRowDebeziumDeserializeSchema (#3499)https://github.com/apache/seatunnel/commit/ff44db116e2.3.0
    [feature][connector][mysql-cdc] add MySQL CDC enumerator (#3481)https://github.com/apache/seatunnel/commit/ff4b32dc282.3.0
    [bugfix][connector-v2] fix cdc mysql reader err (#3465)https://github.com/apache/seatunnel/commit/1b406b5a312.3.0
    [feature][connector] add mysql cdc reader (#3455)https://github.com/apache/seatunnel/commit/ae981df6752.3.0