scala - 测试kafka和flink集成流程

Question

例如，我想测试 Kafka/Flink 与 FlinkKafkaConsumer011 和 FlinkKafkaProducer011 的集成。

过程将是:

1. 使用 Flink 读取 kafka 主题
2. 使用 Flink 进行一些操作
3. 用 Flink 写入另一个 kafka 主题

以字符串为例，从输入主题中读取字符串，转换为大写，写入新主题。

问题是如何测试流量？

当我说测试时，这是单元/集成测试。

谢谢!

Answer 1

Flink 文档中有一个关于如何为转换运算符编写单元\集成测试的小文档:link .该文档还有一小节内容是关于测试检查点和状态处理，以及关于使用 AbstractStreamOperatorTestHarness。

不过，我认为您对端到端集成测试(包括测试源和接收器)更感兴趣。为此，您可以启动一个 Flink 迷你集群。这是启动 Flink 迷你集群的示例代码的链接:link .

您还可以在 JVM 中启动 Kafka Broker 并将其用于测试目的。 Flink 的 Kafka 连接器为集成测试做这件事。这是启动 Kafka 服务器的示例代码:link .

如果您在本地运行，您可以使用一个简单的生成器应用程序为您的源 Kafka 主题生成消息(有很多可用的。您可以连续生成消息或根据不同的配置间隔生成消息)。下面是一个关于如何在本地运行时设置 Flink 作业全局参数的示例:Kafka010Example .

另一种选择是创建一个集成环境(相对于生产环境)来运行端到端测试。您将能够真正了解您的程序在类生产环境中的行为方式。始终建议拥有一个完整的并行测试环境 - 包括测试源\接收器 Kafka 主题。