[BEAM-4681] Add support for portable timers in Flink batch mode by mxm · Pull Request #7008 · apache/beam

mxm · 2018-11-12T15:53:23Z

This adds support for portable timers in Flink Runner's batch mode, and enables the Java/Python Portable ValidatesRunner tests.

CC @tweise @angoenka @robertwb

Post-Commit Tests Status (on master branch)

Lang	Apex	Dataflow	Flink	Gearpump	Samza	Spark
Go	---	---	---	---	---	---
Java
Python	---			---	---	---

mxm · 2018-11-12T15:53:53Z

Run Java Flink PortableValidatesRunner

ryan-williams

looks great! I left a couple comments/questions

...rg/apache/beam/runners/flink/translation/wrappers/streaming/ExecutableStageDoFnOperator.java

...n/java/org/apache/beam/runners/flink/translation/functions/FlinkExecutableStageFunction.java

...rg/apache/beam/runners/flink/translation/wrappers/streaming/ExecutableStageDoFnOperator.java

...va/org/apache/beam/runners/flink/translation/functions/FlinkExecutableStageFunctionTest.java

mxm · 2018-11-13T10:19:05Z

Thanks for the review @ryan-williams!

mxm · 2018-11-13T11:27:29Z

Run Java PreCommit

mxm · 2018-11-13T11:27:32Z

Run Java_Examples_Dataflow PreCommit

mxm · 2018-11-13T13:15:31Z

Run Java Flink PortableValidatesRunner

mxm · 2018-11-13T17:31:55Z

Rebased this to #6981.

mxm · 2018-11-13T17:34:36Z

Run Java Flink PortableValidatesRunner

mxm · 2018-11-13T18:43:54Z

Run Java PreCommit

mxm · 2018-11-13T18:44:01Z

Run Java Flink PortableValidatesRunner

mxm · 2018-11-15T20:06:15Z

Run Python Flink ValidatesRunner

mxm · 2018-11-15T20:06:21Z

Run Java Flink PortableValidatesRunner

mxm · 2018-11-16T10:18:03Z

Run Java PreCommit

mxm · 2018-11-16T10:21:10Z

Run Python PreCommit

mxm · 2018-11-16T10:35:19Z

Run Java Flink PortableValidatesRunner

mxm · 2018-11-16T11:09:47Z

Run Flink ValidatesRunner

mxm · 2018-11-16T12:04:18Z

Run Flink ValidatesRunner

mxm · 2018-11-16T14:02:05Z

Run Flink ValidatesRunner

mxm · 2018-11-16T14:02:11Z

Run Java Flink PortableValidatesRunner

mxm · 2018-11-16T14:05:34Z

Run Python Flink ValidatesRunner

mxm · 2018-11-16T14:14:27Z

Run Python Flink ValidatesRunner

mxm · 2018-11-16T14:20:28Z

Run Flink ValidatesRunner

mxm · 2018-11-16T14:34:37Z

Run Java PreCommit

mxm · 2018-11-16T16:37:58Z

Ran "Java Flink PortableValidatesRunner" locally with the latest master. All tests passed.

mxm · 2018-11-16T16:38:43Z

Unfortunately, the Jenkins setup for it seems to be broken at the moment.

mxm · 2018-11-16T16:39:58Z

Same goes for "Run JavaPortabilityApi PreCommit" which fails continuously at the moment for all commits. My PR doesn't change anything regarding the portability API.

...n/java/org/apache/beam/runners/flink/translation/functions/FlinkExecutableStageFunction.java

tweise · 2018-11-18T21:59:11Z

...n/java/org/apache/beam/runners/flink/translation/functions/FlinkExecutableStageFunction.java

+        int tagInt = unionTag;
+        return receivedElement -> {
+          synchronized (collectorLock) {
+            collector.collect(new RawUnionValue(tagInt, receivedElement));


In streaming doing this in the RPC handler can cause a deadlock when collect triggers the next bundle in an operator chain. Presumably that cannot happen due to different scheduling in batch?

Good question. I think you are right. This can deadlock if the receiving side backpressures because it is waiting for input from another channel generated by this operator. Is that what you meant?

Yes, but I don't know if it is applicable for batch since it is scheduled stage by stage. For streaming the issue would surface just by running wordcount, for batch not.

The stage-by-stage is only in place if the ExecutionMode is set to FORCE_BATCH, instead of the default PIPELINE.

...n/java/org/apache/beam/runners/flink/translation/functions/FlinkExecutableStageFunction.java

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

mxm · 2018-11-19T11:50:27Z

Run Website PreCommit

mxm · 2018-11-19T11:50:48Z

Run Python Flink ValidatesRunner

mxm · 2018-11-19T13:33:19Z

Squashed for merge.

mxm · 2018-11-19T13:33:52Z

Run Python Flink ValidatesRunner

mxm · 2018-11-19T13:53:59Z

Run Website PreCommit

mxm · 2018-11-19T14:13:09Z

Run Website PreCommit

tweise

Minor side note: The commits belonging to BEAM-4681 are one unit of rollback, you decide if you want to squash them or not.

tweise · 2018-11-19T14:06:17Z

...n/java/org/apache/beam/runners/flink/translation/functions/FlinkExecutableStageFunction.java

+        int tagInt = unionTag;
+        return receivedElement -> {
+          synchronized (collectorLock) {
+            collector.collect(new RawUnionValue(tagInt, receivedElement));


Yes, but I don't know if it is applicable for batch since it is scheduled stage by stage. For streaming the issue would surface just by running wordcount, for batch not.

mxm · 2018-11-19T14:31:04Z

The commits belonging to BEAM-4681 are one unit of rollback

I find the structure quite logical and it should enable more incremental rollback:

[BEAM-4681] Add support for portable timers in Flink batch mode
Adds the batch support for timers
[BEAM-4681] Use CoderUtils to decode key
Fixes an issue for Java PVR for streaming timers
[BEAM-4681] Enable Python ValidatesRunner timer tests
Enables Python PVR
[BEAM-4681] Enable Java portable ValidatesRunner timer tests
Enables Java PVR
[BEAM-4681] Simplify by replacing the two timer keys with a single one
Refactoring which can be rolled back independently.

But it is a personal preference and I'd also be fine with squashing.

mxm · 2018-11-19T14:33:26Z

Run Flink ValidatesRunner

mxm · 2018-11-19T15:46:48Z

Run Website PreCommit

mxm force-pushed the flink-timers-batch branch 2 times, most recently from ed93559 to fb53c99 Compare November 12, 2018 16:47

ryan-williams reviewed Nov 13, 2018

View reviewed changes

mxm force-pushed the flink-timers-batch branch from 5a3a299 to 23c278c Compare November 13, 2018 10:17

mxm force-pushed the flink-timers-batch branch from 81f5f67 to 0da9483 Compare November 13, 2018 10:27

mxm force-pushed the flink-timers-batch branch from a916c36 to 48f5ae6 Compare November 13, 2018 17:31

mxm force-pushed the flink-timers-batch branch 2 times, most recently from bf9045b to 7c74482 Compare November 13, 2018 17:34

mxm force-pushed the flink-timers-batch branch 4 times, most recently from 7c247fb to 5cb59b2 Compare November 15, 2018 20:04

mxm requested a review from tweise November 15, 2018 20:09

mxm force-pushed the flink-timers-batch branch from 5cb59b2 to bbefdbd Compare November 16, 2018 12:04

mxm force-pushed the flink-timers-batch branch from 3537b56 to 62a1313 Compare November 16, 2018 14:14

tweise reviewed Nov 18, 2018

View reviewed changes

mxm force-pushed the flink-timers-batch branch from e604fed to 284222a Compare November 19, 2018 11:48

mxm added 6 commits November 19, 2018 14:26

[BEAM-4681] Add support for portable timers in Flink batch mode

b032f5d

[BEAM-4681] Use CoderUtils to decode key

525d7e5

[BEAM-867] Support set/delete of timers by ID in FlinkRunner

a3a96bc

[BEAM-4681] Enable Java portable ValidatesRunner timer tests

53e1c98

[BEAM-4681] Enable Python ValidatesRunner timer tests

604c5ff

[BEAM-4681] Simplify by replacing the two timer keys with a single one

b752ad5

mxm force-pushed the flink-timers-batch branch from 284222a to b752ad5 Compare November 19, 2018 13:32

tweise approved these changes Nov 19, 2018

View reviewed changes

mxm merged commit 50f623c into apache:master Nov 19, 2018

Conversation

mxm commented Nov 12, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Post-Commit Tests Status (on master branch)

Uh oh!

mxm commented Nov 12, 2018

Uh oh!

ryan-williams left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

mxm commented Nov 13, 2018

Uh oh!

mxm commented Nov 13, 2018

Uh oh!

mxm commented Nov 13, 2018

Uh oh!

mxm commented Nov 13, 2018

Uh oh!

mxm commented Nov 13, 2018

Uh oh!

mxm commented Nov 13, 2018

Uh oh!

mxm commented Nov 13, 2018

Uh oh!

mxm commented Nov 13, 2018

Uh oh!

mxm commented Nov 15, 2018

Uh oh!

mxm commented Nov 15, 2018

Uh oh!

mxm commented Nov 16, 2018

Uh oh!

mxm commented Nov 16, 2018

Uh oh!

mxm commented Nov 16, 2018

Uh oh!

mxm commented Nov 16, 2018

Uh oh!

mxm commented Nov 16, 2018

Uh oh!

mxm commented Nov 16, 2018

Uh oh!

mxm commented Nov 16, 2018

Uh oh!

mxm commented Nov 16, 2018

Uh oh!

mxm commented Nov 16, 2018

Uh oh!

mxm commented Nov 16, 2018

Uh oh!

mxm commented Nov 16, 2018

Uh oh!

mxm commented Nov 16, 2018

Uh oh!

mxm commented Nov 16, 2018

Uh oh!

mxm commented Nov 16, 2018

Uh oh!

Uh oh!

tweise Nov 18, 2018

Choose a reason for hiding this comment

Uh oh!

mxm Nov 19, 2018

Choose a reason for hiding this comment

Uh oh!

tweise Nov 19, 2018

Choose a reason for hiding this comment

Uh oh!

mxm Nov 19, 2018

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

mxm commented Nov 12, 2018 •

edited

Loading