[FLINK-4134] Retire Late Windows/Elements in WindowOperator

Before, when processing an element that would end up in a late window
(when using a MergingWindowAssigner), the element would be added to the
MergingWindowSet. After determining that the window is late it would not
be removed from the MergingWindowSet. This can lead to problems with
other elements being merged into these "phantom" windows and causing
triggers to be added for empty windows.

This also fixes the same code in EvictingWindowOperator.

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperator.java

@@ -133,6 +133,7 @@ public class EvictingWindowOperator<K, IN, OUT, W extends Window> extends Window
 				// check if the window is already inactive
 				if (isLate(actualWindow)) {
 					LOG.info("Dropped element " + element + " for window " + actualWindow + " due to lateness.");
+					mergingWindows.retireWindow(actualWindow);
 					continue;
 				}
 

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java

@@ -335,6 +335,7 @@ public class WindowOperator<K, IN, ACC, OUT, W extends Window>
 				// drop if the window is already late
 				if (isLate(actualWindow)) {
 					LOG.info("Dropped element " + element+ " for window " + actualWindow + " due to lateness.");
+					mergingWindows.retireWindow(actualWindow);
 					continue;
 				}
 

flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorTest.java

@@ -1477,6 +1477,8 @@ public class WindowOperatorTest {
 
 		// this is dropped as late
 		testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), 10000));
+		// this is also dropped as late (we test that they are not accidentally merged)
+		testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), 10100));
 
 		testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), 14500));
 		testHarness.processWatermark(new Watermark(20000));