如何对Kafka流进行单元测试

Question

在探索如何对Kafka流进行单元测试时，我遇到了ProcessorTopologyTestDriver，不幸的是，这个类似乎被版本0.10.1.0(KAFKA-4408)破坏了

对于KTable的问题，是否有一个解决方案？

我看到了“mocked streams”项目，但首先它使用的是0.10.2.0，而我使用的是0.10.1.1，其次它是Scala，而我的测试是Java/Groovy。

这里的任何关于如何在不需要引导zookeeper/kafka的情况下对流进行单元测试的帮助都将非常棒。

注意：我确实有使用嵌入式服务器的集成测试，这是用于单元测试的，也就是快速、简单的测试。

编辑

谢谢拉蒙·加西亚

对于在Google搜索中到达这里的人，请注意测试驱动程序类现在是org.apache.kafka.streams.topologyTestDriver

该类位于maven包groupId org.apache.kafka,artifactId kafka-streams-test-utils中

Answer 1

我找到了一个解决这个问题的方法，我不确定它是否是答案，特别是在https://stackoverflow.com/users/4953079/matthias-j-sax注释之后。无论如何，分享我目前所拥有的...

我完全从0.10.1分支（这是我使用的版本）复制了ProcessorTopologyTestDriver。

为了解决KAFKA-4408，我使私有最终模拟消费者 restorestateconsumer 可访问，并将块task=new StreamTask(...移到一个单独的方法，例如bootstrap。

在测试的设置阶段，我执行以下操作

driver = new ProcessorTopologyTestDriver(config, builder)
ArrayList partitionInfos = new ArrayList();
partitionInfos.add(new PartitionInfo('my_ktable', 1, (Node) null, (Node[]) null, (Node[]) null));
driver.restoreStateConsumer.updatePartitions('my_ktable', partitionInfos);
driver.restoreStateConsumer.updateEndOffsets(Collections.singletonMap(new TopicPartition('my_ktable', 1), Long.valueOf(0L)));
driver.bootstrap()

就这样...

奖金

我还遇到了KAFKA-4461，幸运的是，由于我复制了整个类，所以我能够通过一些小的调整来“挑选”已被接受的修复。

一如既往地感谢反馈。虽然显然不是一个正式的测试类，但这个驱动程序被证明是超级有用的！