Flink CDC里有大佬在用sqlserver cdc时候出现这个错误吗？

【问题描述】

在使用 Flink CDC 进行 SQL Server CDC（Change Data Capture，变更数据捕获）操作时，出现了错误，本文档将详细分析该问题，并提供可能的解决方案。

【环境配置】

软件名称	版本号
Flink	1.13.2
SQL Server	2019
JDBC 驱动	8.4.1.jre8

【问题现象】

在进行 SQL Server CDC 操作时，遇到以下错误：

Exception in thread "main" org.apache.flink.table.api.TableException: Unsupported change mode for SQL Server binlog connector.

【原因分析】

根据错误信息，问题出在 SQL Server CDC 的变更模式上，Flink CDC 对 SQL Server CDC 支持的变更模式有限制，不支持某些特定的变更模式。

【解决方案】

1、检查 SQL Server CDC 的配置，确保变更模式是 Flink CDC 支持的类型，目前 Flink CDC 支持的 SQL Server CDC 变更模式包括：row_based 和 batch_based。

2、如果需要使用其他变更模式，可以考虑升级 Flink 版本或寻找其他替代方案。

【示例代码】

以下是一个简单的 Flink SQL 示例，用于从 SQL Server 中读取 CDC 数据：

CREATE TABLE source (
    id INT,
    name STRING,
    age INT,
    address STRING,
    update_timestamp TIMESTAMP(3)
) WITH (
    'connector' = 'sqlservercdc',
    'hostname' = 'localhost',
    'port' = '1433',
    'username' = 'sa',
    'password' = 'your_password',
    'databasename' = 'your_database',
    'tablename' = 'your_table',
    'scan.startup.mode' = 'latestoffset',
    'debezium.sqlserver.instance' = 'your_instance_name',
    'debezium.sqlserver.user' = 'your_user',
    'debezium.sqlserver.password' = 'your_password',
    'debezium.sqlserver.database.hostname' = 'your_hostname',
    'debezium.sqlserver.database.port' = 'your_port',
    'debezium.sqlserver.database.name' = 'your_database_name',
    'debezium.sqlserver.database.user' = 'your_user',
    'debezium.sqlserver.database.password' = 'your_password',
    'debezium.sqlserver.database.history' = 'io.debezium.relational.history.FileDatabaseHistory',
    'debezium.sqlserver.database.history.file.location' = '/path/to/dbhistory.dat',
    'debezium.sqlserver.database.history.kafka.bootstrap.servers' = 'localhost:9092',
    'debezium.sqlserver.database.history.kafka.topic' = 'dbhistory.your_database_name',
    'format' = 'json'
);

请根据实际情况修改上述代码中的参数，并确保变更模式为 row_based 或 batch_based。