[FLINK-3688] WindowOperator.trigger() does not emit Watermark anymore

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

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

@@ -341,6 +341,14 @@ protected void processTriggerResult(TriggerResult triggerResult, K key, W window
 
  @Override
  public final void processWatermark(Watermark mark) throws Exception {
+ processTriggersFor(mark);
+
+ output.emitWatermark(mark);
+
+ this.currentWatermark = mark.getTimestamp();
+ }
+
+ private void processTriggersFor(Watermark mark) throws Exception {
  boolean fire;
 
  do {
@@ -360,10 +368,6 @@ public final void processWatermark(Watermark mark) throws Exception {
  fire = false;
  }
  } while (fire);
-
- output.emitWatermark(mark);
-
- this.currentWatermark = mark.getTimestamp();
  }
 
  @Override
@@ -391,7 +395,7 @@ public final void trigger(long time) throws Exception {
  // Also check any watermark timers. We might have some in here since
  // Context.registerEventTimeTimer sets a trigger if an event-time trigger is registered
  // that is already behind the watermark.
- processWatermark(new Watermark(currentWatermark));
+ processTriggersFor(new Watermark(currentWatermark));
  }
 
  /**