java - Kafka テスト コンテナーの不安定なテスト

Question

私の春のプロジェクトにカフカを統合しました。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
    }
}