Testing Flink job with bounded input

[Jan Lukavský <je...@seznam.cz>]

Hi,

I have a question about how to correctly set up a test that will read 
input from locally provided collection in bounded mode and provide 
outputs at the end of the computation. My test case looks something like 
the following:

String[] lines = ...;
try (StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment()) {
   env.setRuntimeMode(RuntimeExecutionMode.BATCH); env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);DataStream<String> input = env.fromCollection(Arrays.asList(lines)); // the following includes keyBy().window().aggregate() DataStream<Iterable<Tuple2<String, Long>>> output = TopWords.getTopWords(input, 5); // sink that collects outputs to (static) list SinkFunction<Iterable<Tuple2<String, Long>>> sink = TestSink.create(); output.addSink(sink); env.execute(); List<Tuple2<String, Long>> output =
       TestSink.outputs(sink).stream().flatMap(Streams::stream).collect(Collectors.toList()); // output is empty }

When I remove any window() and aggregations I can see outputs at the 
sink. I tried create manual trigger for the window, I can see 
onElement() methods called, but onEventTime() is never called. Also the 
FromElementsFunction never explicitly emits max watermark at the end 
reading the input. Seems that Flink runtime also does not emit the final 
watermark (I cannot see any traces of any watermark emission in the 
logs). Seems that I'm doing something obviously wrong, I just cannot 
figure out how to emit the final watermark. Adding autoWatermarkInterval 
did not help either.

The test seems to behave the same for Flink versions 1.16.3 and 1.18.1.

Thanks in advance for any pointers.

  Jan