单元测试 MessageListener 类

Question

你如何对实现 spring-kafka MessageListener 接口的类进行单元测试？我有一个侦听器类，我正在使用 onMessage 函数手动侦听主题。这个函数非常简单，只接收消息。

我的设置是使用 Spring 5.8、Spring-Kafka 2.2.7、Spring-Kafka-Test、JUnit 和 WITHOUT Spring boot。

我一直在尝试来自 Spring 参考文档和其他帖子的一系列不同示例，但似乎没有一个显示一种简单的方法来测试实现 MessageListener 的 Listener 类。

我

不确定我是否需要设置嵌入式KafkaBroker或EmbeddedKafkaRule，或者是否有其他测试方法。当我尝试使用EmbeddedKafkaRule时，我收到一个错误，说NoClassDefFound。

但是我不明白这个测试用例会如何击中我的onMessage函数。

@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class listenerTest {
    private String topic = "someTopic";
    @ClassRule
    public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, topic);
    private CountDownLatch countDownLatch;
    @Before
    public void setUpTests (){
        Map<String, Object> sProps = KafkaTestUtils.senderProps(embeddedKafka.getEmbeddedKafka().getBrokersAsString());
        ProducerFactory producer = new DefaultKafkaProducerFactory<String, String> (sProps);
        kafkaTemplate = new KafkaTemplate<>(producer);
        kafkaTemplate.setDefaultTopic(topic);
        countDownLatch = new CountDownLatch (1);
    }
    @Test
    public void testReceiver(){
         kafkaTemplate.sendDefault("message");
         assertEquals(latch.getCount(), 0);
    }

类 I 要单元测试

public class listener implements BatchAcknowledgingMessageListener<String, String>{
    private CallbackInterface callback;
    public listener(CallbackInterface callback){
        this.callbackI = callback;
    }
    @Override
    public void onMessage(List<ConsumerRecord<String, String>> records, Acknowledgment ack){
         this.callbackI.handleMessage();
         ack.acknowledge();
    }
}

这抛出了一个奇怪的错误，说这个...NoClassDefFound

Answer 1

对于纯单元测试，您不需要嵌入式代理，您应该直接调用侦听器。

注入模拟回调并验证它是否正确调用。

当我尝试直接调用onMessage函数来测试它时，我收到一个错误，说容器不应该调用函数onMessage。

你叫错onMessage...

public interface BatchMessageListener extends MessageListener {
    @Override
    default void onMessage(Message message) {
        throw new UnsupportedOperationException("Should never be called by the container");
    }
    @Override
    void onMessageBatch(List<Message> messages);
}

编辑

public class MyListener implements BatchAcknowledgingMessageListener<String, String> {
    private final MyService service;
    public MyListener(MyService service) {
        this.service = service;
    }
    @Override
    public void onMessage(List<ConsumerRecord<String, String>> data, Acknowledgment acknowledgment) {
        data.forEach(dat -> this.service.call(dat.value()));
        acknowledgment.acknowledge();
    }
    public interface MyService {
        void call(String toCall);
    }
}

和

class So57192362ApplicationTests {
    @Test
    void test() {
        MyService service = mock(MyService.class);
        MyListener listener = new MyListener(service);
        Acknowledgment acknowledgment = mock(Acknowledgment.class);
        listener.onMessage(Collections.singletonList(new ConsumerRecord<>("foo", 0, 0L, null, "bar")), acknowledgment);
        verify(service).call("bar");
        verify(acknowledgment).acknowledge();
        verifyNoMoreInteractions(service, acknowledgment);
    }
}