Spring Kafka多监听主题共用单个Retry和DLT主题可行性咨询

可行性结论 #

可以实现多个源主题共用同一组重试（Retry）和死信（DLT）主题，核心是正确自定义RetryTopicNamesProviderFactory并调整主题键的生成逻辑，避免重复键冲突。

错误原因分析 #

你遇到的Duplicate key null错误，是因为默认逻辑下，每个源主题生成的重试/DLT主题会以源主题名称作为键的一部分。当多个源主题对应同一个重试/DLT主题时，会出现键重复的情况，导致Spring Kafka无法合并相同目标主题的配置。

具体实现步骤 #

1. 自定义重试/DLT主题名称生成器 #

创建自定义的RetryTopicNamesProvider，统一返回重试和DLT主题名称：

public class CustomRetryTopicNamesProvider extends DefaultRetryTopicNamesProvider {

    private static final String BASE_RETRY_TOPIC = "domain_events_retry";
    private static final String DLT_TOPIC = "domain_events_dlt";

    @Override
    public String getRetryTopicName(String originalTopic, int attempt) {
        // 生成统一格式的重试主题：domain_events_retry-0、domain_events_retry-1
        return String.format("%s-%d", BASE_RETRY_TOPIC, attempt);
    }

    @Override
    public String getDltTopicName(String originalTopic) {
        // 所有源主题共用同一个DLT主题
        return DLT_TOPIC;
    }
}

2. 自定义RetryTopicNamesProviderFactory并调整键生成逻辑 #

重写getTopicNameKey方法，确保同一个重试/DLT主题对应唯一的键：

@Configuration
public class RetryTopicConfiguration {

    @Bean
    public RetryTopicNamesProviderFactory customRetryTopicNamesProviderFactory() {
        return new RetryTopicNamesProviderFactory() {
            @Override
            public RetryTopicNamesProvider createRetryTopicNamesProvider(RetryTopicConfiguration config) {
                return new CustomRetryTopicNamesProvider();
            }

            @Override
            public String getTopicNameKey(String topic, DestinationTopic.Properties properties) {
                // 针对重试主题，用统一的主题名+重试级别作为键
                if (DestinationTopic.Type.RETRY.equals(properties.getType())) {
                    // 从suffix中提取重试级别（比如"-retry-0"中的0）
                    int attempt = Integer.parseInt(properties.getSuffix().split("-")[2]);
                    return String.format("%s-%d", "domain_events_retry", attempt);
                }
                // DLT主题用固定键
                else if (DestinationTopic.Type.DLT.equals(properties.getType())) {
                    return "domain_events_dlt";
                }
                // 其他类型主题沿用默认逻辑
                return super.getTopicNameKey(topic, properties);
            }
        };
    }
}

3. 在@RetryableTopic注解中指定自定义Factory #

修改你的监听注解，关联自定义的名称生成工厂：

@RetryableTopic(
        attempts = "3",
        backoff = @Backoff(delay = 5000, multiplier = 3.0, maxDelay = 60000, random = true),
        autoCreateTopics = "false",
        exclude = {
            JsonProcessingException.class,
            NullPointerException.class,
            IllegalStateException.class,
            IllegalArgumentException.class,
            UnsupportedOperationException.class,
            ConstraintViolationException.class,
            InvalidRequestException.class},
        autoStartDltHandler = "false",
        topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE,
        namesProviderFactory = "customRetryTopicNamesProviderFactory" // 指定自定义工厂
)
@KafkaListener(topics = "#{'${spring.kafka.consumer.properties.topics}'.split('\\s+')}")
public void consumeMessage(String message) {
    // 你的消息处理逻辑
}

注意事项 #

• 分区规划：共用的重试/DLT主题分区数建议与源主题的最大分区数保持一致，避免消费瓶颈。
• 消息兼容性：确保重试/DLT主题中的消息格式能被消费逻辑兼容（比如所有源主题的消息结构一致，或消息携带源主题标识以便区分处理）。
• 消费者组：Spring Kafka会自动为重试/DLT主题创建对应的消费者组，无需手动配置，但要确保这些消费者组有足够的消费能力。

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