如何为AWS MSK环境下的MySQL Debezium连接器新增监控表？

方案1：Schema仅恢复+手动触发单表快照 #

这是最直接的方案，不会触发旧表的全量重同步，仅处理新增的表：

1. 更新连接器配置
   通过Kafka Connect REST API修改配置，添加新表到table.include.list，并设置临时快照模式：

   curl -X PUT -H "Content-Type: application/json" \
     http://<EC2上Connect的地址>:8083/connectors/your-connector-name/config \
     -d '{
       "table.include.list": "shipping.order_shipping,shipping.orders",
       "snapshot.mode": "schema_only_recovery",
       // 保留其他原有配置
     }'
   

   配置更新后，连接器会自动重启，此时仅同步新表的Schema，不会触发旧表快照。

2. 触发新表的快照
   向Debezium信号主题（默认debezium-signals）发送快照触发指令：

   • 若主题未创建，先创建：

     kafka-topics.sh --create --bootstrap-server <MSK集群Bootstrap地址> --topic debezium-signals --partitions 1 --replication-factor 3
     

   • 发送触发信号：

     kafka-console-producer.sh --bootstrap-server <MSK集群Bootstrap地址> --topic debezium-signals <<EOF
     {"type": "execute-snapshot", "data": {"connectorName": "your-connector-name", "tables": ["shipping.orders"]}}
     EOF
     

3. 恢复原有快照模式
   快照完成后，把snapshot.mode改回原有值（如initial或when_needed），避免后续不必要的快照操作。

方案2：使用增量快照（Debezium 1.9+支持） #

如果你的Debezium版本在1.9及以上，增量快照是更优的选择，分块处理新表快照，对数据库压力更小：

1. 更新连接器配置
   添加新表到table.include.list，确保信号主题配置存在（若无则添加signal.data.collection=debezium-signals）：

   curl -X PUT -H "Content-Type: application/json" \
     http://<EC2上Connect的地址>:8083/connectors/your-connector-name/config \
     -d '{
       "table.include.list": "shipping.order_shipping,shipping.orders",
       "signal.data.collection": "debezium-signals",
       // 保留其他原有配置
     }'
   

2. 触发增量快照
   发送增量快照指令到信号主题，仅指定新表：

   kafka-console-producer.sh --bootstrap-server <MSK集群Bootstrap地址> --topic debezium-signals <<EOF
   {"type": "incremental-snapshot", "data": {"connectorName": "your-connector-name", "tables": ["shipping.orders"], "chunk.size": 1000}}
   EOF
   

   这个操作只会对shipping.orders表进行分块快照，完全不影响已同步的shipping.order_shipping表。

排查补充：检查DDL历史捕获情况 #

如果以上方案不生效，可能是Debezium未捕获到新表的创建DDL：

• 临时设置database.history.skip.unparseable.ddl=true并重启连接器，让它重新读取数据库Schema
• 识别新表后，再将该参数改回false，避免跳过后续正常DDL

内容的提问来源于stack exchange，提问作者RushHour