使用EKS Pod Identity时Kafka连接丢失，报Cannot change principals认证错误

问题概述 #

运行一段时间后Kafka连接丢失，抛出以下错误：

Caused by: org.apache.kafka.common.errors.SaslAuthenticationException: Cannot change principals during re-authentication

已知常规方案是设置AWS_ROLE_SESSION_NAME静态会话名，但EKS Pod Identity场景下会话名自动变更，该方案不适用。当前技术栈为Spring Kafka + Reactor-Kafka。

可行解决方案 #

1. 给ReceiverOptions配置认证异常重试策略 #

Reactor-Kafka基于反应式流，可通过配置ReceiverOptions的重试规则，在遇到认证异常时自动重试并重新建立连接。

@Bean
public ReactiveKafkaConsumerTemplate<String, byte[]> createKafkaReceiver(KafkaProperties kafkaProperties, @Value("${kafka.topics}") List<String> topics) {
    ReceiverOptions<String, byte[]> receiverOptions = ReceiverOptions.create(
            kafkaProperties.buildConsumerProperties(null)
    )
            // 过滤出认证相关异常，触发重试
            .withRetry(retrySpec -> retrySpec
                    .filter(throwable -> {
                        Throwable rootCause = throwable;
                        while (rootCause.getCause() != null) {
                            rootCause = rootCause.getCause();
                        }
                        return rootCause instanceof SaslAuthenticationException;
                    })
                    // 指数退避重试，避免频繁请求
                    .retryBackoff(Duration.ofSeconds(2), Duration.ofSeconds(30))
                    .maxRetries(Integer.MAX_VALUE) // 根据业务需求调整重试次数
            )
            .doOnError(throwable -> {
                if (throwable instanceof SaslAuthenticationException) {
                    log.error("Kafka认证失败，触发重试逻辑", throwable);
                }
            });

    return new ReactiveKafkaConsumerTemplate<>(
            receiverOptions.subscription(topics)
    );
}

2. 自定义消费者刷新逻辑 #

通过Spring的事件监听，捕获Kafka认证失败事件，销毁旧的消费者模板并重新创建新实例，确保使用最新的Pod Identity会话凭证。

@Component
public class KafkaAuthFailureHandler implements ApplicationListener<ListenerContainerFailedEvent> {

    private final KafkaProperties kafkaProperties;
    private final List<String> topics;
    private ReactiveKafkaConsumerTemplate<String, byte[]> consumerTemplate;
    private final ConfigurableBeanFactory beanFactory;
    private static final Logger log = LoggerFactory.getLogger(KafkaAuthFailureHandler.class);

    public KafkaAuthFailureHandler(KafkaProperties kafkaProperties,
                                  @Value("${kafka.topics}") List<String> topics,
                                  ApplicationContext context) {
        this.kafkaProperties = kafkaProperties;
        this.topics = topics;
        this.beanFactory = (ConfigurableBeanFactory) context.getAutowireCapableBeanFactory();
        this.consumerTemplate = createNewConsumerTemplate();
    }

    @Bean
    public ReactiveKafkaConsumerTemplate<String, byte[]> reactiveKafkaConsumerTemplate() {
        return consumerTemplate;
    }

    private ReactiveKafkaConsumerTemplate<String, byte[]> createNewConsumerTemplate() {
        ReceiverOptions<String, byte[]> receiverOptions = ReceiverOptions.create(
                kafkaProperties.buildConsumerProperties(null)
        );
        return new ReactiveKafkaConsumerTemplate<>(receiverOptions.subscription(topics));
    }

    @Override
    public void onApplicationEvent(ListenerContainerFailedEvent event) {
        Throwable rootCause = event.getThrowable();
        while (rootCause.getCause() != null) {
            rootCause = rootCause.getCause();
        }

        if (rootCause instanceof SaslAuthenticationException) {
            log.error("检测到Kafka认证失败，重新创建消费者实例", rootCause);
            // 销毁旧Bean并注册新实例
            beanFactory.destroySingleton("reactiveKafkaConsumerTemplate");
            this.consumerTemplate = createNewConsumerTemplate();
            beanFactory.registerSingleton("reactiveKafkaConsumerTemplate", consumerTemplate);
        }
    }
}

3. 升级依赖版本 #

检查并升级以下依赖的版本，新版本可能已适配EKS Pod Identity的会话名变更场景：

• AWS MSK IAM认证库（aws-msk-iam-auth）
• Spring Kafka
• Reactor-Kafka

关于setRestartAfterAuthExceptions的说明 #

你提到的setRestartAfterAuthExceptions(true)是针对传统Spring Kafka同步监听容器的配置，不适用于Reactor-Kafka的反应式消费者模板。后者依赖反应式流的错误重试和重新订阅机制，而非容器重启。

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