confluent-cloud-cdc-tableflow

Confluent Cloud CDC 到 Tableflow 流水线

使用 Debezium、Flink 和 Tableflow 构建生产就绪的 CDC 流水线，将数据库变更通过 Confluent Cloud 流式传输到 Iceberg 或 Delta Lake 表。

概述

此技能可自动搭建完整的 CDC 流水线：

数据库 → 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

关键组件

1. Debezium CDC 源连接器：将数据库变更捕获为事件
2. Schema Registry：管理 Avro/JSON/Protobuf 架构（默认：JSON_SR）
3. Confluent Cloud Flink：解码 Debezium 信封并转换数据
4. Tableflow：Confluent Cloud 原生功能，可将 Kafka 主题物化为 Iceberg 或 Delta 表

关键架构规则

1. 切勿直接在 CDC 源主题上启用 Tableflow。

始终使用 Flink 解码模式：CDC 源主题 → Flink INSERT → 目标主题（changelog.mode = 'upsert'）→ Tableflow。

启用 tombstones.on.delete=true 的 CDC 连接器会在 DELETE 操作时生成值为 null 的 Kafka 记录（墓碑记录）。如果直接在 CDC 源主题上启用 Tableflow，它默认会使用 APPEND 模式，并在遇到墓碑记录时立即暂停：“检测到值为 null 的 Kafka 记录，因此 Tableflow 将暂停。”

Flink 解码层可原生解析 Debezium CDC 语义，将 DELETE 转换为正确的撤回／墓碑消息，以便采用 upsert 模式的 Tableflow 正确处理。

**不要使用 after.state.only=true 作为绕过 Flink 解码步骤的捷径。**虽然该设置会移除 Debezium 信封，但 DELETE 产生的墓碑记录仍会导致 APPEND 模式的 Tableflow 出错。此外，OracleXStreamSource 完全不支持 after.state.only 配置选项。

2. 首次物化后，Tableflow 的变更日志模式不可更改。

Tableflow 首次物化数据时会缓存变更日志模式（APPEND 或 UPSERT）。设置后便无法更改，即使修改 Kafka 主题的 changelog.mode 属性，或删除并重新创建 Tableflow 主题也不行。S3 table_path 以 Kafka 主题名称为键，因此重新创建 Tableflow 主题会复用相同的 S3 路径和缓存状态。

尝试更改模式会导致：“自表物化开始以来，此主题的变更日志模式已修改。”反复切换模式还会进一步破坏状态，导致：“检测到不兼容的架构演进。”

要更改变更日志模式，必须删除 Tableflow 主题、删除底层 Kafka 主题，然后从头重新创建两者。因此，从一开始就使用 'changelog.mode' = 'upsert' 创建目标主题至关重要。

3. 流水线的清理顺序很重要。

重置 CDC 到 Tableflow 的流水线时，请按以下顺序删除资源：

1. Tableflow 主题（位于目标主题上）
2. Flink INSERT 语句
3. Flink 目标表（DROP TABLE）
4. 目标 Kafka 主题
5. CDC 连接器
6. CDC 源 Kafka 主题（包括 dbhistory/schema-changes 主题）
7. Schema Registry 中所有关联的架构（包括 -key 和 -value subject）

切勿在连接器仍在运行时删除 CDC 源 Kafka 主题——连接器将无法恢复或重新创建快照，必须彻底重建。

重要说明

• Tableflow 不是连接器。 它是一项原生的主题级功能，通过 Tableflow API 或 Confluent Cloud UI 启用。
• Confluent Cloud Flink 会自动发现 CDC 表。 无需手动创建源表——带有 Schema Registry 架构的主题会自动作为 Flink 表提供。
• 没有 SR 架构的主题仍可处理——注册一个 JSON 架构（可以是不完整的）、使用架构推断，或使用 Flink 的原始 BYTES 和 JSON 函数。请参阅 references/connector-configs.md 中的“处理没有 Schema Registry 的主题”。
• 所有由 SR 支持的格式都以相同方式工作——JSON_SR、AVRO 和 PROTOBUF 均支持 Flink 自动发现和 Tableflow。请根据吞吐量需求与可调试性进行选择。
• 托管连接器使用 output.data.format，而不是 key.converter/value.converter 类。

工作流阶段

阶段 0：工具选择与 MCP 服务验证（关键）

默认使用 Confluent MCP Server。 MCP 服务是执行所有 Confluent Cloud 操作的首选方式。只有在未安装或无法使用 MCP 服务时，才回退到 Confluent CLI（confluent 命令）和 REST API。

0.1 确认 MCP 服务是否可用

检查是否存在 mcp__confluent__* 工具（list-environments、create-connector、create-flink-statement、create-tableflow-topic、list-schemas、list-topics、consume-messages、search-topics-by-name）。

如果 MCP 不可用，则所有操作都回退到 Confluent CLI（confluent 命令）和 REST API。CLI 回退流程应遵循相同的工作流阶段，但使用 CLI 命令而不是 MCP 工具调用。

CLI 回退示例：

# Environment & cluster discovery
confluent environment list
confluent kafka cluster list --environment <env-id>

# Connector operations
confluent connect cluster create --config-file connector-config.json --cluster <cluster-id> --environment <env-id>
confluent connect cluster describe <connector-id>
confluent connect cluster list --cluster <cluster-id> --environment <env-id>

# Flink operations
confluent flink compute-pool create <pool-name> --cloud <cloud> --region <region> --environment <env-id>
confluent flink statement create <statement-name> --sql "<SQL>" --compute-pool <pool-id> --environment <env-id>
confluent flink statement describe <statement-name> --environment <env-id>
confluent flink statement delete <statement-name> --environment <env-id>

# Topic & schema operations
confluent kafka topic list --cluster <cluster-id> --environment <env-id>
confluent schema-registry subject list --environment <env-id>

# Tableflow operations
confluent tableflow topic enable <topic-name> --cluster <cluster-id> --environment <env-id> --storage-type MANAGED --table-formats ICEBERG
confluent tableflow topic list --cluster <cluster-id> --environment <env-id>
confluent tableflow topic describe <topic-name> --cluster <cluster-id> --environment <env-id>
confluent tableflow topic disable <topic-name> --cluster <cluster-id> --environment <env-id>

REST API 回退： 如果 MCP 和 CLI 均不可用，则直接使用 Confluent Cloud REST API。所有调用都使用 HTTP Basic Auth 和 Cloud API Key（而不是 Kafka API 密钥）。端点模式和示例请参阅 references/rest-api.md。

0.2 向用户收集 Confluent Cloud 详细信息

请用户提供以下 Confluent Cloud 详细信息：

详细信息	示例	用途
Environment ID	env-0ypxv6	所有 MCP 调用中的 environmentId
Kafka Cluster ID	lkc-qo5k36	所有 MCP 调用中的 clusterId
Flink Compute Pool ID	lfcp-3v39xw	Flink 语句中的 computePoolId
Flink Catalog Name	my_environment	Flink 语句中的 catalogName
Flink Database Name	cluster_0	Flink 语句中的 databaseName

凭据： 创建一个 cdc-credentials.properties 文件，其中包含以下占位符：Kafka API Key/Secret（集群级）、数据库用户名/密码。让用户在编辑器中填写，并将其添加到 .gitignore。如果用户不希望 Claude 读取该文件，则回退到 CLI：创建带有占位符的 connector-config.json，由用户填写，然后运行 confluent connect cluster create --config-file connector-config.json。

0.3 验证 MCP 集群目标

快速验证：

1. 运行 mcp__confluent__list-topics，确认 MCP 服务已连接到预期集群
2. 运行 mcp__confluent__list-schemas，确认 Schema Registry 可访问

Schema Registry 在环境级别由所有集群共享。

阶段 1：发现与验证

1.1 检查现有设置

使用 MCP 检查已存在的内容：

mcp__confluent__list-connectors(environmentId, clusterId)  →  Existing CDC connectors
mcp__confluent__list-flink-statements(environmentId, computePoolId)  →  Existing Flink jobs
mcp__confluent__list-tableflow-topics(environmentId, clusterId)  →  Existing Tableflow topics

询问用户：

• “是否已有正在运行的 CDC 连接器？”
• “是否有想要使用的 Flink 计算池，还是需要我创建一个？”
• “数据库是否已配置 CDC？”

1.2 验证主题前缀的唯一性

继续之前，请验证所选的 topic.prefix 不会与现有主题冲突：

mcp__confluent__search-topics-by-name(topicName: "<proposed-prefix>", environmentId, clusterId)

或通过 CLI：

confluent kafka topic list --cluster <cluster-id> --environment <env-id> | grep "^<proposed-prefix>"

如果有任何现有主题与拟定前缀相同，请提醒用户并建议使用唯一前缀。前缀冲突会悄悄地将 CDC 数据与无关主题合并，从而可能破坏两个数据管道。

1.3 检查 Schema Registry 兼容性

默认的 BACKWARD 兼容性可能会在删除数据库列时中止 CDC 连接器。在连接器创建 CDC subject 后，将其设为 FULL_TRANSITIVE：

confluent schema-registry config update --subject "<topic-prefix>.<schema>.<table>-value" --compatibility FULL_TRANSITIVE --environment <env-id>

1.4 收集所需信息

数据库配置：

• 数据库类型（SQL Server、MySQL、PostgreSQL、Oracle 或 DynamoDB）
• 连接详情（主机名、端口、数据库名称）
• 凭据（由用户填写到凭据文件中）
• 要捕获的特定表（格式：schema.table）

Schema 格式： 询问用户选择 JSON_SR（默认，人类可读）、AVRO（载荷更小，吞吐量更高）或 PROTOBUF（强类型）。这些格式都可与 Flink 自动发现和 Tableflow 配合使用，效果相同。切勿使用纯 JSON——它会导致两者都无法正常工作。详细对比请参阅 references/connector-configs.md。

**没有 SR 的现有主题：**有关可选方案（注册 JSON schema、推断 schema 或使用 Flink 原始 BYTES），请参阅 references/connector-configs.md 中的“Handling Topics Without Schema Registry”。

Tableflow 目标位置：

• 目标格式：Iceberg 或 Delta Lake
• 存储：托管式（推荐，由 Confluent 管理 S3）或 BYOB（用户自己的 S3 存储桶，需要 Provider Integration ID）

命名约定：

• 默认值：cdc-pipeline-skill-{database-type}-{YYYYMMDD}
• 示例：cdc-pipeline-skill-postgres-20260323

1.5 验证数据库先决条件

每种数据库都需要特定的 CDC 设置。详情请阅读 references/database-prerequisites.md：

• PostgreSQL：WAL 级别 = logical、复制槽、publication
• MySQL：binlog 格式 = ROW、GTID 模式
• SQL Server：在数据库和表上启用 CDC，并运行 SQL Server Agent
• Oracle XStream：启用 GoldenGate 复制（enable_goldengate_replication=TRUE）、ARCHIVELOG 模式、补充日志记录、具有 DBMS_XSTREAM_AUTH 权限的 XStream 管理用户、通过 DBMS_XSTREAM_ADM.CREATE_OUTBOUND 创建 XStream 出站服务器、具有 XStream 连接权限的连接器用户。完整先决条件：https://docs.confluent.io/cloud/current/connectors/cc-oracle-xstream-cdc-source/prereqs-validation.md
• DynamoDB：启用 Streams，并设置为 NEW_AND_OLD_IMAGES

如果数据库配置不正确，请先指导用户完成设置，再继续操作。

Oracle XStream 的重要限制：

• 仅支持 Amazon RDS for Oracle 上的非 CDB 架构
• 不支持 Oracle Autonomous Databases 或 Oracle Standby（Data Guard）
• 不支持 Downstream Capture
• OracleXStreamSource 不支持 after.state.only
• XStream Out 需要有效的 Confluent 许可证

阶段 2：规划

生成完整的配置计划，并提交给用户审批。

2.1 连接器配置

根据数据库类型，使用 references/connector-configs.md 中相应的模板生成连接器配置。模板包含所有必填字段（name、connector.class、topic.prefix、kafka.api.key、output.data.format、decimal.handling.mode 等）以及数据库专属设置。

根据用户偏好设置 schema 格式（默认值为 JSON_SR）：

{
  "output.data.format": "JSON_SR",
  "output.key.format": "JSON_SR"
}

如果用户要求使用其他格式，请将 JSON_SR 替换为 AVRO 或 PROTOBUF。键和值的格式应保持一致。无论选择哪种格式，其他连接器设置都保持不变。

主题命名模式： {topic.prefix}.{schema}.{table} 示例：当 topic.prefix = "postgres-cdc" 时：postgres-cdc.public.customers

2.2 Flink SQL 语句

在 Confluent Cloud Flink 中，CDC 源表会从 Kafka 主题中自动发现。你只需要：

**注意：**以下示例使用了一个包含示意性列名的 customers 表。请根据从 CDC 主题中发现的架构，替换为用户的实际表名、列名和类型。

1. 创建目标表（将纯 JSON_SR 输出到 Tableflow）：

CREATE TABLE `target_customers` (
  `id` INT NOT NULL,
  `name` STRING,
  `email` STRING,
  `created_at` TIMESTAMP_LTZ(3),
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'changelog.mode' = 'upsert'
);

2. 创建 INSERT 语句（用于解码和转换的持续运行作业）：

INSERT INTO `target_customers`
SELECT
  `id`,
  `name`,
  `email`,
  TO_TIMESTAMP_LTZ(`created_at` / 1000, 3)
FROM `postgres-cdc.public.customers`;

Cloud Flink 的重要差异：

• 在 CREATE TABLE 中，不要指定 'connector'、'value.format'、'properties.bootstrap.servers' 或 Schema Registry URL——Cloud Flink 会自动处理这些配置
• 不要为 CDC 主题创建源表——系统会自动发现这些表
• 不要引用 after.* 字段，也不要按 op 进行筛选——Flink 会原生处理 CDC 变更日志语义
• Debezium 时间戳请使用 TIMESTAMP_LTZ(3)，而不是 TIMESTAMP(3)

**DynamoDB CDC 与 Flink 中的 SQL CDC 不同。**自动发现的表包含 id（键）列和 val 列（一个复杂的 ROW 类型，包含 CDC 信封及其中的 op、before.document、after.document）。Flink 不会像处理 SQL Debezium 连接器那样自动解码此信封。你必须手动提取字段：

CREATE TABLE `target_dynamodb` (
  `id` STRING NOT NULL,
  `document` STRING,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH ('changelog.mode' = 'upsert');

INSERT INTO `target_dynamodb`
SELECT `id`, `val`.`after`.`document`
FROM `dynamodb-cdc-source-topic`;

document 字段是一个 JSON 字符串，包含采用 DynamoDB 原生类型格式的 DynamoDB 项（例如，{"name":{"S":"Alice"},"age":{"N":"30"}}）。

**Debezium 类型转换：**完整的类型映射表请参阅 references/flink-sql-patterns.md。关键转换：MicroTimestamp 使用 TO_TIMESTAMP_LTZ(col / 1000, 3)，Timestamp 使用 TO_TIMESTAMP_LTZ(col, 3)，并确保在连接器上设置了 decimal.handling.mode=string（BYTES 默认值在 Flink 中不可用）。

2.3 Tableflow 配置

Tableflow 是一项原生主题级功能，而不是连接器。它按主题启用。

存储选项：

• 托管（推荐）：由 Confluent 管理 S3 存储。无需凭证。
• BYOB（自带存储桶）：用户提供自己的 S3 存储桶。需要在 Confluent Cloud 中设置 Provider Integration ID（Settings → Provider Integrations）。

**表格式：**Iceberg（推荐）或 Delta Lake

2.4 呈现计划

向用户展示：

1. 连接器配置（敏感字段需遮蔽）
2. 要使用的 Flink 计算池
3. Flink SQL 语句（目标表 + INSERT）
4. Tableflow 配置（存储类型、格式）
5. 预期的主题名称

等待用户明确确认后再继续。

阶段 3：执行

使用 MCP 工具逐步执行，并在每个组件完成后检查状态。

3.1 创建 CDC 源连接器

根据用户的数据库类型，使用 references/connector-configs.md 中的模板构建连接器配置。每个模板都包含所有必需字段，包括 name 字段。

使用 MCP：

mcp__confluent__create-connector(
  connectorName: "<connector-name>",
  environmentId: "<env-id>",
  clusterId: "<cluster-id>",
  connectorConfig: { <config from references/connector-configs.md> }
)

**验证：**托管连接器需要 2-5 分钟进行配置。轮询 mcp__confluent__read-connector——tasks: [] 表示仍在配置中；tasks: [{...}] 表示已就绪。然后使用 mcp__confluent__list-schemas(subjectPrefix: "postgres-cdc") 验证架构。如果 5 分钟后仍没有架构且已有任务，请检查数据库连接。请使用 Confluent Cloud UI 查看连接器错误日志（MCP 不会公开这些日志）。

3.2 创建 Flink 计算池（如有需要）

如果用户没有现有的 Flink 计算池，请在执行 SQL 之前创建一个：

confluent flink compute-pool create <pool-name> --cloud <cloud-provider> --region <region> --environment <env-id>

使用与 Kafka 集群相同的云服务提供商和区域。继续之前，请等待计算池状态变为 RUNNING。

3.3 执行 Flink SQL

步骤 1：确认 CDC 表已自动发现：

mcp__confluent__create-flink-statement(
  statementName: "show-tables-check",
  statement: "SHOW TABLES;",
  environmentId: "<env-id>",
  computePoolId: "<pool-id>",
  catalogName: "<environment-display-name>",
  databaseName: "<cluster-display-name>"
)

然后读取结果：

mcp__confluent__read-flink-statement(statementName: "show-tables-check", environmentId: "<env-id>")

查找 CDC 主题表（例如，postgres-cdc.public.customers）。如果不存在，说明连接器尚未生成数据，请等待后重试。

步骤 2：创建目标表：

mcp__confluent__create-flink-statement(
  statementName: "cdc-create-target-customers",
  statement: "CREATE TABLE `target_customers` (...) WITH ('changelog.mode' = 'upsert');",
  environmentId, computePoolId, catalogName, databaseName
)

步骤 3：创建 INSERT 作业：

mcp__confluent__create-flink-statement(
  statementName: "cdc-decode-customers",
  statement: "INSERT INTO `target_customers` SELECT ... FROM `postgres-cdc.public.customers`;",
  environmentId, computePoolId, catalogName, databaseName
)

INSERT 会创建一个持续运行的 Flink 作业。确认作业状态变为 RUNNING（而不是 FAILED）：

mcp__confluent__read-flink-statement(statementName: "cdc-decode-customers", environmentId)

常见 INSERT 故障：

• “Table does not exist” → CDC 源表尚未自动发现；请等待连接器
• “Incompatible types for sink column” → 类型不匹配；检查上文的 Debezium 类型映射
• “Unsupported format” → 从 CREATE TABLE 中移除所有显式格式属性

提示性警告（可忽略）：

• “Primary key does not match upsert key” — CDC 解码模式下的预期情况
• “Highly state-intensive operators without TTL” — 提示性警告；如用于生产环境，请设置 TTL

3.4 启用 Tableflow

使用 MCP：

mcp__confluent__create-tableflow-topic(
  tableflowTopicConfig: {
    "display_name": "target_customers",
    "storage": { "kind": "Managed", "bucket_name": "managed", "provider_integration_id": "managed" },
    "table_formats": ["ICEBERG"],
    "config": { "record_failure_strategy": "SUSPEND", "retention_ms": "6048000000" }
  }
)

**已知限制：**MCP create-tableflow-topic 工具不接受 environmentId 或 clusterId 参数。它默认使用 MCP 服务中配置的集群。如果 MCP 服务指向的集群与目标主题所在的集群不同，将会因“topic not found”而失败。可使用 CLI 或 UI 作为替代方案。

使用 CLI：

# Managed storage (Confluent manages S3)
confluent tableflow topic enable target_customers \
  --cluster <cluster-id> \
  --environment <env-id> \
  --storage-type MANAGED \
  --table-formats ICEBERG

# BYOB / BYOS (user's own S3 bucket)
confluent tableflow topic enable target_customers \
  --cluster <cluster-id> \
  --environment <env-id> \
  --storage-type BYOS \
  --provider-integration <provider-integration-id> \
  --bucket-name <bucket-name> \
  --table-formats ICEBERG

# Azure Data Lake Storage Gen2
confluent tableflow topic enable target_customers \
  --cluster <cluster-id> \
  --environment <env-id> \
  --storage-type AzureDataLakeStorageGen2 \
  --provider-integration <provider-integration-id> \
  --storage-account-name <account-name> \
  --container-name <container-name> \
  --table-formats DELTA

对于 Delta Lake，请使用 --table-formats DELTA，而不是 Iceberg。

确认 Tableflow 已启用：

mcp__confluent__list-tableflow-topics(environmentId, clusterId)

或通过 CLI：

confluent tableflow topic describe target_customers --cluster <cluster-id> --environment <env-id>
confluent tableflow topic list --cluster <cluster-id> --environment <env-id>

状态将从 PENDING 转变为 ACTIVE。

阶段 4：验证与故障排查

4.1 验证端到端管道

**大型表快照：**如果连接器是在大型表上使用 snapshot.mode: initial 创建的，验证可能需要数小时。若要区分正在运行的快照与故障管道：

1. 确认连接器已分配任务（read-connector → tasks 不为空）
2. 确认架构已注册（list-schemas → 存在键/值架构）——这意味着快照已开始生成数据
3. 在 Confluent Cloud UI 中监控源主题的消息数量——持续增长表示正在推进

如果初始验证时使用了 snapshot.mode: schema_only，请在源数据库中插入测试行以触发 CDC 事件，并验证完整管道。有关快照故障排查的详细信息，请参阅 references/troubleshooting.md。

检查各组件：

检查项	MCP 工具	查看内容
连接器正在运行	read-connector	tasks 数组不为空
架构已注册	list-schemas(subjectPrefix)	CDC 主题的键和值架构
Flink 中的 CDC 表	create-flink-statement("SHOW TABLES")	CDC 主题显示为表
Flink 作业正在运行	read-flink-statement	响应中没有错误
目标主题有数据	consume-messages(topicNames)	显示消息（注意：消费者从最新偏移量开始）
Tableflow 已启用	list-tableflow-topics	状态为 PENDING 或 ACTIVE

从目标主题消费数据，以验证解码后的数据：

mcp__confluent__consume-messages(
  topicNames: ["target_customers"],
  value: { "useSchemaRegistry": true },
  key: { "useSchemaRegistry": true },
  maxMessages: 5,
  timeoutMs: 15000
)

注意：消费者从最新偏移量开始。如果初始快照已完成，在数据库发生新的更改之前，你可能会看到 0 条消息。

通过在源数据库中插入一行来测试实时 CDC（请根据用户的实际架构调整表名和列）：

INSERT INTO public.customers (name, email, created_at)
VALUES ('Test User', '[email protected]', NOW());

4.2 故障排除

详细的故障排除信息请参阅 references/troubleshooting.md。

快速参考 — 阻塞管道的问题：

症状	可能原因	解决方法
连接器任务仍为空	仍在配置中	等待 2-5 分钟，然后重试
5 分钟后仍无架构	数据库连接或凭据问题	检查主机、端口、用户名和密码；验证数据库 CDC 配置
SHOW TABLES 中缺少 CDC 表	连接器尚未开始生成数据	先确认架构存在，然后等待
INSERT：“Incompatible types”	Debezium 类型不匹配	使用 TIMESTAMP_LTZ(3) + TO_TIMESTAMP_LTZ；请参阅 references/flink-sql-patterns.md
Tableflow：“topic not found”	MCP 集群不匹配	使用 CLI：confluent tableflow topic enable，或使用 Confluent Cloud UI
Topic 存在，但不在 SHOW TABLES 中	SR 中没有架构	在 SR 中注册 JSON 架构，或使用架构推断；请参阅 references/connector-configs.md

阶段 5：文档

成功完成设置后，向用户提供：

1. 管道摘要表：所有组件的名称、ID 和状态
2. Topic 名称：源 CDC topic 和目标 topic（包含所用的架构格式）
3. 监控：在 Confluent Cloud UI 中检查连接器、Flink 作业和 Tableflow 状态
4. 测试命令：用于验证实时 CDC 的 SQL INSERT

参考资料

• 数据库先决条件：references/database-prerequisites.md
• 连接器配置：references/connector-configs.md
• Flink SQL 模式：references/flink-sql-patterns.md
• 故障排除指南：references/troubleshooting.md
• REST API 参考：references/rest-api.md
• Confluent Cloud Flink 文档：https://docs.confluent.io/cloud/current/flink/overview.md
• Tableflow 文档：https://docs.confluent.io/cloud/current/topics/tableflow/overview.md
• Debezium CDC 文档：https://debezium.io/documentation/
• Confluent MCP 服务：https://github.com/confluentinc/mcp-confluent