无法对 @KafkaListener 带注释的方法进行单元测试

我正在尝试在 Spring 中对 kafka 消费者类进行单元测试。我想知道如果 kafka 消息发送到它的主题，则侦听器方法被正确调用。我的消费者类注释如下：

@KafkaListener(topics = "${kafka.topics.myTopic}")
public void myKafkaMessageEvent(final String message) { ...

如果我@Autowire消费者，当我发送kafka消息时，监听器方法被正确调用，但我不能断言该方法被调用，因为该类不是模拟。

如果我模拟消费者，当我发送 kafka 消息时，根本不会调用侦听器方法。我可以直接调用该方法，并断言它有效，但这并没有达到我想要的效果，即检查当我向其主题发送 kafka 消息时是否调用该方法。

现在我已经采取在消费者内部放置一个计数器的方法，并在每次调用侦听器方法时递增它，然后检查它的值是否已更改。对我来说，创建一个仅用于测试的变量似乎是一个糟糕的解决方案。

有没有办法让被嘲笑的消费者也收到卡夫卡消息？或者其他一些方法来断言调用了非模拟的消费者侦听器方法？

听起来您正在请求类似于 Spring AMQP 测试框架中的内容：https://docs.spring.io/spring-amqp/docs/2.0.3.RELEASE/reference/html/_reference.html#test-harness https://docs.spring.io/spring-amqp/docs/2.0.3.RELEASE/reference/html/_reference.html#test-harness

所以，如果你不擅长额外的变量，你可以借用它solution https://github.com/spring-projects/spring-amqp/tree/master/spring-rabbit-test/src/main/java/org/springframework/amqp/rabbit/test并实施你自己的“线束”。

我认为这应该是对框架的一个很好的补充，所以请提出适当的建议issue https://github.com/spring-projects/spring-kafka/issues我们可以共同为公众带来这样一个工具。

UPDATE

因此，根据 Spring AMQP 基金会，我在测试配置中执行了此操作：

public static class KafkaListenerTestHarness extends KafkaListenerAnnotationBeanPostProcessor {

    private final Map<String, Object> listeners = new HashMap<>();

    @Override
    protected void processListener(MethodKafkaListenerEndpoint endpoint, KafkaListener kafkaListener,
            Object bean, Object adminTarget, String beanName) {

        bean = Mockito.spy(bean);

        this.listeners.put(kafkaListener.id(), bean);

        super.processListener(endpoint, kafkaListener, bean, adminTarget, beanName);
    }

    @SuppressWarnings("unchecked")
    public <T> T getSpy(String id) {
        return (T) this.listeners.get(id);
    }

}

...

@SuppressWarnings("rawtypes")
@Bean(name = KafkaListenerConfigUtils.KAFKA_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)
@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
public static KafkaListenerTestHarness kafkaListenerAnnotationBeanPostProcessor() {
    return new KafkaListenerTestHarness();
}

然后在目标测​​试用例中我这样使用它：

@Autowired
private KafkaListenerTestHarness harness;
...
Listener listener = this.harness.getSpy("foo");

verify(listener, times(2)).listen1("foo");