测试Flink-Kafka集成

Question

如何设置一个单元测试环境，在这个环境中我可以创建Kafka主题，使用FlinkKafkaProducer对它们进行写入，并使用FlinkKafkaConsumer对它们进行读取？

看起来功能是存在的，但没有很好的文档记录。

我所能得到的最远的结果如下

val server = new KafkaTestEnvironmentImpl()
server.prepare(KafkaTestEnvironment.createConfig().setKafkaServersNumber(1))
server.createTestTopic("my.test.topic", 1, 1)
server.shutdown()

但是这会抛出一个错误java.lang.NoClassDefFoundError: kafka/admin/RackAwareMode。我不确定尝试解决此错误是最好的做法，还是有更简单的解决方案。

Answer 1

我强烈建议用户使用基于docker的方法，例如testcontainers。Kafka的设置非常简单。

如果您需要注入数据，可以使用another framework轻松维护主题和添加数据。如果可以与testcontainers一起使用，或者如果你真的只需要Kafka作为外部系统，它甚至可以完全取代它。