Flink test late elements

[Dario Heinisch <da...@gmail.com>]

Hey there,

Hope everyone is well!

I have a question:
```
StreamExecutionEnvironment env = 
StreamExecutionEnvironment.getExecutionEnvironment();
         env.setParallelism(1);

         DataStream<Integer> dataStream = env.addSource(new CustomSource());

         OutputTag<Integer> outputTag = new OutputTag<Integer>("late") {};

         WatermarkStrategy<Integer> waStrat = WatermarkStrategy
                 .<Integer>forMonotonousTimestamps()
                 .withTimestampAssigner((i, timestamp) -> Long.valueOf(i));

         SingleOutputStreamOperator<Integer> windowOperator =
                 dataStream
                         .assignTimestampsAndWatermarks(waStrat)
.windowAll(TumblingEventTimeWindows.of(Time.milliseconds(1)))
                         .sideOutputLateData(outputTag)
                         .apply(
                                 new AllWindowFunction<Integer, Integer, 
TimeWindow>() {
                                     @Override
                                     public void apply(TimeWindow 
window, Iterable<Integer> values, Collector<Integer> out) {
System.out.println(window.getStart() + " -> " + window.getEnd());
                                         for (Integer val : values) {
                                             System.out.println(val + " 
in window ");
                                         }
                                     }
                                 });

         windowOperator
                 .getSideOutput(outputTag)
                 .flatMap(
                         new FlatMapFunction<Integer, String>() {
                             @Override
                             public void flatMap(Integer value, 
Collector<String> out) {
                                 System.out.println("LATE: " + value);
                             }
                         });

         env.execute();
```

And my custom source:
```
static class CustomSource implements SourceFunction<Integer> {

         @Override
         public void run(SourceContext<Integer> ctx) throws Exception {
             int i = 0;
             while (i++ < 10) {
                 Thread.sleep(100);
                 if (i < 5) {
                     ctx.collect(i * 10);
                 } else {
                     ctx.collect(i);
                 }

             }
         }

         @Override
         public void cancel() {

         }
     }
```

Now in the source if I leave `Thread.sleep(100)` in there I get the 
following output:
```
10 -> 11
10 in window
20 -> 21
20 in window
LATE: 5
30 -> 31
30 in window
LATE: 6
LATE: 7
LATE: 8
LATE: 9
LATE: 10
40 -> 41
40 in window
```

If I comment out the sleep, I get:
```
5 -> 6
5 in window
6 -> 7
6 in window
7 -> 8
7 in window
8 -> 9
8 in window
9 -> 10
9 in window
10 -> 11
10 in window
10 in window
20 -> 21
20 in window
30 -> 31
30 in window
40 -> 41
40 in window
```

I thought by using `forMonotonousTimestamps` it would automatically drop 
the late events from the window and emit them to the side output since 
the first 4 elements will have a higher timestamp than the last 5 but it 
seems
it evaluates all elements emitted by the source at once and then emit 
the watermark after one batch of events by the source has been processed.

I assume there would be a better approach then to sleep in between 
collecting in the source for getting the expected result?

( The background here is that I want to write tests for some event time 
processing and assert that the SideOutput will be used for late events 
and these are being
processed correctly. In production the Kafka source will not always emit 
data continuously, hence I want to test this but I have not found 
anything better than sleeping, maybe that is the correct approach?)

Best regards,

Dario