[FLINK-3200] Fix Triggers by introducing clear() method to clean up state/triggers

flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/windowing/SessionWindowing.java

@@ -45,7 +45,7 @@ public class SessionWindowing {
 
 		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 		env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
-		env.setParallelism(2);
+		env.setParallelism(1);
 
 		final List<Tuple3<String, Long, Integer>> input = new ArrayList<>();
 
@@ -103,7 +103,7 @@ public class SessionWindowing {
 
 		private final Long sessionTimeout;
 
-		private final ValueStateDescriptor<Long> stateDesc = new ValueStateDescriptor<>("last-seen", 1L,
+		private final ValueStateDescriptor<Long> stateDesc = new ValueStateDescriptor<>("last-seen", -1L,
 			BasicTypeInfo.LONG_TYPE_INFO.createSerializer(new ExecutionConfig()));
 
 
@@ -120,12 +120,15 @@ public class SessionWindowing {
 
 			Long timeSinceLastEvent = timestamp - lastSeen;
 
+			ctx.deleteEventTimeTimer(lastSeen + sessionTimeout);
+
 			// Update the last seen event time
 			lastSeenState.update(timestamp);
 
 			ctx.registerEventTimeTimer(timestamp + sessionTimeout);
 
-			if (timeSinceLastEvent > sessionTimeout) {
+			if (lastSeen != -1 && timeSinceLastEvent > sessionTimeout) {
+				System.out.println("FIRING ON ELEMENT: " + element + " ts: " + timestamp + " last " + lastSeen);
 				return TriggerResult.FIRE_AND_PURGE;
 			} else {
 				return TriggerResult.CONTINUE;
@@ -138,6 +141,7 @@ public class SessionWindowing {
 			Long lastSeen = lastSeenState.value();
 
 			if (time - lastSeen >= sessionTimeout) {
+				System.out.println("CTX: " + ctx + " Firing Time " + time + " last seen " + lastSeen);
 				return TriggerResult.FIRE_AND_PURGE;
 			}
 			return TriggerResult.CONTINUE;
@@ -147,6 +151,15 @@ public class SessionWindowing {
 		public TriggerResult onProcessingTime(long time, GlobalWindow window, TriggerContext ctx) throws Exception {
 			return TriggerResult.CONTINUE;
 		}
+
+		@Override
+		public void clear(GlobalWindow window, TriggerContext ctx) throws Exception {
+			ValueState<Long> lastSeenState = ctx.getPartitionedState(stateDesc);
+			if (lastSeenState.value() != -1) {
+				ctx.deleteEventTimeTimer(lastSeenState.value() + sessionTimeout);
+			}
+			lastSeenState.clear();
+		}
 	}
 
 	// *************************************************************************

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/GlobalWindows.java

@@ -87,6 +87,9 @@ public class GlobalWindows extends WindowAssigner<Object, GlobalWindow> {
 		public TriggerResult onProcessingTime(long time, GlobalWindow window, TriggerContext ctx) {
 			return TriggerResult.CONTINUE;
 		}
+
+		@Override
+		public void clear(GlobalWindow window, TriggerContext ctx) throws Exception {}
 	}
 
 	@Override

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousEventTimeTrigger.java

@@ -73,6 +73,9 @@ public class ContinuousEventTimeTrigger<W extends Window> implements Trigger<Obj
 		return TriggerResult.CONTINUE;
 	}
 
+	@Override
+	public void clear(W window, TriggerContext ctx) throws Exception {}
+
 	@Override
 	public String toString() {
 		return "ContinuousProcessingTimeTrigger(" + interval + ")";

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousProcessingTimeTrigger.java

@@ -90,6 +90,9 @@ public class ContinuousProcessingTimeTrigger<W extends Window> implements Trigge
 		return TriggerResult.CONTINUE;
 	}
 
+	@Override
+	public void clear(W window, TriggerContext ctx) throws Exception {}
+
 	@VisibleForTesting
 	public long getInterval() {
 		return interval;

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/CountTrigger.java

@@ -65,6 +65,11 @@ public class CountTrigger<W extends Window> implements Trigger<Object, W> {
 		return TriggerResult.CONTINUE;
 	}
 
+	@Override
+	public void clear(W window, TriggerContext ctx) throws Exception {
+		ctx.getPartitionedState(stateDesc).clear();
+	}
+
 	@Override
 	public String toString() {
 		return "CountTrigger(" +  maxCount + ")";

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/DeltaTrigger.java

@@ -71,6 +71,11 @@ public class DeltaTrigger<T, W extends Window> implements Trigger<T, W> {
 		return TriggerResult.CONTINUE;
 	}
 
+	@Override
+	public void clear(W window, TriggerContext ctx) throws Exception {
+		ctx.getPartitionedState(stateDesc).clear();
+	}
+
 	@Override
 	public String toString() {
 		return "DeltaTrigger(" +  deltaFunction + ", " + threshold + ")";

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/EventTimeTrigger.java

@@ -46,6 +46,11 @@ public class EventTimeTrigger implements Trigger<Object, TimeWindow> {
 		return TriggerResult.CONTINUE;
 	}
 
+	@Override
+	public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
+		ctx.deleteEventTimeTimer(window.maxTimestamp());
+	}
+
 	@Override
 	public String toString() {
 		return "EventTimeTrigger()";

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ProcessingTimeTrigger.java

@@ -44,6 +44,11 @@ public class ProcessingTimeTrigger implements Trigger<Object, TimeWindow> {
 		return TriggerResult.FIRE_AND_PURGE;
 	}
 
+	@Override
+	public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
+		ctx.deleteProcessingTimeTimer(window.maxTimestamp());
+	}
+
 	@Override
 	public String toString() {
 		return "ProcessingTimeTrigger()";

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/PurgingTrigger.java

@@ -78,6 +78,11 @@ public class PurgingTrigger<T, W extends Window> implements Trigger<T, W> {
 		}
 	}
 
+	@Override
+	public void clear(W window, TriggerContext ctx) throws Exception {
+		nestedTrigger.clear(window, ctx);
+	}
+
 	@Override
 	public String toString() {
 		return "PurgingTrigger(" + nestedTrigger.toString() + ")";

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/Trigger.java

@@ -73,6 +73,13 @@ public interface Trigger<T, W extends Window> extends Serializable {
 	 */
 	TriggerResult onEventTime(long time, W window, TriggerContext ctx) throws Exception;
 
+	/**
+	 * Clears any state that the trigger might still hold for the given window. This is called
+	 * when a window is purged. Timers set using {@link TriggerContext#registerEventTimeTimer(long)}
+	 * and {@link TriggerContext#registerProcessingTimeTimer(long)} should be deleted here as
+	 * well as state acquired using {@link TriggerContext#getPartitionedState(StateDescriptor)}.
+	 */
+	void clear(W window, TriggerContext ctx) throws Exception;
 
 	/**
 	 * Result type for trigger methods. This determines what happens with the window.
@@ -150,6 +157,16 @@ public interface Trigger<T, W extends Window> extends Serializable {
 		 */
 		void registerEventTimeTimer(long time);
 
+		/**
+		 * Delete the processing time trigger for the given time.
+		 */
+		void deleteProcessingTimeTimer(long time);
+
+		/**
+		 * Delete the event-time trigger for the given time.
+		 */
+		void deleteEventTimeTimer(long time);
+
 		/**
 		 * Retrieves an {@link State} object that can be used to interact with
 		 * fault-tolerant state that is scoped to the window and key of the current

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

@@ -284,6 +284,10 @@ public class NonKeyedWindowOperator<IN, OUT, W extends Window>
 		if (triggerResult.isFire()) {
 			emitWindow(context);
 		}
+
+		if (triggerResult.isPurge()) {
+			context.clear();
+		}
 	}
 
 	@Override
@@ -516,6 +520,23 @@ public class NonKeyedWindowOperator<IN, OUT, W extends Window>
 			triggers.add(this);
 		}
 
+		@Override
+		public void deleteProcessingTimeTimer(long time) {
+			Set<Context> triggers = processingTimeTimers.get(time);
+			if (triggers != null) {
+				triggers.remove(this);
+			}
+		}
+
+		@Override
+		public void deleteEventTimeTimer(long time) {
+			Set<Context> triggers = watermarkTimers.get(time);
+			if (triggers != null) {
+				triggers.remove(this);
+			}
+
+		}
+
 		public Trigger.TriggerResult onElement(StreamRecord<IN> element) throws Exception {
 			Trigger.TriggerResult onElementResult = trigger.onElement(element.getValue(), element.getTimestamp(), window, this);
 			if (watermarkTimer > 0 && watermarkTimer <= currentWatermark) {
@@ -553,6 +574,10 @@ public class NonKeyedWindowOperator<IN, OUT, W extends Window>
 				return Trigger.TriggerResult.CONTINUE;
 			}
 		}
+
+		public void clear() throws Exception {
+			trigger.clear(window, this);
+		}
 	}
 
 	/**

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

 /**
-* Licensed to the Apache Software Foundation (ASF) under one
-* or more contributor license agreements.  See the NOTICE file
-* distributed with this work for additional information
-* regarding copyright ownership.  The ASF licenses this file
-* to you under the Apache License, Version 2.0 (the
-* "License"); you may not use this file except in compliance
-* with the License.  You may obtain a copy of the License at
-*
-* http://www.apache.org/licenses/LICENSE-2.0
-*
-* Unless required by applicable law or agreed to in writing, software
-* distributed under the License is distributed on an "AS IS" BASIS,
-* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-* See the License for the specific language governing permissions and
-* limitations under the License.
-*/
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package org.apache.flink.streaming.runtime.operators.windowing;
 
 import com.google.common.annotations.VisibleForTesting;
@@ -85,8 +85,8 @@ import static java.util.Objects.requireNonNull;
  * @param <W> The type of {@code Window} that the {@code WindowAssigner} assigns.
  */
 public class WindowOperator<K, IN, ACC, OUT, W extends Window>
-		extends AbstractUdfStreamOperator<OUT, WindowFunction<ACC, OUT, K, W>>
-		implements OneInputStreamOperator<IN, OUT>, Triggerable, InputTypeConfigurable {
+	extends AbstractUdfStreamOperator<OUT, WindowFunction<ACC, OUT, K, W>>
+	implements OneInputStreamOperator<IN, OUT>, Triggerable, InputTypeConfigurable {
 
 	private static final long serialVersionUID = 1L;
 
@@ -164,12 +164,12 @@ public class WindowOperator<K, IN, ACC, OUT, W extends Window>
 	 * Creates a new {@code WindowOperator} based on the given policies and user functions.
 	 */
 	public WindowOperator(WindowAssigner<? super IN, W> windowAssigner,
-			TypeSerializer<W> windowSerializer,
-			KeySelector<IN, K> keySelector,
-			TypeSerializer<K> keySerializer,
-			StateDescriptor<? extends MergingState<IN, ACC>> windowStateDescriptor,
-			WindowFunction<ACC, OUT, K, W> windowFunction,
-			Trigger<? super IN, ? super W> trigger) {
+		TypeSerializer<W> windowSerializer,
+		KeySelector<IN, K> keySelector,
+		TypeSerializer<K> keySerializer,
+		StateDescriptor<? extends MergingState<IN, ACC>> windowStateDescriptor,
+		WindowFunction<ACC, OUT, K, W> windowFunction,
+		Trigger<? super IN, ? super W> trigger) {
 
 		super(windowFunction);
 
@@ -258,8 +258,6 @@ public class WindowOperator<K, IN, ACC, OUT, W extends Window>
 		if (triggerResult.isFire()) {
 			timestampedCollector.setTimestamp(window.maxTimestamp());
 
-			setKeyContext(key);
-
 			MergingState<IN, ACC> windowState = getPartitionedState(window, windowSerializer,
 				windowStateDescriptor);
 
@@ -269,12 +267,13 @@ public class WindowOperator<K, IN, ACC, OUT, W extends Window>
 
 			if (triggerResult.isPurge()) {
 				windowState.clear();
+				context.clear();
 			}
 		} else if (triggerResult.isPurge()) {
-			setKeyContext(key);
 			MergingState<IN, ACC> windowState = getPartitionedState(window, windowSerializer,
 				windowStateDescriptor);
 			windowState.clear();
+			context.clear();
 		}
 	}
 
@@ -293,7 +292,8 @@ public class WindowOperator<K, IN, ACC, OUT, W extends Window>
 
 				context.key = timer.key;
 				context.window = timer.window;
-				Trigger.TriggerResult triggerResult = context.onEventTime(mark.getTimestamp());
+				setKeyContext(timer.key);
+				Trigger.TriggerResult triggerResult = context.onEventTime(timer.timestamp);
 				processTriggerResult(triggerResult, context.key, context.window);
 			} else {
 				fire = false;
@@ -319,7 +319,8 @@ public class WindowOperator<K, IN, ACC, OUT, W extends Window>
 
 				context.key = timer.key;
 				context.window = timer.window;
-				Trigger.TriggerResult triggerResult = context.onProcessingTime(time);
+				setKeyContext(timer.key);
+				Trigger.TriggerResult triggerResult = context.onProcessingTime(timer.timestamp);
 				processTriggerResult(triggerResult, context.key, context.window);
 			} else {
 				fire = false;
@@ -410,6 +411,23 @@ public class WindowOperator<K, IN, ACC, OUT, W extends Window>
 			}
 		}
 
+		@Override
+		public void deleteProcessingTimeTimer(long time) {
+			Timer<K, W> timer = new Timer<>(time, key, window);
+			if (processingTimeTimers.remove(timer)) {
+				processingTimeTimersQueue.remove(timer);
+			}
+		}
+
+		@Override
+		public void deleteEventTimeTimer(long time) {
+			Timer<K, W> timer = new Timer<>(time, key, window);
+			if (watermarkTimers.remove(timer)) {
+				watermarkTimersQueue.remove(timer);
+			}
+
+		}
+
 		public Trigger.TriggerResult onElement(StreamRecord<IN> element) throws Exception {
 			return trigger.onElement(element.getValue(), element.getTimestamp(), window, this);
 		}
@@ -421,6 +439,18 @@ public class WindowOperator<K, IN, ACC, OUT, W extends Window>
 		public Trigger.TriggerResult onEventTime(long time) throws Exception {
 			return trigger.onEventTime(time, window, this);
 		}
+
+		public void clear() throws Exception {
+			trigger.clear(window, this);
+		}
+
+		@Override
+		public String toString() {
+			return "Context{" +
+				"key=" + key +
+				", window=" + window +
+				'}';
+		}
 	}
 
 	/**
@@ -454,8 +484,8 @@ public class WindowOperator<K, IN, ACC, OUT, W extends Window>
 			Timer<?, ?> timer = (Timer<?, ?>) o;
 
 			return timestamp == timer.timestamp
-					&& key.equals(timer.key)
-					&& window.equals(timer.window);
+				&& key.equals(timer.key)
+				&& window.equals(timer.window);
 
 		}
 
@@ -470,10 +500,10 @@ public class WindowOperator<K, IN, ACC, OUT, W extends Window>
 		@Override
 		public String toString() {
 			return "Timer{" +
-					"timestamp=" + timestamp +
-					", key=" + key +
-					", window=" + window +
-					'}';
+				"timestamp=" + timestamp +
+				", key=" + key +
+				", window=" + window +
+				'}';
 		}
 	}