使用@RetryableTopic时自定义DefaultErrorHandler失效的解决方案咨询

环境配置 #

• Spring Boot 版本：2.7.6
• Spring Kafka 版本：2.8.11

问题描述 #

自定义了继承DefaultErrorHandler的CustomDefaultErrorHandler，重写handleOtherException方法处理反序列化异常（跳过坏消息并提交偏移量）。但结合@RetryableTopic与@KafkaListener使用时，注解自动配置的DefaultErrorHandler会覆盖自定义处理器，导致反序列化异常逻辑失效，需要保留注解式重试机制的同时修复该问题。

自定义错误处理器代码 #

public class CustomDefaultErrorHandler extends DefaultErrorHandler {

    private static Logger log = LoggerFactory.getLogger(CustomDefaultErrorHandler.class);
    @Override
    public void handleOtherException(Exception thrownException, Consumer<?, ?> consumer, MessageListenerContainer container, boolean batchListener) {
        manageException(thrownException, consumer);
    }

    private void manageException(Exception ex, Consumer<?, ?> consumer) {
        log.error("Error polling message: " + ex.getMessage());
        if (ex instanceof RecordDeserializationException) {
            RecordDeserializationException rde = (RecordDeserializationException) ex;
            consumer.seek(rde.topicPartition(), rde.offset() + 1L);
            consumer.commitSync();
        } else {
            log.error("Exception not handled");
        }
    }
}

注解式监听代码 #

@RetryableTopic(listenerContainerFactory = "kafkaListenerContainerFactory", backoff = @Backoff(delay = 8000, multiplier = 2.0),
        dltStrategy = DltStrategy.FAIL_ON_ERROR
        , traversingCauses = "true", autoCreateTopics = "true", numPartitions = "3", replicationFactor = "3",
        fixedDelayTopicStrategy = FixedDelayStrategy.MULTIPLE_TOPICS, include = {RetriableException.class, RecoverableDataAccessException.class,
        SQLTransientException.class, CallNotPermittedException.class}
)
@KafkaListener(topics = "${topic.name}", groupId = "order", containerFactory = "kafkaListenerContainerFactory", id = "OTR")
public void consumeOTRMessages(ConsumerRecord<String, PayloadsVO> payload, @Header(KafkaHeaders.RECEIVED_TOPIC) String topicName) throws JsonProcessingException {
    logger.info("Payload :{}", payload.value());
    payloadsService.savePayload(payload.value(), pegasusTopicName);

}

解决方案 #

方法1：通过RetryableTopicConfigurationSupport指定自定义错误处理器 #

创建配置类继承RetryableTopicConfigurationSupport，重写configureErrorHandler方法，替换默认的错误处理器为自定义实现。这样@RetryableTopic生成的所有监听容器都会使用该处理器，同时保留注解的重试逻辑。

@Configuration
public class RetryableTopicConfig extends RetryableTopicConfigurationSupport {

    @Override
    protected DefaultErrorHandler configureErrorHandler(RetryTopicConfigurationConfig config) {
        return new CustomDefaultErrorHandler();
    }
}

方法2：在容器工厂中配置自定义错误处理器 #

自定义ConcurrentKafkaListenerContainerFactory时直接设置自定义错误处理器，确保@RetryableTopic引用的工厂使用该处理器。这种方式能让注解复用工厂的配置，避免被默认处理器覆盖。

@Configuration
public class KafkaConfig {

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory(
            ConsumerFactory<String, Object> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, Object> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        // 配置自定义错误处理器
        factory.setCommonErrorHandler(new CustomDefaultErrorHandler());
        return factory;
    }
}

关键注意事项 #

• @RetryableTopic初始化时，优先使用RetryableTopicConfigurationSupport中配置的错误处理器，其次才会采用容器工厂的配置。
• 自定义处理器时，若需要保留DefaultErrorHandler的默认重试逻辑，可在重写方法中调用super.handleOtherException(...)，避免丢失原有行为。
• 反序列化异常属于消息监听前的异常，不会触发@RetryableTopic的重试逻辑，必须通过handleOtherException方法单独处理。

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