apache-kafka - 测试 KafkaStreams 应用程序

Question

我已经设置了一个简单的聚合，将来自多个流的值平均在一起，我正在尝试对其进行测试。我已经花了很多时间，但似乎无法直接理解这些概念。我的流如下:

// Combine multiple streams together.
KStream<String, IndividualTick> tickerStream = 
priceIndexStreamBuilder.stream(exchangeTopics, Consumed.with(...));

// Group by a key & compute average per key
KStream<K, AveragedTick> avgTickerStream = tickStream.selectKey((key, 
value) -> value.getK())
            .groupByKey(...)
            .aggregate(AvgTick::new,
                    (key, value, aggregate) -> {
                        aggregate.addTick(value);
                        return aggregate;
                    },
                    Materialized.with(...))
            .toStream();

indexTickerStream.to(sinkTopic, Produced.with(...));

我的测试使用 EmbeddedKafka，将一堆记录发布到主题，然后坐在阻塞队列中等待记录到达 sinkTopic。

我对这种聚合如何随时间变化很感兴趣，所以我希望断言每个输出代码的平均值。我可能会添加一定程度的窗口，但我现在尽量保持简单。

当我运行测试时，我得到了不同的结果。假设我的拓扑中有 10 个输入记录:

• 我的聚合器被调用了 10 次
• 我在 AverageTick 序列化器中放置的断点被调用了不同的次数。
• 我在测试中断言记录的值。

我认为这是因为 KIP-63 中定义的缓存功能- 记录非常快速地出现在处理节点上，并被最新记录合并/覆盖。 (虽然我不完全确定。)

我的单元测试通过了 ProcessorTopologyTestDriver，但我正在尝试为包含此逻辑的服务编写一些验收测试。

我还尝试使用我的 commit.interval.ms 配置，以及在发布我的输入记录之间放置 sleep ，以取得不同程度的(不稳定的)成功。

• 这些类型的测试有意义吗？
• 如何针对真实的 Kafka 实例断言此微服务的正确性？

我觉得我在这里做的事情在概念上是错误的——我只是不知道该采取什么其他方法。

Answer 1

您的观察是正确的。缓存使测试变得困难，因为它引入了不确定性。

要编写有用的测试，您有两个选择:

• 通过将缓存大小设置为零来禁用缓存(这样，所有输出记录，包括所有中间记录都是确定性的)
• 只检查每个键的最后一个结果记录(最后一个结果必须始终相同，独立于固定输入数据的缓存)

顺便说一句:在即将到来的 1.1 Kafka 中添加了一个公共(public)测试包，我们计划添加更多:https://cwiki.apache.org/confluence/display/KAFKA/KIP-247%3A+Add+public+test+utils+for+Kafka+Streams