[hotfix][streaming] Reactive sanity check for windowOperator's mergea…

…ble state.

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

@@ -240,15 +240,10 @@ public long getCurrentProcessingTime() {
             // store a typed reference for the state of merging windows - sanity check
             if (windowState instanceof InternalMergingState) {
                 windowMergingState = (InternalMergingState<K, W, IN, ACC, ACC>) windowState;
+            } else if (windowState != null) {
+                throw new IllegalStateException(
+                        "The window uses a merging assigner, but the window state is not mergeable.");
             }
-            // TODO this sanity check should be here, but is prevented by an incorrect test (pending
-            // validation)
-            // TODO see WindowOperatorTest.testCleanupTimerWithEmptyFoldingStateForSessionWindows()
-            // TODO activate the sanity check once resolved
-            //			else if (windowState != null) {
-            //				throw new IllegalStateException(
-            //						"The window uses a merging assigner, but the window state is not mergeable.");
-            //			}
 
             @SuppressWarnings("unchecked")
             final Class<Tuple2<W, W>> typedTuple = (Class<Tuple2<W, W>>) (Class<?>) Tuple2.class;