用途：在 Confluent Cloud 上搭建生产级变更数据捕获（CDC）流水线，将数据库变更通过 Debezium、Flink、Tableflow 流式写入 Iceberg 或 Delta Lake 表。整体链路：数据库 → Debezium CDC 连接器 → Kafka + Schema Registry → Flink（解码与转换）→ Tableflow → Iceberg/Delta 表。
主要能力：支持仅限全托管 Debezium 连接器的数据库，包括 Microsoft SQL Server CDC Source V2、MySQL CDC Source V2、PostgreSQL CDC Source V2、Oracle XStream CDC Source、DynamoDB CDC Source；管理 Schema Registry 模式（默认 JSON_SR，可选 AVRO、PROTOBUF）；用 Confluent Cloud Flink 解码 Debezium 信封并转换数据；用 Tableflow 将 Kafka topic 物化为 Iceberg 或 Delta 表。
必要用法与关键规则：不要在 CDC 源 topic 上直接启用 Tableflow，需使用 Flink 解码模式：CDC Source Topic → Flink INSERT → 目标 Topic（changelog.mode = 'upsert'）→ Tableflow。原因是 CDC 连接器在删除时产生 null 值记录（tombstone），直接对源 topic 启用 Tableflow 会默认使用 APPEND 模式并因检测到 null 值而暂停。Tableflow 的 changelog 模式在首次物化后不可更改，因此目标 topic 需从一开始就设置 'changelog.mode' = 'upsert'；若需更改模式，必须删除 Tableflow topic、底层 Kafka topic 并重建。清理顺序为：Tableflow topics、Flink INSERT 语句、Flink 目标表（DROP TABLE）、目标 Kafka topics、CDC 连接器、CDC 源 Kafka topics（含 dbhistory/schema-changes）、Schema Registry 中的相关模式。
使用方式：优先使用 Confluent MCP Server 执行所有 Confluent Cloud 操作；不可用时回退到 Confluent CLI（confluent 命令）和 REST API。需要向用户收集环境 ID、Kafka 集群 ID、Flink 计算池 ID、Flink 目录名与数据库名，并生成含占位符的 cdc-credentials.properties。使用前需验证数据库 CDC 前提条件（如 PostgreSQL 的 WAL level = logical，MySQL 的 binlog format = ROW，SQL Server 启用 CDC，Oracle XStream 启用 GoldenGate 复制，DynamoDB 启用 Streams 并设置 NEW_AND_OLD_IMAGES）。
限制：DynamoDB CDC 在 Flink 中不会自动解码信封，需手动提取字段；Oracle XStream 仅支持 Amazon RDS for Oracle 的非 CDB 架构，不支持 Autonomous Database、Standby（Data Guard）和 Downstream Capture，且不支持 after.state.only，需要有效的 Confluent 许可。具体数据库前提条件、连接器模板与类型转换细节参见技能参考资料。