1. 问题起源：当MySQL-CDC遇上Oracle-CDC

最近在帮公司做数据中台升级，需要把几个核心的Oracle业务库和MySQL日志库的数据实时同步到数据湖里。我们用的是阿里云Flink全托管服务，想着用Flink CDC来做应该挺省事的。结果刚上手就踩了个大坑：Flink内置的MySQL-CDC连接器，和从社区下载的Oracle-CDC连接器，版本死活对不上，作业一启动就报类冲突。

具体报错信息大概是这样的：

java.lang.NoSuchMethodError: io.debezium.connector.base.ChangeEventQueue.<init>(...)

或者

java.lang.ClassNotFoundException: com.ververica.cdc.debezium.internal.DebeziumChangeConsumer

折腾了半天，终于搞明白了问题的根因。Flink CDC连接器底层都依赖一个叫 Debezium 的组件来做数据库的变更数据捕获。你可以把Debezium想象成一个“数据库监听器”，它负责解析MySQL的binlog或者Oracle的redo log。问题就在于，不同版本的Flink CDC连接器，可能依赖了不同版本、且互不兼容的Debezium库。

比如，阿里云Flink全托管环境内置的MySQL-CDC连接器，可能打包的是Debezium 1.5.4。而我们从Flink CDC官网下载的、支持Oracle的flink-sql-connector-oracle-cdc-2.2.1.jar，里面打包的可能是Debezium 1.6.0。这两个版本的类和方法可能有差异，当它们在同一个Flink作业的ClassLoader里相遇时，打架就在所难免了。

这就像你同时安装了Python 3.7和3.8的某些库，它们名字一样但内部实现不同，程序运行起来肯定要出错。更麻烦的是，阿里云Flink内置的连接器版本我们是改不了的，这属于平台的黑盒。而Oracle-CDC又要求Flink CDC版本至少在2.1以上才能有比较好的功能和稳定性。

所以，唯一的出路就是：我们自己动手，打造一个与Oracle-CDC版本匹配的、自定义的MySQL-CDC连接器，然后上传到阿里云Flink平台使用。听起来有点绕，但实操下来，其实就是个“偷梁换柱”的活儿。

2. 实战第一步：获取并统一连接器版本

既然要解决兼容性问题，第一步就是确保我们使用的两个CDC连接器核心版本一致。这里我强烈推荐使用 Flink CDC 2.2.1 版本，这个版本相对稳定，社区问题也少。

2.1 下载官方连接器JAR包

我们需要去Maven中央仓库下载两个关键的“原材料”：

1. Oracle CDC连接器：这是我们必须从社区引入的，因为阿里云Flink全托管目前没有内置。
2. MySQL CDC连接器：我们将以这个官方JAR包为蓝本进行改造。

你可以直接在浏览器打开下面的链接，或者用wget命令下载：

# Oracle CDC 连接器 (2.2.1版本)
wget https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-oracle-cdc/2.2.1/flink-sql-connector-oracle-cdc-2.2.1.jar

# MySQL CDC 连接器 (2.2.1版本) - 这是我们的改造模板
wget https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-mysql-cdc/2.2.1/flink-sql-connector-mysql-cdc-2.2.1.jar

下载好后，先把oracle-cdc的jar包放到一边。我们当前的重点是处理MySQL这个jar包。你可以先用压缩软件打开这个flink-sql-connector-mysql-cdc-2.2.1.jar看看，里面是一个完整的、包含所有依赖的“Uber Jar”。我们的目标就是修改其中最关键的一个类，然后重新打包。

2.2 理解连接器注册机制：为什么改个名就行？

这里涉及一个Flink SQL连接器的核心机制：连接器工厂（Connector Factory）。当我们在CREATE TABLE语句中写下 'connector' = 'mysql-cdc' 时，Flink的框架会去扫描所有jar包，寻找一个能识别mysql-cdc这个标识符的工厂类。这个工厂类负责创建真正的数据源。

在MySQL CDC连接器的jar包里，这个工厂类通常是 com.ververica.cdc.connectors.mysql.table.MySqlTableSourceFactory。它里面有一个方法叫 factoryIdentifier()，返回值就是字符串 "mysql-cdc"。我们的计划很简单：把这个返回值改成别的，比如 mysql-test-cdc。这样，我们自定义的连接器就不会和平台内置的、标识符为mysql-cdc的连接器冲突了。它们虽然功能完全一样，但在Flink看来，这是两个不同的插件。

这招“改名大法”是解决阿里云这类托管平台连接器冲突的经典思路。平台内置的我们动不了，那就自己造一个“马甲”，让我们的作业代码去调用这个“马甲”就行。

3. 核心操作：反编译、修改与重打包JAR

这一步需要我们有点Java开发的经验，但操作并不复杂，可以理解为“外科手术式”的修改。

3.1 在IDE中定位并修改工厂类

我习惯用IntelliJ IDEA，你如果用Eclipse步骤也类似。关键是要创建一个能编译这个jar包中类的环境。

1. 创建一个新的Maven或Gradle项目，或者直接用你现有的Flink项目。
2. 把下载的 flink-sql-connector-mysql-cdc-2.2.1.jar 作为依赖添加到项目中。在Maven中，可以安装到本地仓库后引用，更简单直接的方式是：
   • 在IDEA中，打开项目结构（Project Structure）。
   • 在Libraries一栏，点击+号，选择Java，然后直接选中我们下载的那个JAR文件添加进去。这样项目就能引用到这个jar里的所有类了。
3. 找到目标类：在IDEA中，连续按两次Shift键，打开全局搜索，输入 MySqlTableSourceFactory，找到这个类。因为它是编译好的.class文件，IDEA会反编译显示其内容。
4. “偷梁换柱”复制源码：我们不能直接修改jar包里的.class文件。正确做法是，在项目的src/main/java目录下，创建完全相同的包路径 com.ververica.cdc.connectors.mysql.table，然后新建一个Java文件 MySqlTableSourceFactory.java。
5. 把IDEA反编译看到的那个类的全部内容，复制到我们新建的Java文件中。这时候，一个.class文件就变成了我们可以编辑的.java源文件了。
6. 执行关键修改：在这个Java文件里，找到 factoryIdentifier() 方法，它大概长这样：

   @Override
   public String factoryIdentifier() {
       return "mysql-cdc"; // 就是这一行！
   }
   

   把它改成：

   @Override
   public String factoryIdentifier() {
       return "mysql-test-cdc"; // 改成我们自定义的名字
   }
   

   保存文件。

3.2 编译并替换JAR包中的类文件

接下来，我们需要把修改后的Java文件编译成.class文件，并塞回原来的JAR包里。

1. 在IDEA里编译你的项目（Build -> Build Project）。编译成功后，在项目的 target/classes 目录（或者你设置的编译输出目录）下，找到刚刚生成的 com/ververica/cdc/connectors/mysql/table/MySqlTableSourceFactory.class 文件。
2. 备份原始JAR包：操作前，先把 flink-sql-connector-mysql-cdc-2.2.1.jar 复制一份作为备份，比如命名为 flink-sql-connector-mysql-cdc-2.2.1.jar.backup。
3. 替换类文件：JAR包本质上就是个ZIP压缩包。我们可以用任何ZIP工具（如WinRAR、7-Zip）打开原始JAR包（不是备份包），注意是“打开”而不是解压。然后导航到 /com/ververica/cdc/connectors/mysql/table/ 路径，你会看到里面已经有一个 MySqlTableSourceFactory.class 文件。直接把我们在 target/classes 里新编译好的同名文件拖进去，覆盖掉原来的。保存并关闭ZIP工具。

至此，一个属于我们自己的、标识符为 mysql-test-cdc 的MySQL CDC连接器JAR包就制作完成了。你可以把它重命名为 flink-sql-connector-mysql-test-cdc-2.2.1.jar 以作区分。

4. 在阿里云Flink平台部署自定义连接器

制作好自定义的JAR包后，接下来就是把它和官方的Oracle-CDC连接器一起，上传到阿里云Flink全托管环境中。

4.1 上传自定义连接器

登录阿里云实时计算Flink版控制台，这个入口可能因阿里云控制台迭代而变化，但通常在工作空间列表的“操作”栏里能找到“控制台”链接。

1. 在控制台左侧导航栏，找到 “连接器” 或 “自定义连接器” 管理页面。
2. 点击 “创建自定义连接器” 按钮。
3. 在上传页面，分别上传我们准备好的两个JAR文件：
   • flink-sql-connector-oracle-cdc-2.2.1.jar （未修改的官方版）
   • flink-sql-connector-mysql-test-cdc-2.2.1.jar （我们修改过的自定义版）
4. 上传后，系统通常会自动解析JAR包中的连接器名称和参数。确认无误后，点击完成。这样，这两个连接器就注册到你的Flink工作空间了。

4.2 在SQL作业中调用自定义连接器

现在，在开发SQL作业时，我们就可以像使用内置连接器一样使用它们了，唯一要注意的就是connector参数的值。

创建Oracle CDC源表：

CREATE TABLE oracle_source (
    -- 你的字段定义
    id BIGINT,
    name STRING,
    update_time TIMESTAMP(0)
) WITH (
    'connector' = 'oracle-cdc', -- 使用官方Oracle连接器
    'hostname' = 'your-oracle-host',
    'port' = '1521',
    'username' = 'your_username',
    'password' = 'your_password',
    'database-name' = 'XE', -- 或你的数据库名
    'schema-name' = 'YOUR_SCHEMA',
    'table-name' = 'YOUR_TABLE'
);

创建自定义MySQL CDC源表：

CREATE TABLE mysql_source (
    -- 你的字段定义
    user_id BIGINT,
    log_action STRING,
    event_time TIMESTAMP(0)
) WITH (
    'connector' = 'mysql-test-cdc', -- 关键！使用我们自定义的连接器名
    'hostname' = 'your-mysql-host',
    'port' = '3306',
    'username' = 'your_username',
    'password' = 'your_password',
    'database-name' = 'your_database',
    'table-name' = 'your_table',
    'server-id' = '5400-5404' -- 别忘了设置一个唯一的server-id范围
);

进行流式关联或写入下游：

-- 示例：将两个源表的数据关联后写入打印器（实际应用中可写入Kafka、Hologres等）
INSERT INTO blackhole_sink
SELECT 
    o.id, 
    o.name, 
    m.log_action, 
    m.event_time
FROM oracle_source o
JOIN mysql_source m ON o.id = m.user_id;

这样提交作业后，Flink运行时加载的就是我们上传的、版本一致的CDC连接器JAR包，之前令人头疼的类冲突问题就迎刃而解了。作业可以稳定运行，实现Oracle和MySQL数据的实时同步与融合计算。

5. 避坑指南与进阶思考

踩过这个坑之后，我也总结了一些额外的经验和注意事项，能帮你走得更顺。

5.1 常见问题排查

• 上传失败：检查JAR包是否完整，网络是否通畅。阿里云Flink对上传的JAR包大小可能有限制，但CDC连接器包通常不会超。
• 作业报“找不到连接器”：检查SQL中connector参数的名字是否和你自定义的完全一致，包括大小写。最好直接从控制台的连接器列表里复制。
• 作业启动后报其他类错误：这很可能是因为你修改工厂类时，项目依赖的Flink版本或CDC版本与阿里云环境不匹配。最稳妥的做法是，用于编译修改类的本地项目，其Flink版本尽量与阿里云Flink工作空间选择的引擎版本（如Flink 1.13或1.15）保持一致。你可以创建一个干净的Maven项目，只引入对应版本的flink-sql-connector-mysql-cdc依赖来修改，避免其他依赖干扰。
• 性能与稳定性：自定义连接器在功能上和官方原版一致。但需要注意，在增量读取阶段，如果源表数据变更非常频繁，可能需要调整Debezium的相关参数，比如debezium.max.queue.size、debezium.max.batch.size等，以避免队列积压。这些参数可以在CREATE TABLE的WITH参数中配置。

5.2 关于版本匹配的深度理解

我们这次解决的是MySQL和Oracle连接器之间的兼容性。实际上，Flink CDC是一个生态，还包括PostgreSQL、SQL Server、MongoDB等连接器。如果你未来需要引入第三个CDC连接器，必须确保所有自定义CDC连接器的Flink CDC版本一致。比如，都用2.2.1版本。因为不同大版本间的API可能有变动。

另外，还要留意阿里云Flink引擎版本（VVR）与Flink CDC社区版的映射关系。根据阿里云官方文档的“社区CDC”章节，不同的VVR版本有推荐的Flink CDC Release版本。例如，VVR 8.0.x 推荐使用 release-3.0 系列的CDC连接器。虽然我们实战的2.2.1版本在很多场景下可用，但对于生产环境，查阅官方最新的兼容性列表是更稳妥的做法。

5.3 更优雅的解决方案展望

手动修改JAR包毕竟是个“体力活”。对于需要持续集成和部署的团队，可以考虑更自动化的方式：

1. 源码构建：直接从Flink CDC官方GitHub仓库拉取对应版本的源码（如release-2.2.1分支），在源码层面修改MySqlTableSourceFactory中的标识符，然后用Maven命令（如 mvn clean package -DskipTests）重新编译整个flink-sql-connector-mysql-cdc模块。这样生成的JAR包是最干净、最标准的。
2. 依赖管理：如果你的公司大量使用自定义连接器，可以搭建内部的Maven仓库，将定制好的连接器JAR包部署上去。这样在开发项目的pom.xml中可以直接引用，方便版本管理。

这次实战让我深刻体会到，在享受云服务便利的同时，保留一定的技术灵活性和“动手能力”是多么重要。阿里云Flink全托管虽然屏蔽了底层集群的运维复杂度，但在连接器生态这种与业务强相关的层面，还是给我们开发者留出了足够的自定义空间。搞定这个兼容性问题后，我们的实时数据管道终于可以顺畅地流淌起Oracle和MySQL的混合数据了。