[BEAM-1726] Obey watermark hold in DoFnOperato.processWatermark2()

cosmoskitten · Mar 21, 2017 · fdc36c9 · fdc36c9
1 parent a974f3c
commit fdc36c9
Showing 1 changed file with 8 additions and 1 deletion.
diff --git a/.../main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java b/.../main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java
@@ -503,7 +503,14 @@ public void processWatermark2(Watermark mark) throws Exception {
       pushbackDoFnRunner.finishBundle();
 
       // maybe output a new watermark
-      processWatermark1(new Watermark(currentInputWatermark));
+      Instant watermarkHold = stateInternals.watermarkHold();
+
+      long potentialOutputWatermark = Math.min(currentInputWatermark, watermarkHold.getMillis());
+
+      if (potentialOutputWatermark > currentOutputWatermark) {
+        currentOutputWatermark = potentialOutputWatermark;
+        output.emitWatermark(new Watermark(currentOutputWatermark));
+      }
     }
   }