我已经设置了一个简单的聚合，将来自多个流的值平均在一起，我正在尝试测试它。我已经花了很多时间，我似乎无法在脑海中理清这些概念。我的直播如下：
// 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配置，以及在发布我的输入记录之间设置睡眠状态，取得了不同程度的(不稳定的)成功。
- 这些测试甚至有意义吗？
- 如何针对真实的 Kafka 实例断言此微服务的正确性？
我觉得我在这里做一些概念上错误的事情 - 我只是不知道采取什么其他方法。
你的观察是正确的。缓存使测试变得困难，因为它引入了非确定性。
要编写有用的测试，您有两种选择：
- 通过将缓存大小设置为零来禁用缓存(这样，所有输出记录，包括所有中间记录都是确定性的)
- 仅检查每个键的最后一个结果记录(最后一个结果必须始终相同，与固定输入数据的缓存无关)
顺便说一句：在即将到来的 1.1 中，Kafka 添加了一个公共测试包，我们计划添加更多： https://cwiki.apache.org/confluence/display/KAFKA/KIP-247%3A+Add+public+test+utils+for+Kafka+Streams