[BEAM-8581] and [BEAM-8582] watermark and trigger fixes #10035

rohdesamuel · 2019-11-08T00:14:24Z

The GeneralTriggerDriver does not put watermark holds on timers, leading to the ontime empty pane being considered late data.

The DefaultTrigger and AfterWatermark do not clear their timers after the watermark passed the end of the endow, leading to duplicate records being emitted.

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.
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	SDK	Apex	Dataflow	Gearpump	Samza	Spark
Go		---	---	---	---
Java
Python		---		---	---
XLang	---	---	---	---	---	---

Pre-Commit Tests Status (on master branch)

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

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

rohdesamuel · 2019-11-08T00:14:56Z

R: @robertwb
R: @dpmills
Can you review this please?

rohdesamuel · 2019-11-11T23:19:08Z

This is blocking #9953

sdks/python/apache_beam/transforms/trigger.py

sdks/python/apache_beam/testing/util.py

sdks/python/apache_beam/runners/direct/util.py

sdks/python/apache_beam/testing/test_stream_test.py

sdks/python/apache_beam/transforms/trigger.py

sdks/python/apache_beam/transforms/trigger_test.py

sdks/python/apache_beam/transforms/trigger.py

rohdesamuel · 2019-11-13T21:20:46Z

Run Portable_Python PreCommit

rohdesamuel · 2019-11-21T23:55:53Z

There hasn't been any review in the last 9 days, so I'm asking @pabloem to merge.

robertwb

I think this is OK. Can you squash to a commit with the testing changes and another with the trigger fixes. After that, LGTM.

robertwb · 2019-11-22T00:04:46Z

sdks/python/apache_beam/transforms/trigger.py

    pass

  def process_entire_key(
-      self, key, windowed_values, output_watermark=MIN_TIMESTAMP):
+      self, key, windowed_values, input_watermark=MIN_TIMESTAMP,


Have we verified we found all callers of this code? Inserting a (default) argument before existing arguments could result in an off-by-one error for unmodified callers.

Also, the ordering should be consistent with process_elements().

Is MIN_TIMESTAMP the correct default?

Yeah, the only other place where these are called directly are in the Batch Dataflow Python Worker. So by setting these to MIN_TIMESTAMP we go back to the old implementation which didn't affect Batch.

robertwb · 2019-11-22T00:08:29Z

sdks/python/apache_beam/transforms/trigger.py

@@ -1036,14 +1074,17 @@ class BatchGlobalTriggerDriver(TriggerDriver):
      index=0,
      nonspeculative_index=0)

-  def process_elements(self, state, windowed_values, unused_output_watermark):
+  def process_elements(self, state, windowed_values,
+                       unused_output_watermark=MIN_TIMESTAMP,


Let's not provide default values here, as it's unclear what they should be.

Ack, I set these to None.

I meant not provide default (None or otherwise).

Ah gotcha. I removed the default value on the output_watermarks, but the input_watermark needs a default value to keep backwards compatibility.

robertwb · 2019-11-22T00:57:06Z

Run Python Precommit

rohdesamuel · 2019-11-22T18:35:10Z

Argh, sorry for the force pushes. Local repo got into a weird state.

robertwb

As requested, can you create one commit with the testing changes and another with the trigger fixes.

rohdesamuel · 2019-11-25T20:50:39Z

Ah gotcha, misread the comment. I just fixed up the commits as asked.

rohdesamuel · 2019-11-26T00:19:32Z

Run Portable_Python PreCommit

rohdesamuel · 2019-11-26T00:19:37Z

Run Python PreCommit

rohdesamuel · 2019-11-26T17:46:14Z

Run Portable_Python PreCommit

rohdesamuel · 2019-11-26T17:46:17Z

Run Portable_Python PreCommit

rohdesamuel · 2019-11-26T17:46:28Z

Run Python PreCommit

rohdesamuel · 2019-11-26T19:28:22Z

Run Portable_Python PreCommit

rohdesamuel · 2019-11-26T19:28:29Z

Run Python PreCommit

rohdesamuel · 2019-11-26T21:53:13Z

Run Python PreCommit

rohdesamuel · 2019-11-26T21:53:19Z

Run Portable_Python PreCommit

rohdesamuel · 2019-11-26T22:56:13Z

Run Portable_Python PreCommit

rohdesamuel · 2019-11-27T00:55:12Z

Run Python PreCommit

rohdesamuel · 2019-11-27T17:19:00Z

Run Python PreCommit

rohdesamuel · 2019-12-02T17:30:58Z

Run Python PreCommit

rohdesamuel · 2019-12-02T20:31:07Z

Run Python PreCommit

robertwb

I was going to merge this but noted some unrelated changes.

sdks/python/apache_beam/transforms/trigger_test.py

rohdesamuel · 2019-12-04T17:42:31Z

Run Python PreCommit

HuangLED · 2019-12-05T21:14:45Z

friendly ping for merging this PR.

robertwb

Looks like there were some merge issues or something. I'm going to go through this again, and fix up the history and merge it if I can.

robertwb · 2019-12-05T23:31:09Z

sdks/python/apache_beam/runners/direct/transform_evaluator.py

@@ -608,7 +606,7 @@ def start_bundle(self):

  def process_timer(self, timer_firing):
    if timer_firing.name not in self.user_timer_map:
-      _LOGGER.warning('Unknown timer fired: %s', timer_firing)
+      logging.warning('Unknown timer fired: %s', timer_firing)


Looks like this was reverted again.

robertwb · 2019-12-05T23:36:06Z

sdks/python/apache_beam/runners/direct/util.py

-    return 'TimerFiring(%r, %r, %s, %s)' % (self.encoded_key,
-                                            self.name, self.time_domain,
-                                            self.timestamp)
+    return 'TimerFiring({}, {}, {}, {})'.format(self.encoded_key,


The use of repr for encoded keys was intentional here.

robertwb · 2019-12-06T00:04:03Z

I fixed the merge conflicts and history and reverted un-intentional changes on the other PR, which has been merged.

dpmills suggested changes Nov 12, 2019

View reviewed changes

sdks/python/apache_beam/transforms/trigger.py Outdated Show resolved Hide resolved

sdks/python/apache_beam/transforms/trigger.py Outdated Show resolved Hide resolved

sdks/python/apache_beam/testing/util.py Outdated Show resolved Hide resolved

robertwb reviewed Nov 12, 2019

View reviewed changes

rohdesamuel force-pushed the triggerfix branch from 84ec65f to e8cd4cd Compare November 13, 2019 00:49

rohdesamuel mentioned this pull request Nov 13, 2019

[BEAM-8335] Adds support for multi-output TestStream #9953

Merged

3 tasks

robertwb reviewed Nov 22, 2019

View reviewed changes

rohdesamuel force-pushed the triggerfix branch from 1fe194c to e9a2606 Compare November 22, 2019 18:34

robertwb reviewed Nov 25, 2019

View reviewed changes

rohdesamuel force-pushed the triggerfix branch from e9a2606 to b2845d8 Compare November 25, 2019 20:50

robertwb reviewed Dec 3, 2019

View reviewed changes

sdks/python/apache_beam/transforms/trigger_test.py Outdated Show resolved Hide resolved

rohdesamuel added 4 commits December 3, 2019 16:29

fix assert equals_to_per_window to actually assert window's existence

bc0df4c

Fix [BEAM-8581] and [BEAM-8582]

c49add3

remove default values on output watermarks for triggers

e8caef8

restore accidentally deleted test

3dfd0b6

rohdesamuel force-pushed the triggerfix branch from 386a396 to 3dfd0b6 Compare December 4, 2019 00:37

robertwb reviewed Dec 5, 2019

View reviewed changes

robertwb mentioned this pull request Dec 6, 2019

[BEAM-8581] and [BEAM-8582] watermark and trigger fixes #10304

Merged

3 tasks

robertwb approved these changes Dec 6, 2019

View reviewed changes

robertwb closed this Dec 6, 2019

[BEAM-8581] and [BEAM-8582] watermark and trigger fixes #10035

[BEAM-8581] and [BEAM-8582] watermark and trigger fixes #10035

Conversation

rohdesamuel commented Nov 8, 2019 • edited Loading

Post-Commit Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

rohdesamuel commented Nov 8, 2019

rohdesamuel commented Nov 11, 2019

rohdesamuel commented Nov 13, 2019

rohdesamuel commented Nov 21, 2019

robertwb left a 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

robertwb commented Nov 22, 2019

rohdesamuel commented Nov 22, 2019

robertwb left a comment

Choose a reason for hiding this comment

rohdesamuel commented Nov 25, 2019

rohdesamuel commented Nov 26, 2019

rohdesamuel commented Nov 26, 2019

rohdesamuel commented Nov 26, 2019

rohdesamuel commented Nov 26, 2019

rohdesamuel commented Nov 26, 2019

rohdesamuel commented Nov 26, 2019

rohdesamuel commented Nov 26, 2019

rohdesamuel commented Nov 26, 2019

rohdesamuel commented Nov 26, 2019

rohdesamuel commented Nov 26, 2019

rohdesamuel commented Nov 27, 2019

rohdesamuel commented Nov 27, 2019

rohdesamuel commented Dec 2, 2019

rohdesamuel commented Dec 2, 2019

robertwb left a comment

Choose a reason for hiding this comment

rohdesamuel commented Dec 4, 2019

HuangLED commented Dec 5, 2019

robertwb left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

robertwb commented Dec 6, 2019

rohdesamuel commented Nov 8, 2019 •

edited

Loading