Beam-2535: Support timer output timestamps in the flink runner #10534

reuvenlax · 2020-01-08T20:33:11Z

Adds the appropriate watermark holds in the flink runner

reuvenlax · 2020-01-08T20:33:27Z

Run Flink ValidatesRunner

reuvenlax · 2020-01-08T20:40:03Z

R: xubii

reuvenlax · 2020-01-08T20:41:11Z

Some context: pr/9677 allows users to set their own watermark holds, by tying it to the output timestamp of a timer (this prevents watermark hold from getting "leaked" as when the timer fires the holds are removed). This PR adds this support to the Flink runner as well, which was excluded from the previous PR.

reuvenlax · 2020-01-08T21:01:57Z

Run Flink ValidatesRunner

reuvenlax · 2020-01-08T21:04:51Z

Run Flink ValidatesRunner

reuvenlax · 2020-01-08T21:53:02Z

Run Flink ValidatesRunner

reuvenlax · 2020-01-08T22:05:17Z

Run Flink ValidatesRunner

reuvenlax · 2020-01-08T22:45:00Z

Run Flink ValidatesRunner

reuvenlax · 2020-01-08T23:13:07Z

Run Flink ValidatesRunner

reuvenlax · 2020-01-08T23:27:32Z

Run Flink ValidatesRunner

reuvenlax · 2020-01-08T23:50:44Z

Run Flink ValidatesRunner

reuvenlax · 2020-01-08T23:59:04Z

Run Flink ValidatesRunner

reuvenlax · 2020-01-09T00:34:32Z

@mxm this doesn't quite work. Presumably I'm misunderstanding something about how the Flink runner works. Do you have any thoughts on how this might be implemented?

mxm · 2020-01-09T13:14:36Z

Thanks @reuvenlax. I'll have a look later today.

mxm · 2020-01-09T17:26:41Z

Run Flink ValidatesRunner

mxm · 2020-01-09T17:27:01Z

Run Java Flink PortableValidatesRunner Streaming

mxm · 2020-01-09T17:28:06Z

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

@@ -1171,6 +1173,8 @@ private void cancelPendingTimerById(String contextTimerId) throws Exception {
 void cleanupPendingTimer(TimerData timer) {
 try {
 pendingTimersById.remove(getContextTimerId(timer.getTimerId(), timer.getNamespace()));
+ // Always update watermark hold after a timer has been fired or removed
+ updateWatermarkHold();


This was the missing piece. Without this, we can't make progress when a timer has fired.

Thanks! I didn't put a call here, because the way the code is structured that would result in two calls to updateWatermarkHold every time you set a timer. Would it be sufficient to add the call in fireTimer, as it appears that's that path we missed.

Yes, that's right. My first thought was to add it in fireTimer, but then I thought I could simplify it further. That led to the duplicate update call when you set the timer. Looks like we have just update the hold on set, delete, and fire, and not for cleanup.

mxm · 2020-01-09T17:58:58Z

Unrelated test failure in https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_PR/170/

mxm · 2020-01-09T17:59:08Z

Run Flink ValidatesRunner

mxm · 2020-01-09T19:32:22Z

The portable ValidatesRunner tests are failing. We need to adapt the watermark handling there. Let me check.

reuvenlax · 2020-01-09T19:39:10Z

Run Flink ValidatesRunner

reuvenlax · 2020-01-09T19:39:20Z

Run Java Flink PortableValidatesRunner Streaming

reuvenlax · 2020-01-09T21:07:48Z

Run Flink ValidatesRunner

reuvenlax · 2020-01-09T21:19:06Z

Run Apex ValidatesRunner

reuvenlax · 2020-01-09T21:19:18Z

Run Samza ValidatesRunner

reuvenlax · 2020-01-09T22:16:54Z

Run Java Flink PortableValidatesRunner Streaming

reuvenlax · 2020-01-10T00:46:03Z

This appears to work except for the portable runner

mxm · 2020-01-10T13:45:20Z

This is an issue with the portable TestStream implementation not working with non-standard coders. Let me see if I can find a fix.

reuvenlax · 2020-01-14T04:59:43Z

Run Java Flink PortableValidatesRunner Streaming

reuvenlax · 2020-01-14T05:55:58Z

Run Java Flink PortableValidatesRunner Streaming

reuvenlax · 2020-01-14T22:15:26Z

Run Java Flink PortableValidatesRunner Streaming

reuvenlax · 2020-01-15T03:22:59Z

Run Java Flink PortableValidatesRunner Streaming

reuvenlax · 2020-01-15T03:23:11Z

Run Flink ValidatesRunner

mxm · 2020-01-15T11:43:38Z

Run Java PreCommit

reuvenlax · 2020-01-15T18:15:48Z

Run Flink ValidatesRunner

reuvenlax · 2020-01-15T18:15:56Z

Run Java Flink PortableValidatesRunner Streaming

reuvenlax · 2020-01-15T18:16:18Z

@mxm I believe this now supports both the old and the portable flink runners

mxm · 2020-01-16T08:42:21Z

Run Java PreCommit

mxm · 2020-01-16T08:42:38Z

Run JavaPortabilityApi PreCommit

mxm · 2020-01-16T09:53:30Z

There are still some test failures:
https://builds.apache.org/job/beam_PreCommit_Java_Phrase/1651/
https://builds.apache.org/job/beam_PreCommit_JavaPortabilityApi_Phrase/175/

After those are fixed please revert #10610 in this PR to reactivate the new timer output timestamp test.

reuvenlax · 2020-01-22T04:55:08Z

Run Java PreCommit

reuvenlax · 2020-01-22T04:55:17Z

Run JavaPortabilityApi PreCommit

reuvenlax · 2020-01-22T06:13:32Z

Run JavaPortabilityApi PreCommit

reuvenlax · 2020-01-22T06:43:51Z

@mxm tests are now fixed. JavaPortabilityApi is failing consistently in pushDockerContainer, which seems unrelated.

mxm

Thank you. Looks good! I agree that the test failures look unrelated.

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java

mxm · 2020-01-22T16:39:47Z

Run Flink ValidatesRunner

mxm · 2020-01-22T16:39:57Z

Run Java Flink PortableValidatesRunner Streaming

mxm · 2020-01-22T16:40:30Z

Run Java Flink PortableValidatesRunner Batch

reuvenlax requested a review from mxm January 8, 2020 20:39

mxm reviewed Jan 9, 2020

View reviewed changes

reuvenlax force-pushed the output_timestamp_flink branch from 609c9c4 to 657f4f9 Compare January 9, 2020 19:39

reuvenlax force-pushed the output_timestamp_flink branch from 2b817d5 to fb686f2 Compare January 14, 2020 05:55

reuvenlax force-pushed the output_timestamp_flink branch from fb686f2 to 806be60 Compare January 14, 2020 22:15

reuvenlax force-pushed the output_timestamp_flink branch from 5d58c2e to df68f73 Compare January 15, 2020 18:15

mxm mentioned this pull request Jan 16, 2020

[BEAM-9083] Exclude testOutputTimestamp from flink streaming tests. #10610

Merged

Reuven Lax added 2 commits January 21, 2020 16:34

Fix issues

9f1942d

fix tests

ec8aa24

reuvenlax force-pushed the output_timestamp_flink branch from 6ed6448 to ec8aa24 Compare January 22, 2020 00:36

mxm approved these changes Jan 22, 2020

View reviewed changes

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java Outdated Show resolved Hide resolved

set nullable

15ecdc2

reuvenlax force-pushed the output_timestamp_flink branch from ac4dba3 to 15ecdc2 Compare January 22, 2020 17:32

reuvenlax merged commit dd0b001 into apache:master Jan 22, 2020

Beam-2535: Support timer output timestamps in the flink runner #10534

Beam-2535: Support timer output timestamps in the flink runner #10534

Conversation

reuvenlax commented Jan 8, 2020

reuvenlax commented Jan 8, 2020

reuvenlax commented Jan 8, 2020

reuvenlax commented Jan 8, 2020

reuvenlax commented Jan 8, 2020

reuvenlax commented Jan 8, 2020

reuvenlax commented Jan 8, 2020

reuvenlax commented Jan 8, 2020

reuvenlax commented Jan 8, 2020

reuvenlax commented Jan 8, 2020

reuvenlax commented Jan 8, 2020

reuvenlax commented Jan 8, 2020

reuvenlax commented Jan 8, 2020

reuvenlax commented Jan 9, 2020

mxm commented Jan 9, 2020

mxm commented Jan 9, 2020

mxm commented Jan 9, 2020

mxm Jan 9, 2020

Choose a reason for hiding this comment

reuvenlax Jan 9, 2020

Choose a reason for hiding this comment

mxm Jan 9, 2020

Choose a reason for hiding this comment

mxm commented Jan 9, 2020

mxm commented Jan 9, 2020

mxm commented Jan 9, 2020

reuvenlax commented Jan 9, 2020

reuvenlax commented Jan 9, 2020

reuvenlax commented Jan 9, 2020

reuvenlax commented Jan 9, 2020

reuvenlax commented Jan 9, 2020

reuvenlax commented Jan 9, 2020

reuvenlax commented Jan 10, 2020

mxm commented Jan 10, 2020

reuvenlax commented Jan 14, 2020

reuvenlax commented Jan 14, 2020

reuvenlax commented Jan 14, 2020

reuvenlax commented Jan 15, 2020

reuvenlax commented Jan 15, 2020

mxm commented Jan 15, 2020

reuvenlax commented Jan 15, 2020

reuvenlax commented Jan 15, 2020

reuvenlax commented Jan 15, 2020

mxm commented Jan 16, 2020

mxm commented Jan 16, 2020

mxm commented Jan 16, 2020

reuvenlax commented Jan 22, 2020

reuvenlax commented Jan 22, 2020

reuvenlax commented Jan 22, 2020

reuvenlax commented Jan 22, 2020

mxm left a comment

Choose a reason for hiding this comment

mxm commented Jan 22, 2020

mxm commented Jan 22, 2020

mxm commented Jan 22, 2020