如何为Kafka Connect中每个主题配置独立的Dead Letter Queue？

有两种可行的方案实现按来源主题分流无效消息：

方案一：利用主题正则捕获组的变量替换（推荐） #

Kafka Connect支持通过${topic.group.N}引用topics.regex中的捕获分组，刚好适配你的场景：
你的正则my-topic-sample\.(.+)会把test1/test2/test3作为第一组捕获值，直接将DLQ主题配置为包含这个变量即可：

errors.tolerance=all
errors.deadletterqueue.topic.name=error_topic.${topic.group.1}

这样来自my-topic-sample.test1的无效消息会进入error_topic.test1，test2的进入error_topic.test2，以此类推。

注意：这个功能需要Kafka Connect 2.0及以上版本支持，若你的版本较低，建议升级或用第二种方案。

方案二：创建多个独立的S3 Sink连接器 #

如果无法使用变量替换，可以为每个源主题单独创建一个连接器：

• 第一个连接器配置：

topics=my-topic-sample.test1
errors.tolerance=all
errors.deadletterqueue.topic.name=error_topic.test1
# 其他S3相关配置保持一致

• 第二个连接器配置：

topics=my-topic-sample.test2
errors.tolerance=all
errors.deadletterqueue.topic.name=error_topic.test2
# 其他S3相关配置保持一致

• 第三个同理对应test3。

这种方式虽然需要维护多个连接器，但兼容性更好，适合所有版本的Kafka Connect。

额外注意事项 #

• 无论用哪种方案，建议提前创建好对应的DLQ主题，或者开启Kafka的auto.create.topics.enable配置（生产环境需谨慎评估）。
• 配置完成后，可以通过发送无效消息测试，查看Connect日志或DLQ主题的消息，验证路由是否正确。

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