[FLINK-3714] Remove Unneccesary Timer in EventTimeTrigger

In onElement() we registered a timer for the case where the watermark is
already past the end of the window and we're firing anyways. Now, only
add a timer if the watermark is not already past the end of the window.

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

@@ -35,11 +35,13 @@ public class EventTimeTrigger extends Trigger<Object, TimeWindow> {
 
 	@Override
 	public TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
-		ctx.registerEventTimeTimer(window.maxTimestamp());
-
-		return (window.maxTimestamp() <= ctx.getCurrentWatermark()) ?
-			TriggerResult.FIRE_AND_PURGE :
-			TriggerResult.CONTINUE;
+		if (window.maxTimestamp() <= ctx.getCurrentWatermark()) {
+			// if the watermark is already past the window fire immediately
+			return TriggerResult.FIRE_AND_PURGE;
+		} else {
+			ctx.registerEventTimeTimer(window.maxTimestamp());
+			return TriggerResult.CONTINUE;
+		}
 	}
 
 	@Override