使用Kafka Signal的Debezium SQL Server源连接器临时快照失败

问题描述 #

配置Debezium的Kafka信号通道后，发送快照信号仅收到如下日志，但未触发实际快照：

connect | 2024-05-13 06:43:40,834 INFO || Requested 'INCREMENTAL' snapshot of data collections '[testDB.dbo.orders]' with additional conditions '[AdditionalCondition{dataCollection=testDB.dbo.orders, filter='product_id>102'}]' and surrogate key 'id' [io.debezium.pipeline.signal.actions.snapshotting.ExecuteSnapshot]

同时存在以下现象：

• 使用source信号通道时，通过信号表可正常执行临时快照
• 切换为Kafka信号方式后，信号表未追加任何快照相关记录

当前操作步骤：

1. 创建信号表：

CREATE TABLE debezium_signal (id VARCHAR(42) PRIMARY KEY, type VARCHAR(32) NOT NULL, data VARCHAR(2048) NULL);

2. 源连接器配置片段：

"signal.data.collection": "testDB.dbo.debezium_signal",
  "signal.enabled.channels": "source,kafka",
  "signal.kafka.topic": "kafka-dbz",
  "signal.kafka.bootstrap.servers": "****",

3. 发送Kafka信号的命令：

echo 'CDC:{"id":"d139b9b7-7777-4547-917d-e1775ea61d46","type":"execute-snapshot","data": {"data-collections": ["testDB.dbo.orders"], "surrogate-key": "id", "type": "INCREMENTAL", "additional-conditions": [{"data-collection": "testDB.dbo.orders", "filter": "product_id>102 and product_id<107"}]}}' | docker-compose exec -T kafka /kafka/bin/kafka-console-producer.sh --bootstrap-server kafka:9092 --topic kafka-dbz --property "parse.key=true" --property "key.separator=":" 

排查与解决要点 #

1. 修正Kafka信号消息格式 #

• Debezium的Kafka信号消息不需要前缀CDC:，多余前缀会导致信号解析失败，直接发送JSON内容即可
• 注意JSON中的转义字符：product_id>102是HTML转义格式，应改为JSON兼容的原始符号（shell中需用反斜杠转义，避免被解析）

修正后的发送命令示例：

echo '{"id":"d139b9b7-7777-4547-917d-e1775ea61d46","type":"execute-snapshot","data": {"data-collections": ["testDB.dbo.orders"], "surrogate-key": "id", "type": "INCREMENTAL", "additional-conditions": [{"data-collection": "testDB.dbo.orders", "filter": "product_id\>102 and product_id\<107"}]}}' | docker-compose exec -T kafka /kafka/bin/kafka-console-producer.sh --bootstrap-server kafka:9092 --topic kafka-dbz

2. 确认Kafka通道权限与连通性 #

• 检查Debezium连接器使用的Kafka客户端账号，是否拥有kafka-dbz主题的消费权限（Kafka信号通道是连接器主动消费该主题消息）
• 验证signal.kafka.bootstrap.servers配置的地址正确性，确保连接器能正常连接Kafka集群

3. 验证增量快照前置条件 #

增量快照必须满足：

• 目标表testDB.dbo.orders存在主键，或已通过message.key.columns配置了自定义键
• 连接器未修改过snapshot.mode为不兼容增量快照的模式（默认配置支持增量快照）
• 过滤器中的product_id字段存在且为可过滤的数值类型

4. 查看完整日志定位问题 #

开启连接器的DEBUG级别日志，检查是否存在信号解析失败、权限不足、表结构不兼容等错误信息，定位具体阻塞点

关于信号表无记录的说明 #

Kafka信号通道的消息不会写入debezium_signal信号表，只有source通道的操作才会在该表留下记录，这是两种通道的正常独立行为，不属于异常情况

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