scala - Akka Streams Kafka - 消费者单元测试

Question

我试图通过设置 Kafka 服务器并使用生产者发送消息来在本地测试我的代码，但我想知道是否有一种方法可以为这段代码编写单元测试（测试消息是否由消费者是正确的）。

val consumerSettings = ConsumerSettings(system, 
  new ByteArrayDeserializer, new StringDeserializer)
  .withBootstrapServers("localhost:9092")
  .withGroupId("group1")
  .withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")

val done = Consumer.committableSource(consumerSettings, 
  Subscriptions.topics("topic1"))
  .map { msg =>
    msg.committableOffset.commitScaladsl()
  }
  .runWith(Sink.ignore)

Answer 1

您可以使用以下工具测试您的代码：

• ScalaTest
• scalatest-embedded-kafka：允许创建与 ScalaTest 一起使用的内存中 Kafka 实例。
• Akka Streams Testkit：提供一个TestSink.probe可以检查和控制流元素（即消息）的需求。

Akka Streams Kafka 项目在其自己的测试中使用上述内容。看看它IntegrationSpec，您可以根据自己的需要进行调整。