Merge pull request apache#2 from aljoscha/finish-pr-2263-flink-passer…

…t-check [BEAM-1726] Obey watermark hold in DoFnOperato.processWatermark2()
cosmoskitten · Mar 21, 2017 · 6aba015 · 6aba015
2 parents a974f3c + fdc36c9
commit 6aba015
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));
+      }
     }
   }