如何配置两个携带独立DLT处理器的RetryTopicConfiguration实例

问题根因 #

Spring Kafka 重试主题组件在匹配多个RetryTopicConfiguration时，默认会优先匹配后注册的配置Bean，且如果没有指定独立分组，框架会复用DLT处理器等公共组件，导致你指定的topic匹配规则失效，所有失败消息最终被路由到后注册的兜底配置对应的processDltForABC方法。

修复步骤 #

按以下要求调整你的配置代码即可：

• 给处理特定topic的配置设置更高优先级，保证它先被匹配到
• 两个配置都使用includeTopics明确指定要处理的topic范围，不要混用excludeTopics做兜底
• 给每个配置设置独立的分组名，避免框架跨配置复用组件
• 确保两个KafkaTemplate的泛型对应的序列化/反序列化配置和对应topic的消息格式匹配

修正后的配置代码示例 #

import org.springframework.core.annotation.Order;

// 优先级设置为1，数值越小优先级越高，优先匹配XYZ/WXYZ的topic
@Bean
@Order(1)
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, MyPojo> template) {
     return RetryTopicConfigurationBuilder
        .newInstance()
        .groupName("xyzRetryGroup") // 独立分组名
        .dltHandlerMethod(KafkaConsumerXYZ.class, "processDltForXYZ")
        .exponentialBackoff(delayMs, backoffMultiplier, maxIntervalInMs)
        .maxAttempts(retryAttempt)
        .doNotAutoCreateRetryTopics()
        .includeTopics("XYZ","WXYZ") // 明确指定要处理的topic
        .setTopicSuffixingStrategy(TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE)
        .create(template);
}

// 优先级低于上面的配置，处理剩余的指定topic
@Bean
@Order(2)
public RetryTopicConfiguration myOtherRetryTopic(KafkaTemplate<String, MyOtherPojo> template) {
    return RetryTopicConfigurationBuilder
        .newInstance()
        .groupName("abcRetryGroup") // 独立分组名，和上面不一致
        .dltHandlerMethod(KafkaConsumerABC.class, "processDltForABC")
        .exponentialBackoff(delayMs, backoffMultiplier, maxIntervalInMs)
        .maxAttempts(retryAttempt)
        .doNotAutoCreateRetryTopics()
        // 替换为你实际需要该配置处理的topic列表，不要用excludeTopics兜底
        .includeTopics("ABC", "其他需要该配置处理的topic名")
        .setTopicSuffixingStrategy(TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE)
        .create(template); 
}

验证说明 #

调整后重启服务，分别向XYZ、ABC主题发送消费失败的测试消息，即可看到失败消息分别被路由到processDltForXYZ和processDltForABC方法。

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