将 kafka 集成到我的 spring 项目中。使用 TestContainer 方法编写了一个集成,但测试有时会失败。似乎 kafka 服务器的初始化有些问题。
这是我下面的代码
def setupSpec() {
kafka = new KafkaContainer()
kafka.start()
System.setProperty("kafka.consumer.endpoint", kafka.bootstrapServers.replace("PLAINTEXT://", ""))
}
def setup() {
RestAssured.port = port
}
def "test profile update events"() {
given:
String INPUT_TOPIC = "EventXX"
when:
KafkaProducer<String, String> kafkaProducer = createProducer()
kafkaProducer.send(new ProducerRecord<>(INPUT_TOPIC, "foo")).get().topic()
kafkaProducer.send(new ProducerRecord<>(INPUT_TOPIC, "foo")).get().topic()
kafkaProducer.send(new ProducerRecord<>(INPUT_TOPIC, "foo")).get().topic()
kafkaProducer.send(new ProducerRecord<>(INPUT_TOPIC, "foo")).get().topic()
then:
TestUtil.waitFor({ EventConsumer.msgConsumed == true }, 5000)
kafkaProducer.close()
}
现在有趣的是,如果我在测试中发送消息之前添加Thread.sleep(10000)它总是有效的,但这种方法对我来说看起来有点脏。在运行任何测试之前,我们如何确保 kafka 服务器已启动并运行。我通过验证setupSpec中的kafkaSendRecieve尝试了以下方法,但失败了。我正在粘贴下面的代码
def validatekafkaSendRecieve() {
def started = false
String INPUT_TOPIC = "kafkaTest"
def producer = createProducer()
def consumer = createConsumer(INPUT_TOPIC)
Thread.sleep(9000)
while (!started) {
producer.send(new ProducerRecord<>(INPUT_TOPIC, "foo")).get()
started = consumeMessage(INPUT_TOPIC, consumer)
}
producer.close()
consumer.close()
}
def consumeMessage(String topic, KafkaConsumer kafkaConsumer) {
def message = kafkaConsumer.poll(3)
if (!message.isEmpty()) {
def messageList = message.records(topic).asList()
if (messageList != null && !message.isEmpty()) {
return true
}
} else {
return false
}
}