使用Kafka SMT路由消息至Snowflake表时的异常问题求助

一、解决"No current assignment for partition"异常 #

问题原因 #

使用ExtractTopic$Key转换时，Snowflake Sink Connector会将从键中提取的__table值误判为实际Kafka主题，导致连接器尝试订阅不存在的table_1、table_2主题分区，与实际订阅的all_tables主题分区产生冲突，触发分区分配异常。

解决方案 #

替换ExtractTopic转换逻辑，改用Snowflake连接器自带的动态表名模板配置，直接从消息内容中提取表名，避免混淆逻辑主题与实际Kafka主题：

1. 修改Snowflake Sink Connector配置，移除原有的ExtractTopicFromKeyField转换，添加以下配置：

# 显式订阅all_tables主题
topics: all_tables
# 从消息value的__table字段提取表名
table.name.template: "{{value.__table}}"
# 可选：若源端未做unwrap，此处添加转换确保__table字段在value中
transforms: unwrap
transforms.unwrap.type: io.debezium.transforms.ExtractNewRecordState
transforms.unwrap.add.fields: table

2. 重置连接器偏移量（可选）：
   停止连接器后，执行命令将消费者偏移量设置为最新：

   curl -X POST http://<CONNECTOR_HOST>:<PORT>/connectors/<CONNECTOR_NAME>/offsets -H "Content-Type: application/json" -d '{"offset": {"all_tables-0": {"offset": -1}, "all_tables-1": {"offset": -1}}}'
   

3. 重启连接器，确认任务正常运行。

二、避免自动创建all_tables表 #

问题原因 #

连接器启动初期，若存在未正确转换的消息（或转换逻辑未生效），会默认使用订阅的Kafka主题名all_tables作为表名创建表。

解决方案 #

1. 确保源端转换逻辑完整生效：
   验证源连接器的转换顺序正确（顺序直接影响转换结果），确保所有消息的value中包含__table字段：

   transforms: unwrap,Reroute,ValueToKey
   transforms.unwrap.type: io.debezium.transforms.ExtractNewRecordState
   transforms.unwrap.add.fields: table
   transforms.Reroute.type: io.debezium.transforms.ByLogicalTableRouter
   transforms.Reroute.topic.regex: (.*)
   transforms.Reroute.topic.replacement: all_tables
   transforms.ValueToKey.type: org.apache.kafka.connect.transforms.ValueToKey
   transforms.ValueToKey.fields: __table
   

2. 使用table.name.template强制指定表名来源：
   通过table.name.template: "{{value.__table}}"让连接器从消息中提取表名，避免 fallback 到主题名创建表。

3. 清理已创建的错误表：
   在Snowflake中手动删除all_tables表，避免后续数据写入错误。

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