Kafka消费者重试超限制未停止且未触发死信队列问题

你同时配置了RetryTemplate（监听器层面重试）和SeekToCurrentErrorHandler（容器层面重试），两者的重试逻辑会叠加，导致：

• 重试次数远超预期（监听器重试3次后，容器层面还会再重试4次）
• 监听器重试耗尽后，异常抛给ErrorHandler，但此时消息偏移量已被手动处理，死信转发逻辑无法触发

移除RetryTemplate相关配置，完全通过SeekToCurrentErrorHandler统一控制重试次数和死信转发，调整后的配置代码如下：

public class KafkaConsumerConfig {

    @Value("${employee_profile_events_topic}")
    private String kafkaTopicName;

    @Value("${spring.kafka.bootstrap-servers}")
    private String kafkaURL;

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Bean
    public ConsumerFactory<String, EmployeeDTO> consumerFactory() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaURL);
        configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); // 修正配置与构造器的序列化类不一致问题
        configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        configs.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaTopicName);
        configs.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); // 改用布尔值更规范
        configs.put(JsonDeserializer.TRUSTED_PACKAGES, "*"); // 允许反序列化指定包下的DTO

        return new DefaultKafkaConsumerFactory<>(configs, new StringDeserializer(), new JsonDeserializer<>(EmployeeDTO.class));
    }

    @Bean(name = "employeeProfileChangeKafkaListenerContainerFactory")
    public ConcurrentKafkaListenerContainerFactory<String, EmployeeDTO> employeeProfileChangeKafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, EmployeeDTO> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        // 移除冲突的RetryTemplate和statefulRetry配置
        factory.setErrorHandler(errorHandler(deadLetterPublishingRecoverer(kafkaTemplate)));
        return factory;
    }

    @Bean
    public DeadLetterPublishingRecoverer deadLetterPublishingRecoverer(KafkaTemplate<String, String> kafkaTemplate) {
        // 可选：自定义死信主题，替换默认的<原主题>.DLT格式
        // TopicNameStrategy customStrategy = (topic, partition) -> "employee_profile_events_dlt";
        // return new DeadLetterPublishingRecoverer(kafkaTemplate, customStrategy);
        return new DeadLetterPublishingRecoverer(kafkaTemplate);
    }

    @Bean
    public SeekToCurrentErrorHandler errorHandler(DeadLetterPublishingRecoverer deadLetterPublishingRecoverer) {
        // FixedBackOff参数：重试间隔500ms，重试3次（含首次尝试共执行4次，与原RetryTemplate逻辑一致）
        return new SeekToCurrentErrorHandler(deadLetterPublishingRecoverer, new FixedBackOff(500L, 3));
    }
}

1. 移除冲突配置：RetryTemplate和SeekToCurrentErrorHandler不能同时使用，否则会导致重试逻辑叠加，死信转发失效。
2. 统一重试控制：SeekToCurrentErrorHandler的FixedBackOff第二个参数为重试次数，设置为3后，逻辑和你原来RetryTemplate的SimpleRetryPolicy(3)完全一致。
3. 修正序列化配置：原代码中键的序列化配置与构造器实际使用的类不一致，会导致反序列化异常，需统一为StringDeserializer。
4. 死信主题验证：默认死信主题为<原主题>.DLT，请确保该主题已创建；如需自定义主题，可通过TopicNameStrategy指定。
5. 手动提交注意：使用MANUAL AckMode时，需在监听器方法中调用Acknowledgment.acknowledge()确认正常消息；重试失败的消息会由SeekToCurrentErrorHandler自动转发死信，无需手动处理。

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