如何使用 Embedded-Kafka-lib 测试 Kafka Consumer，完全使用“withRunningKafka”？

Question

我应该测试我的代码，以便通过嵌入式“withRunningKafka”使用来自 kafka-server 的所有消息，如下所示：https://github.com/manub/scalatest-embedded-kafka

1. 我尝试通过创建的嵌入式生产者向主题发送消息。
2. 我尝试通过项目中的代码使用生成的消息（由嵌入式生产者创建）。

“使用自定义生产者和消费者进行测试”应该{

"work" in {

    withRunningKafka {

      1. val producer: KafkaProducer[String, String] =
               aKafkaProducer[String](valueSerializer, config)

         val topic = "topic-to-test"

         producer.send(new ProducerRecord[String, String](topic, "some message 1"))
         producer.send(new ProducerRecord[String, String](topic, "some message 2"))
         producer.close()

      2. val ok: Future[Done] = Consumer
        .committableSource(
            consumerSettings,
            Subscriptions.topics(topic))
        .map(msg => println(msg.record.value()))
        .runWith(Sink.ignore)

       ok should be (Done)
    }
}}

问题在这里：'ok' 没有给出'Done' 的结果。 一般来说，我测试消费者的逻辑是否正确？

Answer 1

欢迎使用 stackoverflow！

原因ok 从未完成并有结果，因为源正在等待可能的进一步消息。在map前添加.take(2)，源将在两个元素后停止，让okfuture完成。

Answer 2

我认为您同时面临两个问题：

1. Kafka 消费者无限等待元素（如@dvim 所说），因此您需要 .take() 才能真正结束

2. 默认情况下，kafka 消费者组将从当前主题的末尾开始，而不是开始，因此不会消耗在旋转之前发布的消息。您需要设置一个设置，使其从主题的开头而不是结尾开始。