Testing

    Usually, one can assume that Flink produces correct results outside of a user-defined function. Therefore, it is recommended to test those classes that contain the main business logic with unit tests as much as possible.

    For example, let’s take the following stateless MapFunction.

    Java

    Scala

    1. class IncrementMapFunction extends MapFunction[Long, Long] {
    3.     override def map(record: Long): Long = {
    4.         record + 1
    5.     }
    6. }

    It is very easy to unit test such a function with your favorite testing framework by passing suitable arguments and verifying the output.

    Java

    1. public class IncrementMapFunctionTest {
    3.     @Test
    4.     public void testIncrement() throws Exception {
    5.         // instantiate your function
    6.         IncrementMapFunction incrementer = new IncrementMapFunction();
    8.         // call the methods that you have implemented
    9.         assertEquals(3L, incrementer.map(2L));
    10.     }
    11. }

    Scala

    1. class IncrementMapFunctionTest extends FlatSpec with Matchers {
    3.     "IncrementMapFunction" should "increment values" in {
    4.         // instantiate your function
    5.         val incrementer: IncrementMapFunction = new IncrementMapFunction()
    7.         // call the methods that you have implemented
    8.         incremeter.map(2) should be (3)
    9.     }
    10. }

    Similarly, a user-defined function which uses an org.apache.flink.util.Collector (e.g. a FlatMapFunction or ProcessFunction) can be easily tested by providing a mock object instead of a real collector. A FlatMapFunction with the same functionality as the IncrementMapFunction could be unit tested as follows.

    Java

    1. public class IncrementFlatMapFunctionTest {
    3.     @Test
    4.     public void testIncrement() throws Exception {
    5.         // instantiate your function
    6.         IncrementFlatMapFunction incrementer = new IncrementFlatMapFunction();
    8.         Collector<Integer> collector = mock(Collector.class);
    10.         // call the methods that you have implemented
    11.         incrementer.flatMap(2L, collector);
    13.         //verify collector was called with the right output
    14.         Mockito.verify(collector, times(1)).collect(3L);
    15.     }
    16. }

    Scala

    1. class IncrementFlatMapFunctionTest extends FlatSpec with MockFactory {
    3.     "IncrementFlatMapFunction" should "increment values" in {
    4.        // instantiate your function
    5.       val incrementer : IncrementFlatMapFunction = new IncrementFlatMapFunction()
    7.       val collector = mock[Collector[Integer]]
    9.       //verify collector was called with the right output
    10.       (collector.collect _).expects(3)
    12.       // call the methods that you have implemented
    13.       flattenFunction.flatMap(2, collector)
    14.   }
    15. }

    Testing the functionality of a user-defined function, which makes use of managed state or timers is more difficult because it involves testing the interaction between the user code and Flink’s runtime. For this Flink comes with a collection of so called test harnesses, which can be used to test such user-defined functions as well as custom operators:

    • OneInputStreamOperatorTestHarness (for operators on DataStreams)
    • KeyedOneInputStreamOperatorTestHarness (for operators on KeyedStreams)
    • TwoInputStreamOperatorTestHarness (for operators of ConnectedStreams of two DataStreams)
    • KeyedTwoInputStreamOperatorTestHarness (for operators on ConnectedStreams of two KeyedStreams)

    To use the test harnesses a set of additional dependencies (test scoped) is needed.

    1. <dependency>
    2.     <groupId>org.apache.flink</groupId>
    3.     <artifactId>flink-test-utils_2.11</artifactId>
    4.     <version>1.14.4</version>    
    5.     <scope>test</scope>
    6. </dependency>

    Copied to clipboard!

    Copied to clipboard!

    1. <dependency>
    2.     <groupId>org.apache.flink</groupId>
    3.     <artifactId>flink-streaming-java_2.11</artifactId>
    4.     <version>1.14.4</version>    
    5.     <scope>test</scope>    
    6.     <classifier>tests</classifier>
    7. </dependency>

    Copied to clipboard!

    Java

    2. public class StatefulFlatMapTest {
    3.     private OneInputStreamOperatorTestHarness<Long, Long> testHarness;
    4.     private StatefulFlatMap statefulFlatMapFunction;
    6.     @Before
    7.     public void setupTestHarness() throws Exception {
    9.         //instantiate user-defined function
    10.         statefulFlatMapFunction = new StatefulFlatMapFunction();
    12.         // wrap user defined function into a the corresponding operator
    13.         testHarness = new OneInputStreamOperatorTestHarness<>(new StreamFlatMap<>(statefulFlatMapFunction));
    15.         // optionally configured the execution environment
    16.         testHarness.getExecutionConfig().setAutoWatermarkInterval(50);
    18.         // open the test harness (will also call open() on RichFunctions)
    19.         testHarness.open();
    20.     }
    22.     @Test
    24.         //push (timestamped) elements into the operator (and hence user defined function)
    27.         //trigger event time timers by advancing the event time of the operator with a watermark
    28.         testHarness.processWatermark(100L);
    30.         //trigger processing time timers by advancing the processing time of the operator directly
    31.         testHarness.setProcessingTime(100L);
    33.         //retrieve list of emitted records for assertions
    34.         assertThat(testHarness.getOutput(), containsInExactlyThisOrder(3L));
    36.         //retrieve list of records emitted to a specific side output for assertions (ProcessFunction only)
    37.         //assertThat(testHarness.getSideOutput(new OutputTag<>("invalidRecords")), hasSize(0))
    38.     }
    39. }

    Scala

    1. class StatefulFlatMapFunctionTest extends FlatSpec with Matchers with BeforeAndAfter {
    3.   private var testHarness: OneInputStreamOperatorTestHarness[Long, Long] = null
    4.   private var statefulFlatMap: StatefulFlatMapFunction = null
    6.   before {
    7.     //instantiate user-defined function
    8.     statefulFlatMap = new StatefulFlatMap
    10.     // wrap user defined function into a the corresponding operator
    11.     testHarness = new OneInputStreamOperatorTestHarness[Long, Long](new StreamFlatMap(statefulFlatMap))
    13.     // optionally configured the execution environment
    14.     testHarness.getExecutionConfig().setAutoWatermarkInterval(50);
    16.     // open the test harness (will also call open() on RichFunctions)
    17.     testHarness.open();
    18.   }
    20.   "StatefulFlatMap" should "do some fancy stuff with timers and state" in {
    23.     //push (timestamped) elements into the operator (and hence user defined function)
    24.     testHarness.processElement(2, 100);
    26.     //trigger event time timers by advancing the event time of the operator with a watermark
    27.     testHarness.processWatermark(100);
    29.     //trigger proccesign time timers by advancing the processing time of the operator directly
    30.     testHarness.setProcessingTime(100);
    32.     //retrieve list of emitted records for assertions
    33.     testHarness.getOutput should contain (3)
    35.     //retrieve list of records emitted to a specific side output for assertions (ProcessFunction only)
    36.     //testHarness.getSideOutput(new OutputTag[Int]("invalidRecords")) should have size 0
    37.   }
    38. }

    KeyedOneInputStreamOperatorTestHarness and KeyedTwoInputStreamOperatorTestHarness are instantiated by additionally providing a KeySelector including TypeInformation for the class of the key.

    Java

    2. public class StatefulFlatMapFunctionTest {
    3.     private OneInputStreamOperatorTestHarness<String, Long, Long> testHarness;
    4.     private StatefulFlatMap statefulFlatMapFunction;
    6.     @Before
    7.     public void setupTestHarness() throws Exception {
    9.         //instantiate user-defined function
    10.         statefulFlatMapFunction = new StatefulFlatMapFunction();
    12.         // wrap user defined function into a the corresponding operator
    13.         testHarness = new KeyedOneInputStreamOperatorTestHarness<>(new StreamFlatMap<>(statefulFlatMapFunction), new MyStringKeySelector(), Types.STRING);
    15.         // open the test harness (will also call open() on RichFunctions)
    16.         testHarness.open();
    17.     }
    19.     //tests
    21. }

    Scala

    1. class StatefulFlatMapTest extends FlatSpec with Matchers with BeforeAndAfter {
    3.   private var testHarness: OneInputStreamOperatorTestHarness[String, Long, Long] = null
    4.   private var statefulFlatMapFunction: FlattenFunction = null
    6.   before {
    7.     //instantiate user-defined function
    8.     statefulFlatMapFunction = new StateFulFlatMap
    10.     // wrap user defined function into a the corresponding operator
    11.     testHarness = new KeyedOneInputStreamOperatorTestHarness(new StreamFlatMap(statefulFlatMapFunction),new MyStringKeySelector(), Types.STRING())
    13.     // open the test harness (will also call open() on RichFunctions)
    14.     testHarness.open();
    15.   }
    17.   //tests
    19. }

    Many more examples for the usage of these test harnesses can be found in the Flink code base, e.g.:

    • org.apache.flink.streaming.runtime.operators.windowing.WindowOperatorTest is a good example for testing operators and user-defined functions, which depend on processing or event time.
    • org.apache.flink.streaming.api.functions.sink.filesystem.LocalStreamingFileSinkTest shows how to test a custom sink with the AbstractStreamOperatorTestHarness. Specifically, it uses AbstractStreamOperatorTestHarness.snapshot and AbstractStreamOperatorTestHarness.initializeState to tests its interaction with Flink’s checkpointing mechanism.

    Note Be aware that AbstractStreamOperatorTestHarness and its derived classes are currently not part of the public API and can be subject to change.

    Unit Testing ProcessFunction

    Given its importance, in addition to the previous test harnesses that can be used directly to test a ProcessFunction, Flink provides a test harness factory named ProcessFunctionTestHarnesses that allows for easier test harness instantiation. Considering this example:

    Note Be aware that to use this test harness, you also need to introduce the dependencies mentioned in the last section.

    Java

    1. public static class PassThroughProcessFunction extends ProcessFunction<Integer, Integer> {
    3.     @Override
    4.     public void processElement(Integer value, Context ctx, Collector<Integer> out) throws Exception {
    5.         out.collect(value);
    6.     }

    Scala

    It is very easy to unit test such a function with ProcessFunctionTestHarnesses by passing suitable arguments and verifying the output.

    Java

    3.     @Test
    4.     public void testPassThrough() throws Exception {
    6.         //instantiate user-defined function
    7.         PassThroughProcessFunction processFunction = new PassThroughProcessFunction();
    9.         // wrap user defined function into a the corresponding operator
    10.         OneInputStreamOperatorTestHarness<Integer, Integer> harness = ProcessFunctionTestHarnesses
    11.             .forProcessFunction(processFunction);
    13.         //push (timestamped) elements into the operator (and hence user defined function)
    14.         harness.processElement(1, 10);
    16.         //retrieve list of emitted records for assertions
    17.         assertEquals(harness.extractOutputValues(), Collections.singletonList(1));
    18.     }
    19. }

    Scala

    1. class PassThroughProcessFunctionTest extends FlatSpec with Matchers {
    3.   "PassThroughProcessFunction" should "forward values" in {
    5.     //instantiate user-defined function
    6.     val processFunction = new PassThroughProcessFunction
    8.     // wrap user defined function into a the corresponding operator
    9.     val harness = ProcessFunctionTestHarnesses.forProcessFunction(processFunction)
    11.     //push (timestamped) elements into the operator (and hence user defined function)
    12.     harness.processElement(1, 10)
    14.     //retrieve list of emitted records for assertions
    15.     harness.extractOutputValues() should contain (1)
    16.   }
    17. }

    For more examples on how to use the ProcessFunctionTestHarnesses in order to test the different flavours of the ProcessFunction, e.g. KeyedProcessFunction, KeyedCoProcessFunction, BroadcastProcessFunction, etc, the user is encouraged to look at the ProcessFunctionTestHarnessesTest.

    To use MiniClusterWithClientResource one additional dependency (test scoped) is needed.

    1. <dependency>
    2.     <groupId>org.apache.flink</groupId>
    3.     <artifactId>flink-test-utils_2.11</artifactId>
    4.     <version>1.14.4</version>    
    5.     <scope>test</scope>
    6. </dependency>

    Copied to clipboard!

    Let us take the same simple MapFunction as in the previous sections.

    Java

    1. public class IncrementMapFunction implements MapFunction<Long, Long> {
    3.     @Override
    4.     public Long map(Long record) throws Exception {
    5.         return record + 1;
    6.     }
    7. }

    Scala

    1. class IncrementMapFunction extends MapFunction[Long, Long] {
    3.     override def map(record: Long): Long = {
    4.         record + 1
    5.     }
    6. }

    A simple pipeline using this MapFunction can now be tested in a local Flink cluster as follows.

    Java

    1. public class ExampleIntegrationTest {
    3.      @ClassRule
    4.      public static MiniClusterWithClientResource flinkCluster =
    5.          new MiniClusterWithClientResource(
    6.              new MiniClusterResourceConfiguration.Builder()
    7.                  .setNumberSlotsPerTaskManager(2)
    8.                  .setNumberTaskManagers(1)
    9.                  .build());
    11.     @Test
    12.     public void testIncrementPipeline() throws Exception {
    13.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    15.         // configure your test environment
    16.         env.setParallelism(2);
    18.         // values are collected in a static variable
    19.         CollectSink.values.clear();
    21.         // create a stream of custom elements and apply transformations
    22.         env.fromElements(1L, 21L, 22L)
    23.                 .map(new IncrementMapFunction())
    24.                 .addSink(new CollectSink());
    26.         // execute
    27.         env.execute();
    29.         // verify your results
    30.         assertTrue(CollectSink.values.containsAll(2L, 22L, 23L));
    31.     }
    33.     // create a testing sink
    34.     private static class CollectSink implements SinkFunction<Long> {
    36.         // must be static
    37.         public static final List<Long> values = Collections.synchronizedList(new ArrayList<>());
    39.         @Override
    40.         public void invoke(Long value, SinkFunction.Context context) throws Exception {
    41.             values.add(value);
    42.         }
    43.     }
    44. }

    Scala

    A few remarks on integration testing with MiniClusterWithClientResource:

    • In order not to copy your whole pipeline code from production to test, make sources and sinks pluggable in your production code and inject special test sources and test sinks in your tests.

    • The static variable in CollectSink is used here because Flink serializes all operators before distributing them across a cluster. Communicating with operators instantiated by a local Flink mini cluster via static variables is one way around this issue. Alternatively, you could write the data to files in a temporary directory with your test sink.

    • You can implement a custom parallel source function for emitting watermarks if your job uses event time timers.

    • It is recommended to always test your pipelines locally with a parallelism > 1 to identify bugs which only surface for the pipelines executed in parallel.

    • If your pipeline contains custom state handling, you can test its correctness by enabling checkpointing and restarting the job within the mini cluster. For this, you need to trigger a failure by throwing an exception from (a test-only) user-defined function in your pipeline.