Flink 中复杂拓扑（多输入）的集成测试答案

【问题标题】：Integration test for complex topology (multiple inputs) in FlinkFlink 中复杂拓扑（多输入）的集成测试
【发布时间】：2018-09-03 17:04:05
【问题描述】：

我需要为 flink 流式拓扑编写单元测试。它基本上是CoFlatMapFunction，它有 2 个输入。

我试图从这个页面获得一些灵感：https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/testing.html

输入的顺序对我的拓扑很重要，所以当我测试时，我不能对每个输入使用StreamExecutionEnvironment#fromCollection，因为我无法控制在每个输入中注入数据点的顺序。

我尝试使用 StreamExecutionEnvironment#fromCollection 创建单个输入，并根据它们的类型将每个元素分派到我的 CoFlatMapFunction 的实际输入，但在此操作中元素的顺序丢失了。

还有其他方法可以编写这个测试吗？

【问题讨论】：

标签： java unit-testing apache-flink flink-streaming

【解决方案1】：

flink 训练练习中有一个使用 TwoInputStreamOperatorTestHarness 的例子，可以参考：

https://github.com/dataArtisans/flink-training-exercises/blob/master/src/test/java/com/dataartisans/flinktraining/exercises/datastream_java/process/EventTimeJoinTest.java

您将需要这些依赖项：

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-test-utils-junit</artifactId>
  <version>${flink.version}</version>
</dependency>

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-streaming-java_2.11</artifactId>
  <version>${flink.version}</version>
  <scope>test</scope>
  <type>test-jar</type>
</dependency>

<dependency>
  <groupId>org.mockito</groupId>
  <artifactId>mockito-all</artifactId>
  <version>1.10.19</version>
  <type>jar</type>
  <scope>test</scope>
</dependency>

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-runtime_2.11</artifactId>
  <version>${flink.version}</version>
  <scope>test</scope>
  <type>test-jar</type>
</dependency>

您应该记住，这不是一个受支持的公共接口，因此它可能会以意想不到的方式发展。

【讨论】：

【解决方案2】：

您想使用TwoInputStreamOperatorTestHarness 类。不幸的是，文档有点稀疏。我有一个使用这个类的测试，但它还没有推送到flink-crawler 的133_stream-test-harness 分支。

【讨论】：