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作品的创作。
• Akka Streams Testkit：提供可以检查和控制流元素（即消息）的需求的TestSink.probe。

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