配置怎么提交
Debezium 连接器以 JSON 提交到 Kafka Connect 的 REST 接口：POST /connectors 新建，PUT /connectors/<name>/config 更新已存在的配置，DELETE 删除，GET .../status 查看运行状态与失败原因。配置写在 config 对象里，键值与 Debezium 官方文档中的属性名一一对应。
三个必须想清楚的点
第一，主题前缀决定生成主题名，必须全小写且不含特殊字符；第二，server.id 让连接器以从库身份接入，每个连接器必须独占编号；第三，schema 历史主题保存表结构变更记录，一旦丢失，连接器重启后无法正确解析后续事件，等于要重新做一次快照。
基本信息
| 提交方式 | Kafka Connect REST 接口，JSON 请求体 |
|---|---|
| 核心项 | connector.class、topic.prefix、database.server.id、库表过滤清单 |
| 历史主题 | schema.history.internal.* 保存表结构变更记录，不能丢 |
| 快照模式 | snapshot.mode 控制启动时是否先做全量快照 |
| 查看状态 | GET /connectors/<name>/status 返回连接器与各任务状态 |
关键配置读法
| 配置项 | 作用 | 建议 |
|---|---|---|
| connector.class | 指定连接器实现类 | 与数据源类型匹配，写错直接提交失败 |
| topic.prefix | 生成主题的前缀（2.x 起取代 database.server.name） | 全小写，避免特殊字符 |
| database.server.id | 以从库身份接入使用的编号 | 每个连接器独占，且不与真实从库冲突 |
| database.include.list / table.include.list | 过滤需要捕获的库与表 | 按需收窄，避免生成大量无用主题 |
| snapshot.mode | 初始快照策略：initial / schema_only / no_data 等 | 大表评估快照时长与源库压力 |
连接器 JSON（提交到 Connect）
连接器 JSON（提交到 Connect）
{
"name": "mysql-cdc-orders",
"config": {
"connector.class": "io.debezium.connector.mysql.MySqlConnector",
"topic.prefix": "db1",
"database.hostname": "10.0.0.5",
"database.port": "3306",
"database.user": "cdc_ro",
"database.password": "***",
"database.server.id": "184054",
"database.include.list": "app",
"table.include.list": "app.orders,app.order_items",
"schema.history.internal.kafka.bootstrap.servers": "kafka1:9092",
"schema.history.internal.kafka.topic": "schema-changes.app",
"snapshot.mode": "initial"
}
}提交、更新与查看
提交、更新与查看
# 新建连接器
curl -s -X POST http://kafka-connect:8083/connectors \
-H 'Content-Type: application/json' -d @debezium-mysql.json
# 查看状态：连接器本身与各任务是否有失败
curl -s http://kafka-connect:8083/connectors/mysql-cdc-orders/status
# 更新已有配置（表过滤清单变更时用 PUT，不会重置位点）
curl -s -X PUT http://kafka-connect:8083/connectors/mysql-cdc-orders/config \
-H 'Content-Type: application/json' -d @config-only.jsonsnapshot.mode=initial 会先做一次全量快照再转入增量；只想要增量变更（例如已有存量数据由别的方式导入）就用 no_data 或 schema_only，选错会重复导入全部存量。
schema 历史主题（旧名 database.history.kafka.topic，2.x 起改为 schema.history.internal.*）不能丢也不能改名：它记录表结构演进，丢了之后连接器重启无法解析后续事件，只能重新快照。