Spring Integration+Spring Kafka自定义Listener遇授权异常后无法恢复

问题背景 #

使用Spring Integration结合Spring Kafka搭建消息消费链路，自定义监听器从指定主题读取消息并发送至Spring Integration通道。Pod运行10-12小时重启后，应用启动无报错，但运行一段时间后出现授权异常，且异常涉及的kafka_outgoing_topic不属于当前微服务，代码中也未出现该主题。异常发生后，自定义监听器无法自动重启。

异常日志 #

org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [kafka_outgoing_topic]

"2022-08-13T09:24:12.617+0000","2022-08-13 09:24:01.975","1 --- [customlistner-C-1] essageListenerContainer$ListenerConsumer : Authorization Exception and no authorizationExceptionRetryInterval set

org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [kafka_outgoing_topic]

[customlistner-C-1] essageListenerContainer$ListenerConsumer : group-id : Consumer stopped.

相关代码 #

@Bean("ConsumerBean")
public KafkaMessageDrivenChannelAdapter<String, String> kafkaMessageDrivenChannelAdapter() {
    KafkaMessageDrivenChannelAdapter<String, String> kafkaMessageDrivenChannelAdapter = new KafkaMessageDrivenChannelAdapter<>(listener());
    kafkaMessageDrivenChannelAdapter.setOutputChannel(messageFromKafka());
    kafkaMessageDrivenChannelAdapter.setAutoStartup(true);
    return kafkaMessageDrivenChannelAdapter;
}

@Bean("customlistner")
public ConcurrentMessageListenerContainer<String, String> listener()  {
    ContainerProperties properties = new ContainerProperties(kafkaTopic);
    properties.setGroupId("group-id");
    return (ConcurrentMessageListenerContainer<String, String>) new ConcurrentMessageListenerContainer<>(consumerFactory, properties);
}

排查方向与解决方案 #

1. 消费者组ID冲突 #

当前使用的group-id为通用名称，极有可能存在其他微服务使用同一组ID订阅了kafka_outgoing_topic。Kafka消费者组会同步组内所有订阅主题的元数据，导致当前消费者被Broker要求校验该无关主题的权限，最终触发授权异常。

• 解决：将group-id修改为当前服务专属的唯一标识，例如your-service-name-kafka-consumer-group。

2. 配置授权异常重试机制 #

日志明确提示no authorizationExceptionRetryInterval set，Spring Kafka默认遇到授权异常时不会自动重试，直接停止消费者容器。

• 解决：在ContainerProperties中添加重试间隔配置，让容器定期尝试重启：

  properties.setAuthorizationExceptionRetryInterval(Duration.ofMinutes(5)); // 可根据实际情况调整间隔
  

3. 检查Kafka ACL与Broker配置 #

联系运维团队确认：

• 当前消费者账号的ACL权限是否被误关联到kafka_outgoing_topic；
• 是否存在其他服务使用同一消费者组ID订阅该无关主题，清理无效的组订阅关系。

4. 排查隐藏的主题引用 #

全局搜索项目代码（包括YAML/Properties配置文件），确认是否存在Spring Integration出站适配器或其他组件无意中引用了kafka_outgoing_topic，比如配置文件中的占位符或自动配置导致的隐式绑定。

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