Spring KafkaListener：如何知道何时准备就绪

Question

我有一个简单的 Spring Boot 应用程序，它从 Kafka 读取并写入 Kafka。我写了一个SpringBootTestusing anEmbeddedKafka来测试所有这些。

主要问题是：有时测试失败是因为测试过早发送Kafka消息。这样，在 Spring 应用程序（或者KafkaListener准确地说）准备就绪之前，消息已经写入 Kafka 。由于侦听器从latest偏移量中读取（我不想为我的测试更改任何配置 - 除了 bootstrap.servers），它不会收到该测试中的所有消息。

有谁知道我如何在测试中知道KafkaListener已准备好接收消息？

我能想到的唯一方法是等到/health可用，但我不知道我是否可以确定这意味着KafkaListener完全准备好了。

任何帮助是极大的赞赏！

此致。

Answer 1

如果您有KafkaMessageListenerContainer实例，那么使用起来非常方便org.springframework.kafka.test.utils.ContainerTestUtils.waitForAssignment(Object container, int partitions)。

https://docs.spring.io/spring-kafka/api/org/springframework/kafka/test/utils/ContainerTestUtils.html

例如，调用ContainerTestUtils.waitForAssignment(container, 1);您的测试设置将会阻塞，直到容器分配了 1 个分区。

Answer 2

所以，我刚刚读到@PostConstruct，事实证明您也可以在测试中轻松使用它：

@PostConstruct
public void checkApplicationReady() {
    applicationReady = true;
}

现在我添加了一个@Before方法来等待该标志设置为 true。

到目前为止，这似乎工作得非常好！