Spring Kafka出现Kafka错误时如何禁止监听器容器停止

Spring Kafka 内置的默认错误处理器会将Kafka broker返回的认证类异常（如SaslAuthenticationException、SecurityDisabledException等）归类为不可恢复异常，触发消费容器自动停止的逻辑，避免无效重试占用资源。可以通过自定义错误处理器的方式关闭该默认行为，实现认证错误时自动重试、容器不停止的效果，具体配置如下：

配置步骤（Spring Kafka 2.8+版本） #

1. 自定义CommonErrorHandler，将认证类异常从不可恢复异常列表中移除，同时配置重试策略

@Configuration
public class KafkaConsumerConfig {
    @Bean
    public CommonErrorHandler customKafkaErrorHandler() {
        // 配置重试策略：此处设置为间隔10秒无限次重试，可根据业务需求调整间隔时长和最大重试次数
        DefaultErrorHandler errorHandler = new DefaultErrorHandler(
            new FixedBackOff(10000L, FixedBackOff.UNLIMITED_ATTEMPTS)
        );
        // 移除认证类异常的不可恢复标记，遇到这类异常时不停止容器、按策略重试
        errorHandler.removeNotRetryableException(org.apache.kafka.common.errors.SaslAuthenticationException.class);
        errorHandler.removeNotRetryableException(org.apache.kafka.common.errors.SaslException.class);
        errorHandler.removeNotRetryableException(org.apache.kafka.common.errors.SecurityDisabledException.class);
        errorHandler.removeNotRetryableException(org.apache.kafka.common.errors.AuthorizationException.class);
        return errorHandler;
    }

    // 将自定义错误处理器绑定到Kafka监听容器工厂
    @Bean
    public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
            ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
            ConsumerFactory<Object, Object> consumerFactory,
            CommonErrorHandler customKafkaErrorHandler) {
        ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        configurer.configure(factory, consumerFactory);
        factory.setCommonErrorHandler(customKafkaErrorHandler);
        return factory;
    }
}

Spring Kafka 2.8之前版本适配 #

2.8版本之前的Spring Kafka使用ErrorHandler接口而非CommonErrorHandler，配置如下：

@Bean
public ErrorHandler legacyKafkaErrorHandler() {
    SeekToCurrentErrorHandler errorHandler = new SeekToCurrentErrorHandler(
        new FixedBackOff(10000L, FixedBackOff.UNLIMITED_ATTEMPTS)
    );
    errorHandler.removeNotRetryableException(org.apache.kafka.common.errors.SaslAuthenticationException.class);
    return errorHandler;
}

注意事项 #

• 配置完成后无需修改原有@KafkaListener注解的任何参数即可生效
• 若设置为无限次重试，建议配套添加Kafka消费状态的监控告警，避免长时间认证故障未被及时感知
• 可根据业务实际需求调整重试间隔和最大重试次数，重试次数达到上限后默认仍会停止容器

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