SAMZA-1578: Fix watermark bug found by BEAM tests

The problem is getOutputWatermark() does not return the real outputWatermark. This caused problem in user override watermark function. Author: xiliu <[email protected]> Reviewers: Jagadish <[email protected]> Closes apache#415 from xinyuiscool/SAMZA-1578
bdevore17 · Jan 26, 2018 · 7e68e4b · 7e68e4b
1 parent 9674836
commit 7e68e4b
Showing 1 changed file with 4 additions and 5 deletions.
diff --git a/samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImpl.java b/samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImpl.java
@@ -356,7 +356,7 @@ private final void onWatermark(long watermark, MessageCollector collector, TaskC
         // use samza-provided watermark handling
         // default is to propagate the input watermark
         output = handleWatermark(currentWatermark, collector, coordinator);
-        outputWm = getOutputWatermark();
+        outputWm = currentWatermark;
       }
 
       if (!output.isEmpty()) {
@@ -402,14 +402,13 @@ final long getInputWatermark() {
   }
 
   /**
-   * Returns the output watermark, default is the same as input.
-   * Operators which keep track of watermark should override this to return the current watermark.
+   * Returns the output watermark,
    * @return output watermark
    */
-  protected long getOutputWatermark() {
+  final long getOutputWatermark() {
     if (usedInCurrentTask) {
       // default as input
-      return this.currentWatermark;
+      return this.outputWatermark;
     } else {
       // always emit the max to indicate no input will be emitted afterwards
       return Long.MAX_VALUE;