谁知道Flink CDC中本地怎么debug 那些源码？

在Flink CDC中进行本地调试源码，可以通过以下步骤实现。这些步骤结合了阿里云实时计算 Flink版的相关文档和最佳实践。

1. 准备开发环境

在本地调试Flink CDC源码之前，需要确保开发环境已正确配置： - JDK版本：确保使用JDK 8或更高版本。 - Maven配置：在pom.xml文件中添加必要的依赖项，例如Flink核心库和CDC连接器的依赖。示例如下：

<properties>
    <maven.compiler.source>8</maven.compiler.source>
    <maven.compiler.target>8</maven.compiler.target>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <java.version>8</java.version>
    <flink.version>1.19.0</flink.version>
    <vvr.version>xxx</vvr.version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-core</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <!-- 添加Flink CDC连接器依赖 -->
    <dependency>
        <groupId>com.ververica</groupId>
        <artifactId>flink-connector-mysql-cdc</artifactId>
        <version>${vvr.version}</version>
    </dependency>
</dependencies>

• Flink版本一致性：确保本地开发使用的Flink版本与部署环境一致，避免因版本不匹配导致的问题。

2. 配置调试参数

在本地运行和调试包含Flink CDC连接器的作业时，可以使用以下方法： - 启用本地调试模式：通过IDE（如IntelliJ IDEA）直接运行Flink作业，并设置断点进行调试。 - 模拟数据源：如果无法直接连接到生产数据库，可以使用模拟数据生成连接器来生成测试数据。示例如下：

CREATE TABLE mock_source (
    id INT,
    name STRING,
    ts TIMESTAMP(3)
) WITH (
    'connector' = 'datagen',
    'rows-per-second' = '10'
);

• 指定调试数据：在Flink开发控制台中，可以选择使用线上数据或自定义的调试数据进行验证。

3. 调试MySQL CDC源码

针对MySQL CDC源表的调试，可以按照以下步骤操作： 1. 配置MySQL CDC源表： - 确保MySQL实例的binlog已启用，并且server-id范围满足并发需求。 - 示例配置如下：

CREATE TABLE mysql_source (
    order_id INT,
    order_status STRING,
    update_time TIMESTAMP(3)
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'localhost',
    'port' = '3306',
    'username' = 'flinkuser',
    'password' = 'flinkpw',
    'database-name' = 'mydb',
    'table-name' = 'orders'
);

1. 启动模式选择：

   • 使用scan.startup.mode参数控制启动模式。例如，跳过Snapshot阶段，直接从Binlog开始读取：

     'scan.startup.mode' = 'latest-offset'
     

2. 监控全量和增量阶段：

   • 全量阶段会通过SELECT语句读取数据，可能增加MySQL查询压力。
   • 增量阶段通过Binlog Client读取变更数据，需注意连接数限制。

3. 查看日志和指标：

   • 在本地调试时，可以通过日志确认是否已完成全量数据同步。例如，查找BinlogSplitReader is created日志。
   • 监控currentEmitEventTimeLag指标，判断是否进入增量阶段。

4. 解决常见问题

在本地调试过程中，可能会遇到以下问题及解决方案： - ClassNotFoundException异常： - 确保所有依赖均已正确加载。如果使用IDEA调试，检查是否存在MySqlSourceReaderMetrics类未找到的问题。 - 解决方案：将相关依赖打包为FAT JAR，或手动添加缺失的依赖。

• 带宽消耗过高：

  • Binlog是实例级别的，即使只读取特定表的数据，也会传输整个实例的变更记录。
  • 解决方案：通过Source复用减少带宽使用。

• 时区问题：

  • 如果增量阶段读取的timestamp字段时区相差8小时，检查server-time-zone参数是否与MySQL服务器时区一致。

5. 使用DataStream API调试

如果使用DataStream API构建MySQL CDC Source，可以参考以下配置： - tableList选项：确保表名包含数据库名，格式为yourDatabaseName.yourTableName。 - 自定义序列化器：在处理复杂数据类型时，使用自定义序列化器以确保数据正确解析。

6. 注意事项

• 调试功能限制：仅SQL作业支持作业调试功能，CTAS和CDAS语法不支持调试。
• Session集群要求：调试功能需要已创建Session集群。
• 并发和资源管理：在全量阶段使用并发读取历史数据，在增量阶段单并发读取以保证全局有序。

通过以上步骤，您可以在本地环境中高效地调试Flink CDC源码，并快速定位和解决问题。