Usually, one can assume that Flink produces correct results outside of a user-defined Function. Therefore, it is recommended to test Function classes that contain the main business logic with unit tests as much as possible.
For example, if one implements the following ReduceFunction:
It is very easy to unit test it with your favorite framework by passing suitable arguments and verify the output:
In order to end-to-end test Flink streaming pipelines, you can also write integration tests that are executed against a local Flink mini cluster.
In order to do so add the test dependency flink-test-utils:
For example, if you want to test the following MapFunction:
You could write the following integration test:
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, for example, write the data to files in a temporary directory with your test sink.
You can also implement your own custom sources for emitting watermarks.
Testing checkpointing and state handling
One way to test state handling is to enable checkpointing in integration tests.
You can do that by configuring your StreamExecutionEnvironment in the test:
And for example, adding to your Flink application an identity mapper operator that will throw an exception
once every 1000ms. However, writing such a test could be tricky because of time dependencies between the actions.
Another approach is to write a unit test using the Flink internal testing utility AbstractStreamOperatorTestHarness from the flink-streaming-java module.
For an example of how to do that please have a look at the org.apache.flink.streaming.runtime.operators.windowing.WindowOperatorTest also in the flink-streaming-java module.
Be aware that AbstractStreamOperatorTestHarness is currently not a part of public API and can be subject to change.