我试图通过设置kafka服务器并使用producer发送消息来在本地测试我的代码，但是我想知道是否有一种方法可以为这段代码编写单元测试（测试使用者收到的消息是否正确）。
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)
1条答案
按热度按时间s4chpxco1#
您可以使用以下工具测试代码：
缩放试验
scalatest embedded kafka：支持创建使用scalatest的内存中kafka示例。
akka streams测试套件：提供
TestSink.probe它可以检查并控制流元素（即消息）的需求。akka-streams-kafka项目在自己的测试中使用了上述方法。看看它的
IntegrationSpec，您可以根据自己的需要进行调整。