SpringBootTest中Kafka @EventListener失效问题求助

问题原因 #

你的测试类默认不会被Spring作为Bean纳入上下文管理，因此类中的@EventListener注解无法被Spring的事件监听机制解析。而@KafkaListener能生效是因为Spring Kafka的特殊处理逻辑（通过KafkaListenerAnnotationBeanPostProcessor检测注解并注册监听器），但@EventListener的解析依赖于类本身是Spring管理的Bean。

解决方案 #

方案一：将测试类标记为Spring组件 #

给测试类添加@Component注解，让Spring将其纳入上下文管理，这样@EventListener方法会被自动扫描并注册：

@Slf4j
@SpringBootTest
@ActiveProfiles("kafka-test")
@Component // 新增该注解
class EmbeddedKafkaIntegrationTest {

private static final String TOPIC = "my-topic";

private final BlockingQueue<ConsumerRecord<String, MyEvent>> consumptionQueue = new LinkedBlockingDeque<>();

@EventListener(condition = "event.listenerId.startsWith('test-listener')")
public void idleEventHandler(ListenerContainerIdleEvent event) {
    log.info(event.toString());
}

@KafkaListener(id = "test-listener", idIsGroup = false, topics = "${beb.topic.prefix}" + TOPIC,
        autoStartup = "true", containerFactory = "kafkaListenerContainerFactory")
private void listen(ConsumerRecord<String, MyEvent> consumerRecord) throws InterruptedException {
    log.info("Consume key={}, value={}", consumerRecord.key(), consumerRecord.value());
    consumptionQueue.put(consumerRecord);
}
}

方案二：单独注册事件监听器Bean #

如果不想将测试类本身作为Spring组件，可以把事件监听逻辑封装到内部配置类中，通过@Bean注册为Spring Bean：

@Slf4j
@SpringBootTest
@ActiveProfiles("kafka-test")
class EmbeddedKafkaIntegrationTest {

private static final String TOPIC = "my-topic";

private final BlockingQueue<ConsumerRecord<String, MyEvent>> consumptionQueue = new LinkedBlockingDeque<>();

// 新增内部配置类，注册事件监听器Bean
@Configuration
static class TestEventListenerConfig {

    @Slf4j
    static class IdleEventListener {
        @EventListener(condition = "event.listenerId.startsWith('test-listener')")
        public void idleEventHandler(ListenerContainerIdleEvent event) {
            log.info(event.toString());
        }
    }

    @Bean
    public IdleEventListener idleEventListener() {
        return new IdleEventListener();
    }
}

@KafkaListener(id = "test-listener", idIsGroup = false, topics = "${beb.topic.prefix}" + TOPIC,
        autoStartup = "true", containerFactory = "kafkaListenerContainerFactory")
private void listen(ConsumerRecord<String, MyEvent> consumerRecord) throws InterruptedException {
    log.info("Consume key={}, value={}", consumerRecord.key(), consumerRecord.value());
    consumptionQueue.put(consumerRecord);
}
}

验证说明 #

两种方案本质都是让@EventListener所在的类成为Spring管理的Bean，这样Spring的EventListenerMethodProcessor会扫描到该注解方法，将其注册为事件监听器，从而正常接收ListenerContainerIdleEvent事件。

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