[BEAM-11481] emit output watermark on watermark hold change #13571

je-ik · 2020-12-17T10:43:51Z

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Choose reviewer(s) and mention them in a comment (R: @username).
Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

Post-Commit Tests Status (on master branch)

Lang	Dataflow	Samza	Twister2
Go	---	---	---
Java
Python		---	---
XLang		---	---

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website	Whitespace	Typescript
Non-portable
Portable	---		---	---	---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

dmvk · 2020-12-17T10:46:45Z

Can you please add a new DoFnOperatorTest for this?

dmvk · 2020-12-17T10:51:29Z

...src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java

+        try {
+          processInputWatermark(false);
+        } catch (Exception ex) {
+          // should not happen


can you please add this as a message for IllegalStateException?

Isn't IllegalStateException meant to be used for an illegal state, which is a state that should not happen? :)

I just love seeing "This should never happen" messages in logs when debugging ;)

dmvk

This makes total sense, we want to progress output WM as fast as possible 👍 In case of splittable DoFn, does input watermark progress at all?

dmvk · 2020-12-17T10:59:02Z

My best guess is that impulse progresses input WM from MIN to MAX and we let watermark hold to handle the progression... in that case, SDF probably never worked properly on flink runner 🤔

...src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java

mxm · 2020-12-17T11:01:11Z

...src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java

+      if (newWatermarkHold > currentWatermarkHold) {
+        try {
+          processInputWatermark(false);


This logic should be removed. All logic which deals with watermark emission should be handled through processWatermark.

I disagree - handling watermark emission in processWatermark only is what causes the issues.

mxm · 2020-12-17T11:02:15Z

...src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java

@@ -739,9 +741,12 @@ public final void processWatermark1(Watermark mark) throws Exception {
    }

    currentInputWatermark = mark.getTimestamp();


We need the following to generalize watermark emission to be able to call this method from other places.

if (mark.getTimestamp > currentInputWatermark) { currentInputWatermark = mark.getTimestamp(); }

Input watermark is not what is concerned by this PR. That logic did not change, we only need to be able to progress output watermark when watermark hold changes.

You would need this change in order to go through the regular watermark emission code without changing the latest seen input watermark.

I would like to generalize the existing code, instead of adding code paths for every exception which is bound to be error-prone.

mxm · 2020-12-17T11:02:56Z

...src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java


+  private void processInputWatermark(boolean advanceInputWatermark) throws Exception {


Please remove if we don't want to scatter the logic about watermark advancement.

I think that is what we must do. We could call processWatermark from processElement, but the call does stuff not necessary to do in processElement. That's why I simply wrapped what is needed to be done in processInputWatermark (maybe we could find a better name to make it clearer).

mxm · 2020-12-17T11:04:34Z

...src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java

    doFnRunner.processElement(streamRecord.getValue());
    checkInvokeFinishBundleByCount();
+    emitWatermarkIfHoldChanged(oldHold);


Is this required on every element? I'd rather trigger this only if we set / remove a hold.

This is very cheap call and actually tests exactly if hold was set (actually reset).

je-ik · 2020-12-17T15:25:12Z

@dmvk test added, @mxm did I answer the questions?

je-ik · 2020-12-17T16:14:13Z

Run Flink ValidatesRunner

je-ik · 2020-12-17T16:17:30Z

Run Java Flink PortableValidatesRunner Streaming

dmvk

Thanks for adding the test, LGTM 👍 I don't have any strong opinions about Max's concerns.

Maybe we could just call processWatermark1(currentInputWatermark) after each element / timer to make processWatermark1 the single code path for watermark advancement / emission, but not sure if that's doable for performance reasons.

mxm

I'd appreciate if we could use processWatermark to contain all watermark logic including emission of new output watermarks if watermark holds change. This avoids scattering the logic for watermarks. Also, I'm not sure it works correctly with the portability code otherwise.

In processWatermark, instead of advancing the input watermark immediately, the code has to first calculate the output watermark based on the latest min watermark hold. Then emit a new output watermark before advancing the input watermark which may result in another output watermark emission:

def processWatermark(watermark):
  newOutputWatermark = calculateOutputWatermark()
  maybeEmitOutputWatermark(newOutputWatermark)
  if watermark > currentInputWatermark:
    currentInputWatermark = watermark
    timeServiceManage.advanceWatermark(watermark)
    processWatermark(watermark)

We can call this on every element, the call is cheap.

je-ik · 2020-12-18T13:53:12Z

@mxm I refactored the code a little, please have a look now.

je-ik · 2020-12-18T13:56:04Z

Run Flink ValidatesRunner

je-ik · 2020-12-18T13:56:10Z

Run Java Flink PortableValidatesRunner Streaming

je-ik · 2020-12-18T14:19:54Z

Run Python_PVR_Flink PreCommit

je-ik · 2020-12-18T14:20:39Z

Run Java PreCommit

mxm · 2020-12-18T16:19:58Z

Feel free to proceed however you like. No need to block this on me. After all, it's a good fix!

je-ik · 2020-12-18T17:06:39Z

Run Java PreCommit

probot-autolabeler bot added flink runners labels Dec 17, 2020

je-ik requested review from mxm and dmvk December 17, 2020 10:43

dmvk reviewed Dec 17, 2020

View reviewed changes

dmvk requested changes Dec 17, 2020

View reviewed changes

mxm reviewed Dec 17, 2020

View reviewed changes

je-ik force-pushed the BEAM-11481 branch from 02f1326 to 8b19bc5 Compare December 17, 2020 15:20

[BEAM-11481] emit output watermark on watermark hold change

8de6faa

je-ik force-pushed the BEAM-11481 branch from 8b19bc5 to 8de6faa Compare December 17, 2020 15:23

dmvk approved these changes Dec 18, 2020

View reviewed changes

mxm reviewed Dec 18, 2020

View reviewed changes

je-ik force-pushed the BEAM-11481 branch from 8de6faa to 415501f Compare December 18, 2020 13:51

je-ik force-pushed the BEAM-11481 branch from 415501f to 8de6faa Compare December 18, 2020 16:21

je-ik merged commit bea6f88 into apache:master Dec 18, 2020

je-ik deleted the BEAM-11481 branch December 18, 2020 17:37

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[BEAM-11481] emit output watermark on watermark hold change #13571

[BEAM-11481] emit output watermark on watermark hold change #13571

je-ik commented Dec 17, 2020

dmvk commented Dec 17, 2020

dmvk Dec 17, 2020

je-ik Dec 17, 2020 •

edited

dmvk Dec 18, 2020

dmvk left a comment

dmvk commented Dec 17, 2020

mxm Dec 17, 2020

je-ik Dec 17, 2020

mxm Dec 17, 2020

je-ik Dec 17, 2020

mxm Dec 18, 2020

mxm Dec 18, 2020

mxm Dec 17, 2020

je-ik Dec 17, 2020

mxm Dec 17, 2020

je-ik Dec 17, 2020

je-ik commented Dec 17, 2020

je-ik commented Dec 17, 2020

je-ik commented Dec 17, 2020

dmvk left a comment

mxm left a comment •

edited

je-ik commented Dec 18, 2020 •

edited

je-ik commented Dec 18, 2020

je-ik commented Dec 18, 2020

je-ik commented Dec 18, 2020

je-ik commented Dec 18, 2020

mxm commented Dec 18, 2020

je-ik commented Dec 18, 2020

		@@ -739,9 +741,12 @@ public final void processWatermark1(Watermark mark) throws Exception {
		}

		currentInputWatermark = mark.getTimestamp();


		private void processInputWatermark(boolean advanceInputWatermark) throws Exception {

[BEAM-11481] emit output watermark on watermark hold change #13571

[BEAM-11481] emit output watermark on watermark hold change #13571

Conversation

je-ik commented Dec 17, 2020

Post-Commit Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

GitHub Actions Tests Status (on master branch)

dmvk commented Dec 17, 2020

Choose a reason for hiding this comment

je-ik Dec 17, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dmvk left a comment

Choose a reason for hiding this comment

dmvk commented Dec 17, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

je-ik commented Dec 17, 2020

je-ik commented Dec 17, 2020

je-ik commented Dec 17, 2020

dmvk left a comment

Choose a reason for hiding this comment

mxm left a comment • edited

Choose a reason for hiding this comment

je-ik commented Dec 18, 2020 • edited

je-ik commented Dec 18, 2020

je-ik commented Dec 18, 2020

je-ik commented Dec 18, 2020

je-ik commented Dec 18, 2020

mxm commented Dec 18, 2020

je-ik commented Dec 18, 2020

je-ik Dec 17, 2020 •

edited

mxm left a comment •

edited

je-ik commented Dec 18, 2020 •

edited