如何在Spock中验证@KafkaListener接收Kafka生产者消息

问题原因 #

Spock的交互验证（如1 * eventListener.consumer(event)）是绑定到测试生命周期的断言逻辑，它会在then块执行结束时一次性验证调用次数。而PollingConditions.eventually会重复执行闭包内的代码，第一次执行时如果交互未发生，Spock会直接触发断言失败，不会继续重试检查。

解决方案 #

方案一：通过状态标记替代交互验证 #

在测试中记录方法调用的状态（比如计数器），然后在PollingConditions中检查状态是否符合预期：

@EmbeddedKafka
class KafkaSpec extends Specification {
    @Autowired
    Producer<String, TheEvent> theEventProducer
    
    @SpringSpy
    EventListener eventListener

    def "produce and consume event successfully"() {
        given: "An event"
        def event = MockEventFactory.createEvent()
        def invokeCount = new AtomicInteger(0)

        // 拦截consumer方法调用，记录调用次数
        eventListener.consumer(_) >> { invokeCount.incrementAndGet() }

        when: "生产者发送事件（等待发送完成）"
        theEventProducer.send(new ProducerRecord<>("topic", "123", event)).get()

        then: "消费者最终收到事件"
        new PollingConditions(timeout: 10, delay: 0.5).eventually {
            invokeCount.get() == 1
        }
    }
}

关键细节：

• 调用send().get()确保生产者完成消息发送，避免消息还在缓冲区就开始检查
• 设置delay: 0.5减少频繁检查的开销
• 如果需要精确匹配事件对象，要确保TheEvent类正确实现了equals()方法

方案二：使用@Timeout注解配合原生交互验证 #

如果不需要动态轮询，也可以用Spock的@Timeout注解让测试等待足够时间，直到交互发生：

@EmbeddedKafka
class KafkaSpec extends Specification {
    @Autowired
    Producer<String, TheEvent> theEventProducer
    
    @SpringSpy
    EventListener eventListener

    @Timeout(value = 10, unit = TimeUnit.SECONDS)
    def "produce and consume event successfully"() {
        given: "An event"
        def event = MockEventFactory.createEvent()

        when: "生产者发送事件（等待发送完成）"
        theEventProducer.send(new ProducerRecord<>("topic", "123", event)).get()

        then: "消费者收到事件"
        1 * eventListener.consumer(event)
    }
}

这种方式会一直阻塞直到交互发生或超时，适合简单场景。

注意事项 #

• 确保@EmbeddedKafka配置正确，默认的分区、副本数（均为1）足够满足测试需求
• 如果事件是序列化对象，要确保Kafka的序列化/反序列化配置正确，否则消费者无法解析事件
• 测试环境中消费者组ID要唯一，避免与其他测试实例冲突

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