很简单，主要参考官方文档1、官方文档2和这篇文档。其中会补充我实践时使用的情况。
flink版本 1.11.2。本文例子来自官网，完整例子来自github

编写单元测试是设计生产应用程序的基本任务之一。如果不进行测试，那么一个很小的代码变更都会导致生产任务的失败。因此，无论是清理数据、模型训练的简单作业，还是复杂的多租户实时数据处理系统，我们都应该为所有类型的应用程序编写单元测试。下面我们将提供有关 Apache Flink 应用程序的单元测试指南。Apache Flink 提供了一个强大的单元测试框架，以确保我们的应用程序在上线后符合我们的预期。

1. Maven依赖

如果我们要使用 Apache Flink 提供的单元测试框架，我们需要引入如下依赖：

  1. <dependency>
  2.     <groupId>org.apache.flink</groupId>
  3.     <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
  4.     <version>${flink.version}</version>
  5.     <scope>test</scope>
  6.     <classifier>tests</classifier>
  7. </dependency>
  9. <dependency>
  10.     <groupId>org.apache.flink</groupId>
  11.     <artifactId>flink-test-utils_${scala.binary.version}</artifactId>
  12.     <version>${flink.version}</version>
  13.     <scope>test</scope>
  14. </dependency>
  16. <dependency>
  17.     <groupId>org.apache.flink</groupId>
  18.     <artifactId>flink-runtime_${scala.binary.version}</artifactId>
  19.     <version>${flink.version}</version>
  20.     <scope>test</scope>
  21.     <classifier>tests</classifier>
  22. </dependency>

注意：

  1. 由于需要测试 JAR 包：org.apache.flink:flink-runtime_2.11:tests:1.11.2 和 org.apache.flink:flink-streaming-java_2.11:tests:1.11.2，所以依赖需要制定 classifier 为 tests。
  2. 我使用的是官方推荐maven构建flink的项目，pom.xml文件中已经有了flink-streaming-java_${scala.binary.version}这个项，那么此时不能直接使用上面的依赖写法，会由冲突，编译很久或失败。问题的关键点在于 test，删除第一个即可。
  3. 构建单元测试时，需要在src下，复制一份main的目录结构，并将main改为test，大致如下。测试哪个文件，其测试文件名加个Test。

image.png

对于不同的算子，单元测试的编写也不一样。我们可以分为如下三种：

  • 无状态算子
  • 有状态算子
  • 定时处理算子（ProcessFunction）

    2. 无状态算子

    只有在使用 Test Harnesses 时，我们才需要上述 Maven 依赖，所以在编写无状态算子的单元测试时，可以不添加上述 Maven 依赖。

无状态算子的单元测试编写比较简单。我们只需要遵循编写测试用例的基本规范，即创建函数类的实例并测试适当的方法。我们以一个简单的 Map 算子为例：

  1. package com.flink.example.test;
  3. import org.apache.flink.api.common.functions.FlatMapFunction;
  4. import org.apache.flink.util.Collector;
  6. /**
  7.  * 无状态算子单元测试 FlatMap
  8.  * Created by wy on 2020/11/8.
  9.  */
  10. public class MyStatelessMap implements MapFunction<String, String> {
  11.   @Override
  12.   public String map(String in) throws Exception {
  13.     String out = "hello " + in;
  14.     return out;
  15.   }
  16. }

还有一个无状态的FlatMap算子：

  1. package com.flink.example.test;
  3. import org.apache.flink.api.common.functions.FlatMapFunction;
  4. import org.apache.flink.util.Collector;
  6. public class MyStatelessFlatMap implements FlatMapFunction<String, String> {
  7.     @Override
  8.     public void flatMap(String s, Collector<String> collector) throws Exception {
  9.         String out = "hello " + s;
  10.         collector.collect(out);
  11.     }
  12. }

上述两个算子的单元测试用例如下所示：

  1. package com.flink.example.test;
  3. import com.google.common.collect.Lists;
  4. import org.apache.flink.api.common.functions.util.ListCollector;
  5. import org.junit.Assert;
  6. import org.junit.Test;
  8. import java.util.ArrayList;
  9. import java.util.List;
  11. public class StatelessUnitTest {
  12.     @Test
  13.     public void MyStatelessMap() throws Exception {
  14.         MyStatelessMap statelessMap = new MyStatelessMap();
  15.         String out = statelessMap.map("world");
  16.         Assert.assertEquals("hello world", out);
  17.     }
  19.     @Test
  20.     public void MyStatelessFlatMap() throws Exception {
  21.         MyStatelessFlatMap statelessFlatMap = new MyStatelessFlatMap();
  22.         List<String> out = new ArrayList<>();
  23.         ListCollector<String> listCollector = new ListCollector<>(out);
  24.         statelessFlatMap.flatMap("world", listCollector);
  25.         Assert.assertEquals(Lists.newArrayList("hello world"), out);
  26.     }
  27. }

FlatMap 算子需要一个 Collector 对象以及一个输入参数。编写测试用例，我们有如下两种方式：

  • 使用 Mockito 模拟 Collector 对象
  • 使用 Flink 提供的 ListCollector

我更喜欢第二种方法，因为只需要很少的代码即可，并且适合大多数情况。

3. 有状态算子

测试有状态算子（使用状态或者定时器）会比较困难。因为用户代码需要与 Flink Runtime 进行交互。为此，Flink 提供了一组 TestHarness，可用于测试用户定义的函数以及自定义算子：

  • OneInputStreamOperatorTestHarness：适用于 DataStreams 上的算子
  • KeyedOneInputStreamOperatorTestHarness：适用于 KeyedStreams 上的算子
  • TwoInputStreamOperatorTestHarness：用于两个数据流的 ConnectedStream 的算子
  • KeyedTwoInputStreamOperatorTestHarness：用于两个 KeyedStream 的 ConnectedStream 上的算子

我们以有状态的 FlatMap 函数为例：

  1. package com.flink.example.test;
  3. import org.apache.flink.api.common.functions.RichFlatMapFunction;
  4. import org.apache.flink.api.common.state.ValueState;
  5. import org.apache.flink.api.common.state.ValueStateDescriptor;
  6. import org.apache.flink.api.common.typeinfo.Types;
  7. import org.apache.flink.configuration.Configuration;
  8. import org.apache.flink.util.Collector;
  10. public class MyStatefulFlatMap extends RichFlatMapFunction<String, Long> {
  11.     ValueState<Long> counterState;
  13.     @Override
  14.     public void open(Configuration parameters) throws Exception {
  15.         ValueStateDescriptor<Long> descriptor = new ValueStateDescriptor<>(
  16.                 "Counter",
  17.                 Types.LONG
  18.         );
  19.         this.counterState = getRuntimeContext().getState(descriptor);
  20.     }
  22.     @Override
  23.     public void flatMap(String s, Collector<Long> collector) throws Exception {
  24.         Long count = 0L;
  25.         if (this.counterState.value() != null) {
  26.             count = this.counterState.value();
  27.         }
  28.         count ++;
  29.         this.counterState.update(count);
  30.         collector.collect(count);
  31.     }
  32. }

编写上述类的单元测试最复杂部分是模拟应用程序的配置以及运行时上下文。我们使用 Flink 提供的 TestHarness 类，这样我们就不必自己创建模拟对象。使用 KeyedOperatorHarness 类进行单元测试如下所示：

  1. package com.flink.example.test;
  3. import com.google.common.collect.Lists;
  4. import org.apache.flink.api.common.typeinfo.Types;
  5. import org.apache.flink.streaming.api.operators.StreamFlatMap;
  6. import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
  7. import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness;
  8. import org.junit.Assert;
  9. import org.junit.Before;
  10. import org.junit.Test;
  12. public class MyStatefullFlatMapUnitTest {
  14.     private KeyedOneInputStreamOperatorTestHarness<String, String, Long>  testHarness;
  15.     private MyStatefulFlatMap statefulFlatMap;
  17.     @Before
  18.     public void setupTestHarness() throws Exception {
  19.         statefulFlatMap = new MyStatefulFlatMap();
  20.         // KeyedOneInputStreamOperatorTestHarness 需要三个参数：算子对象、键 Selector、键类型
  21.         testHarness = new KeyedOneInputStreamOperatorTestHarness<>(
  22.                 new StreamFlatMap<>(statefulFlatMap),
  23.                 x -> "1",
  24.                 Types.STRING
  25.         );
  26.         testHarness.open();
  27.     }
  29.     @Test
  30.     public void MyStatefulFlatMap() throws Exception{
  31.         // test first record
  32.         testHarness.processElement("a", 10);
  33.         //
  34.         Assert.assertEquals(
  35.                 Lists.newArrayList(new StreamRecord<>(1L, 10)),
  36.                 this.testHarness.extractOutputStreamRecords()
  37.         );
  39.         // test second record
  40.         testHarness.processElement("b", 20);
  41.         Assert.assertEquals(
  42.                 Lists.newArrayList(
  43.                         new StreamRecord<>(1L, 10),
  44.                         new StreamRecord<>(2L, 20)
  45.                 ),
  46.                 testHarness.extractOutputStreamRecords()
  47.         );
  49.         // test other record
  50.         testHarness.processElement("c", 30);
  51.         testHarness.processElement("d", 40);
  52.         testHarness.processElement("e", 50);
  53.         Assert.assertEquals(
  54.                 Lists.newArrayList(
  55.                         new StreamRecord<>(1L, 10),
  56.                         new StreamRecord<>(2L, 20),
  57.                         new StreamRecord<>(3L, 30),
  58.                         new StreamRecord<>(4L, 40),
  59.                         new StreamRecord<>(5L, 50)
  60.                 ),
  61.                 testHarness.extractOutputStreamRecords()
  62.         );
  63.     }
  64. }

KeyedOneInputStreamOperatorTestHarness 中参数数据类型依次为：key的类型，后两个是MyStatefulFlatMap extends RichFlatMapFunction 的Input和Output.

TestHarness 提供了许多辅助方法，上述代码使用了其中三种：

  • open：使用相关参数调用 FlatMap 函数的 open 方法，同时还会对上下文进行初始化。
  • processElement：允许用户传入输入元素以及与该元素关联的时间戳。
  • extractOutputStreamRecords：从 Collector 获取输出记录以及时间戳。

TestHarness 极大地简化了有状态算子的单元测试。

4. 定时处理算子

为与时间有关的 Process Function 编写单元测试与为有状态算子编写单元测试非常相似，我们都需要使用 TestHarness。但是，在这我们还需要考虑另一个方面，即为事件提供时间戳并控制应用程序的当前时间。通过设置当前（处理时间或事件时间）时间，我们可以触发注册的计时器，并调用该函数的 onTimer 方法：

  1. package com.flink.example.test;
  3. import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
  4. import org.apache.flink.util.Collector;
  6. public class TimerProcessFunction extends KeyedProcessFunction<String, String, String> {
  7.     @Override
  8.     public void processElement(String s, Context context, Collector<String> collector) throws Exception {
  9.         context.timerService().registerProcessingTimeTimer(50);
  10.         String out = "hello " + s;
  11.         collector.collect(out);
  12.     }
  14.     @Override
  15.     public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
  16.         // 到达时间点触发事件操作
  17.         out.collect(String.format("Timer triggered at timestamp %d", timestamp));
  18.     }

我们需要测试 KeyedProcessFunction 中的两个方法，即 processElement 和 onTimer 方法。使用 TestHarness，我们可以控制函数的当前时间。因此，我们可以随意触发计时器，而不必等待特定的时间：

  1. package com.flink.example.test;
  3. import com.google.common.collect.Lists;
  4. import org.apache.flink.api.common.typeinfo.Types;
  5. import org.apache.flink.streaming.api.operators.KeyedProcessOperator;
  6. import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
  7. import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness;
  8. import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
  9. import org.junit.Assert;
  10. import org.junit.Before;
  11. import org.junit.Test;
  13. public class TimerProcessFunctionUnitTest {
  15.     private OneInputStreamOperatorTestHarness<String, String> testHarness;
  16.     private TimerProcessFunction processFunction;
  18.     @Before
  19.     public void setupTestHarness() throws Exception {
  20.         processFunction = new TimerProcessFunction();
  22.         // KeyedOneInputStreamOperatorTestHarness 需要三个参数：算子对象、键 Selector、键类型
  23.         testHarness = new KeyedOneInputStreamOperatorTestHarness<>(
  24.                 new KeyedProcessOperator<>(processFunction),
  25.                 x -> "1",
  26.                 Types.STRING
  27.         );
  28.         // Function time is initialized to 0
  29.         testHarness.open();
  30.     }
  32.     @Test
  33.     public void testProcessElement() throws Exception{
  35.         testHarness.processElement("world", 10);
  36.         Assert.assertEquals(
  37.                 Lists.newArrayList(
  38.                         new StreamRecord<>("hello world", 10)
  39.                 ),
  40.                 testHarness.extractOutputStreamRecords()
  41.         );
  42.     }
  44.     @Test
  45.     public void testOnTimer() throws Exception {
  46.         // test first record
  47.         testHarness.processElement("world", 10);
  48.         Assert.assertEquals(1, testHarness.numProcessingTimeTimers());
  50.         // Function time 设置为 50
  51.         testHarness.setProcessingTime(50);
  52.         Assert.assertEquals(
  53.                 Lists.newArrayList(
  54.                         new StreamRecord<>("hello world", 10),
  55.                         new StreamRecord<>("Timer triggered at timestamp 50")
  56.                 ),
  57.                 testHarness.extractOutputStreamRecords()
  58.         );
  59.     }
  60. }

考虑到 ProcessFunction 的重要性，除了上面可以直接用于测试 ProcessFunction 的 TestHarness 之外，Flink 还提供了一个名为 ProcessFunctionTestHarnesses 的 TestHarness 工厂，可以大大简化 TestHarness 的实例化。如下所示：

  1. package com.flink.example.test;
  3. import org.apache.flink.streaming.api.functions.ProcessFunction;
  4. import org.apache.flink.util.Collector;
  7. public class MyProcessFunction extends ProcessFunction<Integer, Integer> {
  8.     @Override
  9.     public void processElement(Integer integer, Context context, Collector<Integer> collector) throws Exception {
  10.         collector.collect(integer);
  11.     }
  12. }

通过传递适当的参数并验证输出，使用 ProcessFunctionTestHarnesses 进行单元测试会更加容易：

  1. package com.flink.example.test;
  3. import com.google.common.collect.Lists;
  4. import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
  5. import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
  6. import org.apache.flink.streaming.util.ProcessFunctionTestHarnesses;
  7. import org.junit.Assert;
  8. import org.junit.Test;
  11. public class ProcessFunctionUnitTest {
  12.     @Test
  13.     public void testPassThrough() throws Exception {
  15.         MyProcessFunction processFunction = new MyProcessFunction();
  17.         OneInputStreamOperatorTestHarness<Integer, Integer> testHarness = ProcessFunctionTestHarnesses
  18.                 .forProcessFunction(processFunction);
  20.         testHarness.processElement(1, 10);
  21.         Assert.assertEquals(
  22.                 Lists.newArrayList(
  23.                         new StreamRecord<>(1, 10)
  24.                 ),
  25.                 testHarness.extractOutputStreamRecords()
  26.         );
  27.     }
  28. }

有关如何使用 ProcessFunctionTestHarnesses 测试 ProcessFunction 不同风味（例如 KeyedProcessFunction，KeyedCoProcessFunction，BroadcastProcessFunction等）的更多示例，，可以参阅 ProcessFunctionTestHarnessesTest：

  1. public static <IN, OUT> OneInputStreamOperatorTestHarness<IN, OUT> forProcessFunction(ProcessFunction<IN, OUT> function) throws Exception {
  2. }
  4. public static <K, IN, OUT> KeyedOneInputStreamOperatorTestHarness<K, IN, OUT> forKeyedProcessFunction(KeyedProcessFunction<K, IN, OUT> function, KeySelector<IN, K> keySelector, TypeInformation<K> keyType) throws Exception {
  5. }
  7. public static <IN1, IN2, OUT> TwoInputStreamOperatorTestHarness<IN1, IN2, OUT> forCoProcessFunction(CoProcessFunction<IN1, IN2, OUT> function) throws Exception {
  8. }
  10. public static <K, IN1, IN2, OUT> KeyedTwoInputStreamOperatorTestHarness<K, IN1, IN2, OUT> forKeyedCoProcessFunction(KeyedCoProcessFunction<K, IN1, IN2, OUT> function, KeySelector<IN1, K> keySelector1, KeySelector<IN2, K> keySelector2, TypeInformation<K> keyType) throws Exception {
  11. }
  13. public static <IN1, IN2, OUT> BroadcastOperatorTestHarness<IN1, IN2, OUT> forBroadcastProcessFunction(BroadcastProcessFunction<IN1, IN2, OUT> function, MapStateDescriptor... descriptors) throws Exception {
  14. }
  16. public static <K, IN1, IN2, OUT> KeyedBroadcastOperatorTestHarness<K, IN1, IN2, OUT> forKeyedBroadcastProcessFunction(KeyedBroadcastProcessFunction<K, IN1, IN2, OUT> function, KeySelector<IN1, K> keySelector, TypeInformation<K> keyType, MapStateDescriptor... descriptors) throws Exception {
  17. }

另一个完整的单元测试样例：

  1. package com.flink.example.test;
  3. import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
  4. import org.apache.flink.api.common.typeinfo.TypeInformation;
  5. import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
  6. import org.apache.flink.streaming.api.functions.ProcessFunction;
  7. import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
  8. import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
  9. import org.apache.flink.streaming.util.KeyedTwoInputStreamOperatorTestHarness;
  10. import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
  11. import org.apache.flink.streaming.util.ProcessFunctionTestHarnesses;
  12. import org.apache.flink.streaming.util.TwoInputStreamOperatorTestHarness;
  13. import org.apache.flink.util.Collector;
  14. import org.junit.Assert;
  15. import org.junit.Test;
  17. import java.util.Arrays;
  18. import java.util.Collections;
  20. /**
  21.  * ProcessFunction 单元测试示例
  22.  * Created by wy on 2020/11/9.
  23.  */
  24. public class ProcessFunctionTestHarnessesTest {
  25.     @Test
  26.     public void testHarnessForProcessFunction() throws Exception {
  27.         ProcessFunction<Integer, Integer> function = new ProcessFunction<Integer, Integer> () {
  29.             @Override
  30.             public void processElement(
  31.                     Integer value, Context ctx, Collector<Integer> out) throws Exception {
  32.                 out.collect(value);
  33.             }
  34.         };
  35.         OneInputStreamOperatorTestHarness<Integer, Integer> harness = ProcessFunctionTestHarnesses
  36.                 .forProcessFunction(function);
  38.         harness.processElement(1, 10);
  40.         Assert.assertEquals(harness.extractOutputValues(), Collections.singletonList(1));
  41.     }
  43.     @Test
  44.     public void testHarnessForKeyedProcessFunction() throws Exception {
  45.         KeyedProcessFunction<Integer, Integer, Integer> function = new KeyedProcessFunction<Integer, Integer, Integer>() {
  46.             @Override
  47.             public void processElement(Integer value, Context ctx, Collector<Integer> out) throws Exception {
  48.                 out.collect(value);
  49.             }
  50.         };
  51.         OneInputStreamOperatorTestHarness<Integer, Integer> harness = ProcessFunctionTestHarnesses
  52.                 .forKeyedProcessFunction(function, x -> x, BasicTypeInfo.INT_TYPE_INFO);
  54.         harness.processElement(1, 10);
  56.         Assert.assertEquals(harness.extractOutputValues(), Collections.singletonList(1));
  57.     }
  59.     @Test
  60.     public void testHarnessForCoProcessFunction() throws Exception {
  61.         CoProcessFunction<Integer, String, Integer> function = new CoProcessFunction<Integer, String, Integer>() {
  63.             @Override
  64.             public void processElement1(Integer value, Context ctx, Collector<Integer> out) throws Exception {
  65.                 out.collect(value);
  66.             }
  68.             @Override
  69.             public void processElement2(String value, Context ctx, Collector<Integer> out) throws Exception {
  70.                 out.collect(Integer.parseInt(value));
  71.             }
  72.         };
  73.         TwoInputStreamOperatorTestHarness<Integer, String, Integer> harness = ProcessFunctionTestHarnesses
  74.                 .forCoProcessFunction(function);
  76.         harness.processElement2("0", 1);
  77.         harness.processElement1(1, 10);
  79.         Assert.assertEquals(harness.extractOutputValues(), Arrays.asList(0, 1));
  80.     }
  82.     @Test
  83.     public void testHarnessForKeyedCoProcessFunction() throws Exception {
  84.         KeyedCoProcessFunction<Integer, Integer, Integer, Integer> function = new KeyedCoProcessFunction<Integer, Integer, Integer, Integer>() {
  86.             @Override
  87.             public void processElement1(Integer value, Context ctx, Collector<Integer> out) throws Exception {
  88.                 out.collect(value);
  89.             }
  91.             @Override
  92.             public void processElement2(Integer value, Context ctx, Collector<Integer> out) throws Exception {
  93.                 out.collect(value);
  94.             }
  95.         };
  97.         KeyedTwoInputStreamOperatorTestHarness<Integer, Integer, Integer, Integer> harness = ProcessFunctionTestHarnesses
  98.                 .forKeyedCoProcessFunction(function, x -> x, x -> x, TypeInformation.of(Integer.class));
  100.         harness.processElement1(0, 1);
  101.         harness.processElement2(1, 10);
  103.         Assert.assertEquals(harness.extractOutputValues(), Arrays.asList(0, 1));
  104.     }
  105. }