Spring Boot2.1集成Spring Kafka2.2使用@KafkaListener启动报错

问题背景 #

集成spring-kafka与Spring Boot时出现启动失败，当前使用版本组合为Spring Boot v2.1 + spring-kafka v2.2，符合官方版本兼容矩阵要求，排查多份同类问题方案均未解决。项目未引入JMS、RabbitMQ、其他MQ类消息组件，不存在多消息组件依赖冲突场景，错误仅在使用@KafkaListener注解时触发。

报错日志 #

***************************
APPLICATION FAILED TO START
***************************

Description:

An attempt was made to call the method org.springframework.kafka.listener.AbstractMessageListenerContainer.getContainerProperties()Lorg/springframework/kafka/listener/config/ContainerProperties; but it does not exist. Its class, org.springframework.kafka.listener.AbstractMessageListenerContainer, is available from the following locations:

    jar:file:/C:/Users/Ali/.m2/repository/org/springframework/kafka/spring-kafka/2.2.0.RELEASE/spring-kafka-2.2.0.RELEASE.jar!/org/springframework/kafka/listener/AbstractMessageListenerContainer.class

It was loaded from the following location:

    file:/C:/Users/Ali/.m2/repository/org/springframework/kafka/spring-kafka/2.2.0.RELEASE/spring-kafka-2.2.0.RELEASE.jar


Action:

Correct the classpath of your application so that it contains a single, compatible version of org.springframework.kafka.listener.AbstractMessageListenerContainer

2022-06-28 13:21:16,716 INFO  [,,,] [main] ThreadPoolTaskExecutor[shutdown]: Shutting down ExecutorService 'applicationTaskExecutor'

现有配置与代码 #

pom依赖配置 #

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

消息监听器代码 #

@Component
public class MyListenerComponent {

  @KafkaListener(topics="myTopic")
  public void Listen(MyModelClass obj) {
    //do stuff (nothing complicated, basic java stuff)
  }
}

项目启动类代码 #

@Configuration
@SpringBootApplication
@EnableKafka
public class Application extends SpringBootServletInitializer {

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}

根因说明 #

报错核心原因是本地加载的spring-kafka-2.2.0.RELEASE为2.2大版本的最早发布版，该版本中AbstractMessageListenerContainer#getContainerProperties方法的返回值尚未迁移到org.springframework.kafka.listener.config.ContainerProperties路径，和Spring Boot 2.1后续补丁版本中Kafka自动配置类的方法签名预期不匹配，属于小版本不兼容问题，并非跨大版本的依赖冲突。

修复方案 #

按优先级依次执行以下操作即可修复：

• 完全移除pom中spring-kafka依赖的手动版本配置，直接使用Spring Boot官方依赖管理中内置的匹配版本，若项目继承spring-boot-starter-parent，直接删除所有自定义的spring-kafka版本号配置即可。
• 执行强制依赖更新命令：mvn clean install -U，同时手动删除本地maven仓库下路径C:/Users/Ali/.m2/repository/org/springframework/kafka/spring-kafka/2.2.0.RELEASE/的所有损坏缓存文件，避免旧版本包被重复加载。
• 若项目未继承Spring Boot官方parent，需显式指定spring-kafka版本为2.2.x分支的最新稳定小版本，禁止使用2.2.0.RELEASE这个存在签名问题的早期版本。

替代实现方案（无需@KafkaListener注解） #

如果不想使用注解驱动的监听模式，可以直接手动注册消息监听容器，启动类上移除@EnableKafka注解即可，示例代码如下：

@Bean
public ConcurrentMessageListenerContainer<String, MyModelClass> kafkaMessageListener(
        ConsumerFactory<String, MyModelClass> consumerFactory) {
    ContainerProperties containerProps = new ContainerProperties("myTopic");
    ConcurrentMessageListenerContainer<String, MyModelClass> container =
            new ConcurrentMessageListenerContainer<>(consumerFactory, containerProps);
    container.setupMessageListener((MessageListener<String, MyModelClass>) record -> {
        MyModelClass obj = record.value();
        // 原有消息处理逻辑
    });
    return container;
}

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