1

我们正在尝试测试一个 cloud-stream-kafka 应用程序，在测试中我们有多个发送消息的测试方法，以及一个等待响应的 @KafkaListener。

但是，第一次测试往往会通过，而第二次测试往往会失败。

任何指针将不胜感激。

@SpringBootTest(properties = "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}")
@EmbeddedKafka(topics = "input", partitions = 1)
@DirtiesContext
class EmbeddedKafkaListenerTest {

  private CountDownLatch latch;

  private String message;

  @BeforeEach
  void setUp() {
    this.message = null;
    this.latch = new CountDownLatch(1);
  }

  @Test
  void testSendFirstMessage(@Autowired KafkaTemplate<String, byte[]> template)
      throws InterruptedException {
    template.send("input", "Hello World 1".getBytes());
    assertTrue(latch.await(10, TimeUnit.SECONDS));
    assertEquals("Hello World 1", message);
  }

  @Test
  void testSendSecondMessage(@Autowired KafkaTemplate<String, byte[]> template)
      throws InterruptedException {
    template.send("input", "Hello World 2".getBytes());
    assertTrue(latch.await(10, TimeUnit.SECONDS));
    assertEquals("Hello World 2", message);
  }

  @KafkaListener(topics = "input", id = "kafka-listener-consumer")
  void listener(Message<byte[]> message) {
    this.message = new String(message.getPayload(), StandardCharsets.UTF_8);
    this.latch.countDown();
  }
}

似乎@KafkaListener正在为每个测试注册一个实例，因为我们注意到使用该id值会导致java.lang.IllegalStateException: Another endpoint is already registered with id 'kafka-listener-consumer'

@RabbitListener当使用的消息传递框架是 RabbitMQ 时，我使用过类似的测试。我希望我可以做类似的事情，因为一些测试用例涉及等待没有消息被发布，我们可以用assertFalse(latch.await(10, TimeUnit.SECONDS))

4

1 回答 1

1

我认为你的@KafkaListener方法应该进入@Configuration类，否则它确实EmbeddedKafkaListenerTest是每个测试方法实例化的，因此@KafkaListener被解析与你有测试方法一样多。

另一种方法是使用@DirtiesContext(classMode = ClassMode.AFTER_EACH_TEST_METHOD)，因此您不仅将拥有EmbeddedKafkaListenerTest每个测试方法的新实例，而且还拥有 Spring ApplicationContext。

而且还有一种使用方法@TestInstance(TestInstance.Lifecycle.PER_CLASS)，因此您将拥有一个EmbeddedKafkaListenerTest用于整个测试套件的单一测试套件，并且您 @KafkaListener不会被多次解析。

于 2019-09-03T14:54:40.323 回答