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

【问题标题】：Spring KafkaListener: How to know when it's readySpring KafkaListener：如何知道它何时准备就绪
【发布时间】：2019-09-01 06:17:21
【问题描述】：

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

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

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

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

非常感谢任何帮助！

最好的问候。

【问题讨论】：

标签： spring spring-boot apache-kafka spring-kafka spring-kafka-test

【解决方案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 个分区。

【讨论】：

【解决方案2】：

所以，我刚刚阅读了有关 @PostConstruct 的信息，结果证明您也可以在测试中轻松使用它：

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

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

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

【讨论】：

不，问题仍然存在。
当前想法是实现ConsumerSeekAware 并在调用onPartitionsAssigned 时设置一个布尔值。现在这是稳定的（至少在大约 100 次运行之后）