[BEAM-7730] Introduce Flink 1.9 Runner #9296

dmvk · 2019-08-08T13:23:02Z

Runner for Flink 1.9. JIRA.

sunjincheng121

Thanks for the PR @dmvk!
The changes looks good to me.
Only one suggestion here: Could you please upgrade the flink version in https://github.com/apache/beam/blob/master/runners/flink/1.8/build.gradle. i.e., flink_version = '1.8.0' -> flink_version = '1.8.1'?
Best,
Jincheng

dmvk · 2019-08-26T12:46:44Z

@sunjincheng121 Thanks for the review! I'll try to finish this one later this week. I think we can not upgrade to 1.8.1 due to:

and will have to wait for 1.8.2 release.

Best,
D.

dmvk · 2019-08-26T16:11:09Z

Run Java PreCommit

sunjincheng121 · 2019-08-27T08:20:40Z

I see. Thanks @dmvk

dmvk · 2019-08-27T11:38:43Z

@sunjincheng121 I'm still unable to fix the portability test for timers (the one that is @Ignored in this PR). It looks up state for a key from a different key-group during @OnTimer execution, which results in NPE. I'm still unsure about the cause. When I change paralellism to 1, it works as expected because all the keys are in a single partition.

Flink PR that caused the "regression" from 1.8 is most likely this one https://issues.apache.org/jira/browse/FLINK-12693.

Any ideas?

sunjincheng121 · 2019-08-30T05:42:15Z

Thanks @dmvk, I would like to check it, then feedback you! :)

sunjincheng121 · 2019-09-03T05:55:25Z

Hi @dmvk，I creat a hotfix to address the problem. (see: #9464)

Currently, state keys are encoded as NESTED while the keys in keyselector(KvToByteBufferKeySelector)
are encoded as OUTER. This makes the keyGroupId of the same key are different and
lead to the test failure.
The newest flink version makes this bug more visible.

I have check it in my local with your patch. Could you please have look at the changes of the hotfix and double check it whether the hotfix is make sense to you. :)

Thank you!
Jincheng

dmvk · 2019-09-04T15:16:29Z

@sunjincheng121 Thanks for the help! ;) I've changed the flink side coder to NESTED for now, until #9464 discussion gets resolved. I'm unsure about the implications changing context to OUTER would have on other runners.

dmvk · 2019-09-05T09:16:41Z

@sunjincheng121 All checks are finally green ;) Can please you do the final review?

sunjincheng121 · 2019-09-05T10:08:35Z

Thanks for the update @dmvk
LGTM. +1 to mreged.
Best, Jincheng

mxm

Thanks @dmvk for taking the time to work on the Flink 1.9 Runner. Much appreciated!

There are some questions inline, especially I'm concerned about the high number of duplicated files, for which we already have two versions. It should be possible to avoid those duplications and keep at most two versions of the suspect files (current state).

Another point is the addition of a feature unrelated to Flink 1.9. I don't think his PR should add any new features but Flink 1.9. Additionally, the stop functionality will even be removed in the next Flink release, so I don't see a reason for adding this. Even if we decide to add it, it should be handled in a separate PR.

mxm · 2019-09-06T14:49:38Z

...a/org/apache/beam/runners/flink/translation/wrappers/streaming/io/BeamStoppableFunction.java

+ *     href="https://github.com/apache/flink/commit/e95b347dda5233f22fb03e408f2aa521ff924996">Flink
+ *     interface removal commit.</a>
+ */
+public interface BeamStoppableFunction extends StoppableFunction {}


We are adding support for this now, although it has already been removed upstream? Please make this a separate PR.

Hi Max, this is just an interface for a backward compatibility.

Originally RichParallelSourceFunction was implementing StoppableFunction, which was removed as of Flink 1.9., so the following code wouldn't compile:

beam/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingTransformTranslators.java

Line 1400 in 9043b3e

unboundedSourceWrapper.stop();

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

Line 365 in 9043b3e

isRunning = false;

I didn't want to break this for the previous runner versions, so I introduced this interface that would handle the transition.

Other options would be completely remove the stop() implementation from the above mentioned classes.

WDYT?

Thanks for the explanation, that makes sense to me.

Other options would be completely remove the stop() implementation from the above mentioned classes.

That wouldn't work across different versions due to stop function having to be implemented.

Originally RichParallelSourceFunction was implementing StoppableFunction, which was removed as of Flink 1.9., so the following code wouldn't compile:

Actually, I just checked this again in Flink. StoppableFunction always has been optionally. It was not implemented by RichParallelSourceFunction. The only place where we use it in Beam is in UnboundedSourceWrapper. Like you also suggested, I'd remove the stop implementation because it does not allow for a more graceful stop than cancel. This simplifies the setup here and in any case this will be broken in 1.9, so good to also make this change in Beam now.

I'd rather create a JIRA to remove this once we get rid of 1.7 runner as some users may still rely on this.

...k/1.9/src/main/java/org/apache/beam/runners/flink/translation/types/CoderTypeSerializer.java

...a/org/apache/beam/runners/flink/translation/wrappers/streaming/io/BeamStoppableFunction.java

.../src/test/java/org/apache/beam/runners/flink/streaming/FlinkBroadcastStateInternalsTest.java

...flink/1.9/src/test/java/org/apache/beam/runners/flink/streaming/FlinkStateInternalsTest.java

mxm · 2019-09-06T15:00:39Z

...link/src/test/java/org/apache/beam/runners/flink/FlinkStreamingTransformTranslatorsTest.java

-        applyReadSourceTransform(transform, PCollection.IsBounded.BOUNDED, env);
+    SourceTransformation<?> sourceTransform =
+        (SourceTransformation)
+            applyReadSourceTransform(transform, PCollection.IsBounded.BOUNDED, env);


Why do we need to cast here, change the variable type?

Because SourceTransformation and OneInputTrasformation no longer extend a common interface, so applyReadSourceTransform needs to return Object instead.

mxm · 2019-09-06T15:01:55Z

...test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperatorTest.java

@@ -1038,7 +1039,7 @@ void pushbackDataCheckpointing(

    assertThat(
        stripStreamRecordFromWindowedValue(testHarness.getOutput()),
-        contains(helloElement, worldElement));
+        containsInAnyOrder(helloElement, worldElement));


Why is the order reversed here?

I think this is the last issue, will try to investigate today.

We can't guarantee the order here for keys in the state backend after restoring from a savepoint. This just so happens to work for older Flink versions. So seems fair to change this.

sdks/java/build-tools/src/main/resources/beam/suppressions.xml

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

....9/src/main/java/org/apache/beam/runners/flink/translation/types/EncodedValueSerializer.java

tweise

@dmvk there are a few more general things in this PR that should have been brought to the attention of the community upfront, in the form of an ML thread. Did such discussion happen? I apologize if I missed it. In the meantime I'm placing a stop sign on this PR until there is clarity specifically how we can strike the balance between handling differences between multiple Flink versions and maintainability / tech debt.

dmvk · 2019-09-07T09:03:53Z

@tweise Hi Thomas, there was no upfront discussion as this was first developed in our internal build and now we are in the process of contributing the runner to beam. The runner is made the exactly same way as the runners for previous flink versions. Unfortunately this time, there were many breaking changes on the flink side (due to stream & batch unification effort), so the problem with the current approach became more visible.

There were discussions on ML about removal of 1.5, 1.6 versions, which would IMO eliminate the overhead to a bearable level (these versions are unsupported by the flink community anyway). Most of the classes that we "copy & paste" between versions are due to 1.5 compatibility.

I'll try to bring this to ML to see, if anyone has a better idea how to handle these differences.

Thanks,
D.

dmvk · 2019-09-07T09:30:11Z

It seems that 1.5 and 1.6 discussion was never brought to the ML, I've probably recalled it from a private conversation + BEAM-7962.

mxm · 2019-09-08T08:13:16Z

...a/org/apache/beam/runners/flink/translation/wrappers/streaming/io/BeamStoppableFunction.java

+ *     href="https://github.com/apache/flink/commit/e95b347dda5233f22fb03e408f2aa521ff924996">Flink
+ *     interface removal commit.</a>
+ */
+public interface BeamStoppableFunction extends StoppableFunction {}


Thanks for the explanation, that makes sense to me.

Other options would be completely remove the stop() implementation from the above mentioned classes.

That wouldn't work across different versions due to stop function having to be implemented.

....9/src/main/java/org/apache/beam/runners/flink/translation/types/EncodedValueSerializer.java

.../org/apache/beam/runners/flink/translation/wrappers/streaming/io/UnboundedSourceWrapper.java

runners/flink/1.9/build.gradle

mxm · 2019-09-08T10:35:59Z

IMHO there is no need to block this PR. When the comments are resolved, this looks good to merge. As for your concerns @tweise, there will be a major simplification in many regards (directory structure, code workarounds) once we remove 1.5. Supporting multiple versions does not come for free, especially when interfacing with the internals of a stream processor. That said, the current layout is very practical, we just need to make sure we do not duplicate code if it is not necessary.

tweise · 2019-09-08T13:52:23Z

@mxm Let's continue the larger discussion on the ML.

tweise · 2019-09-08T13:56:25Z

It seems that 1.5 and 1.6 discussion was never brought to the ML, I've probably recalled it from a private conversation + BEAM-7962.

I remember that being a topic several times, but don't recall exactly where. But such change should be an explicit discussion on the ML. In any case, there seems to be consensus to drop 1.5/1.6.

tweise · 2019-09-08T14:07:38Z

IMHO there is no need to block this PR.

AFAIK this PR is blocked pending resolution of the key encoding issue, so that it passes tests. Hopefully in parallel we can make progress on trimming down the code duplication also.

mxm · 2019-09-12T00:14:32Z

@dmvk Let us know when you have addressed the comments. Happy to take another look then.

sunjincheng121 · 2019-09-27T09:29:52Z

Hi @dmvk @mxm I have left some comments in both #9484 and #9464.

runners/flink/1.5/src/test/java/org/apache/beam/runners/flink/streaming/StreamSources.java

mxm · 2019-10-09T08:54:10Z

...a/org/apache/beam/runners/flink/translation/wrappers/streaming/io/BeamStoppableFunction.java

+ *     href="https://github.com/apache/flink/commit/e95b347dda5233f22fb03e408f2aa521ff924996">Flink
+ *     interface removal commit.</a>
+ */
+public interface BeamStoppableFunction extends StoppableFunction {}


Originally RichParallelSourceFunction was implementing StoppableFunction, which was removed as of Flink 1.9., so the following code wouldn't compile:

Actually, I just checked this again in Flink. StoppableFunction always has been optionally. It was not implemented by RichParallelSourceFunction. The only place where we use it in Beam is in UnboundedSourceWrapper. Like you also suggested, I'd remove the stop implementation because it does not allow for a more graceful stop than cancel. This simplifies the setup here and in any case this will be broken in 1.9, so good to also make this change in Beam now.

dmvk · 2019-10-10T11:31:32Z

Run Java PreCommit

dmvk · 2019-10-11T13:38:13Z

Run Python PreCommit

sunjincheng121 · 2019-10-14T01:52:52Z

Run CommunityMetrics PreCommit

mxm

We still need to update the Flink Runner page. Looks good other than the comments below.

mxm · 2019-10-17T10:49:01Z

runners/flink/flink_runner.gradle

@@ -142,7 +165,7 @@ class ValidatesRunnerConfig {

 def createValidatesRunnerTask(Map m) {
  def config = m as ValidatesRunnerConfig
-  tasks.create(name: config.name, type: Test) {
+  tasks.register(config.name, Test) {


Why this change?

mxm · 2019-10-17T10:49:27Z

runners/flink/flink_runner.gradle

@@ -178,8 +201,8 @@ def createValidatesRunnerTask(Map m) {
 createValidatesRunnerTask(name: "validatesRunnerBatch", streaming: false)
 createValidatesRunnerTask(name: "validatesRunnerStreaming", streaming: true)

-task validatesRunner {
-  group = "Verification"
+tasks.register('validatesRunner') {


Why this change?

this is recommended approach since 5.1.x as it creates the task lazily

https://docs.gradle.org/current/userguide/task_configuration_avoidance.html

mxm · 2019-10-17T10:51:16Z

runners/flink/1.8/build.gradle

-  main_source_dirs = ["$basePath/src/main/java", "./src/main/java"]
-  test_source_dirs = ["$basePath/src/test/java", "./src/test/java"]
-  main_resources_dirs = ["$basePath/src/main/resources"]
-  test_resources_dirs = ["$basePath/src/test/resources"]


Could we keep the ability to override resource dirs?

mxm · 2019-10-17T10:51:46Z

runners/flink/1.7/build.gradle

-  main_source_dirs = ["$basePath/src/main/java", "./src/main/java"]
-  test_source_dirs = ["$basePath/src/test/java", "./src/test/java"]
-  main_resources_dirs = ["$basePath/src/main/resources"]
-  test_resources_dirs = ["$basePath/src/test/resources"]


Could we keep the ability to override resource dirs?

mxm · 2019-10-17T10:55:54Z

runners/flink/1.9/build.gradle

+  flink_version = '1.9.0'
+  // Version specific code overrides.
+  main_source_overrides = ["${basePath}/1.7/src/main/java", "${basePath}/1.8/src/main/java", './src/main/java']
+  test_source_overrides = ["${basePath}/1.7/src/test/java", "${basePath}/1.8/src/test/java", './src/test/java']


Not sure if that it easier to reason about the source, but at least we do not have code duplication. So looks fine for now.

mxm

This is practically ready. Going forward, we can merge this and address any follow ups related to documentation and build flexibility.

mxm · 2019-10-21T21:29:17Z

Thanks @dmvk.

mxm · 2019-10-21T21:29:25Z

Retest this please

mxm · 2019-10-21T21:29:47Z

Run ValidatesRunner

mxm · 2019-10-21T21:30:06Z

Run Java Flink PortableValidatesRunner Streaming

mxm · 2019-10-21T21:30:10Z

Run Java Flink PortableValidatesRunner Batch

sunjincheng121

+1

dmvk · 2019-10-22T08:22:34Z

@mxm Thanks! I'll follow-up with docs, build improvement suggestions and 1.9.1 upgrade.

mxm · 2019-10-23T17:23:45Z

Thanks for letting me know.

…

On 23.10.19 17:25, David Moravek wrote: ***@***.**** commented on this pull request. ------------------------------------------------------------------------ In runners/flink/flink_runner.gradle <#9296 (comment)>: > @@ -178,8 +201,8 @@ def createValidatesRunnerTask(Map m) { createValidatesRunnerTask(name: "validatesRunnerBatch", streaming: false) createValidatesRunnerTask(name: "validatesRunnerStreaming", streaming: true) -task validatesRunner { - group = "Verification" +tasks.register('validatesRunner') { this is recommended approach since 5.1.x as it creates the task lazily https://docs.gradle.org/current/userguide/task_configuration_avoidance.html — You are receiving this because you modified the open/close state. Reply to this email directly, view it on GitHub <#9296?email_source=notifications&email_token=AAGMMZMGWRCQKBO3FE6OPEDQQBUGRA5CNFSM4IKKKQD2YY3PNVWWK3TUL52HS4DFWFIHK3DMKJSXC5LFON2FEZLWNFSXPKTDN5WW2ZLOORPWSZGOCI6OV7I#discussion_r338117884>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAGMMZMUHND43ANPDMCXMFTQQBUGRANCNFSM4IKKKQDQ>.

dmvk force-pushed the BEAM-7730 branch 2 times, most recently from 0f5805f to 09fc860 Compare August 12, 2019 11:13

sunjincheng121 reviewed Aug 26, 2019

View reviewed changes

dmvk force-pushed the BEAM-7730 branch from 09fc860 to 6d586a3 Compare August 26, 2019 12:57

sunjincheng121 mentioned this pull request Sep 3, 2019

[BEAM-8162] Encode keys as NESTED for flink keyselector #9464

Merged

dmvk force-pushed the BEAM-7730 branch from 6d586a3 to 3a7f851 Compare September 4, 2019 15:14

dmvk changed the title ~~WIP: [BEAM-7730] Introduce Flink 1.9 Runner~~ [BEAM-7730] Introduce Flink 1.9 Runner Sep 4, 2019

dmvk force-pushed the BEAM-7730 branch from 3a7f851 to 16d02ff Compare September 5, 2019 07:43

dmvk requested a review from mxm September 5, 2019 09:17

mxm requested changes Sep 6, 2019

View reviewed changes

tweise requested changes Sep 6, 2019

View reviewed changes

mxm reviewed Sep 8, 2019

View reviewed changes

mxm mentioned this pull request Sep 24, 2019

[BEAM-8157] Ensure key encoding for state requests is consistent across SDKs #9484

Merged

dmvk force-pushed the BEAM-7730 branch 2 times, most recently from c89b232 to 791e3ee Compare October 8, 2019 13:42

mxm reviewed Oct 9, 2019

View reviewed changes

dmvk force-pushed the BEAM-7730 branch from 8d66d34 to 7bab7e9 Compare October 9, 2019 11:51

David Moravek added 3 commits October 10, 2019 11:06

[BEAM-7730] Introduce Flink 1.9 Runner.

f454f97

[BEAM-7730] Reuse sources overrides from previous flink versions.

4a64965

[BEAM-7730] Fix spotless plugin for flink runner.

d34bbbd

dmvk force-pushed the BEAM-7730 branch from 7bab7e9 to d34bbbd Compare October 10, 2019 09:55

mxm reviewed Oct 17, 2019

View reviewed changes

mxm approved these changes Oct 21, 2019

View reviewed changes

sunjincheng121 approved these changes Oct 22, 2019

View reviewed changes

mxm merged commit 686bfd2 into apache:master Oct 22, 2019

dmvk mentioned this pull request Oct 23, 2019

[BEAM-7730] Add ability to override resources for a specific flink ru… #9857

Merged

[BEAM-7730] Introduce Flink 1.9 Runner #9296

[BEAM-7730] Introduce Flink 1.9 Runner #9296

Conversation

dmvk commented Aug 8, 2019

sunjincheng121 left a comment • edited Loading

Choose a reason for hiding this comment

dmvk commented Aug 26, 2019

dmvk commented Aug 26, 2019

sunjincheng121 commented Aug 27, 2019

dmvk commented Aug 27, 2019

sunjincheng121 commented Aug 30, 2019

sunjincheng121 commented Sep 3, 2019

dmvk commented Sep 4, 2019

dmvk commented Sep 5, 2019

sunjincheng121 commented Sep 5, 2019

mxm 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tweise left a comment

Choose a reason for hiding this comment

dmvk commented Sep 7, 2019 • edited Loading

dmvk commented Sep 7, 2019

Choose a reason for hiding this comment

mxm commented Sep 8, 2019

tweise commented Sep 8, 2019 • edited Loading

tweise commented Sep 8, 2019

tweise commented Sep 8, 2019

mxm commented Sep 12, 2019

sunjincheng121 commented Sep 27, 2019

Choose a reason for hiding this comment

dmvk commented Oct 10, 2019

dmvk commented Oct 11, 2019

sunjincheng121 commented Oct 14, 2019

mxm 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

mxm left a comment

Choose a reason for hiding this comment

mxm commented Oct 21, 2019

mxm commented Oct 21, 2019

mxm commented Oct 21, 2019

mxm commented Oct 21, 2019

mxm commented Oct 21, 2019

sunjincheng121 left a comment

Choose a reason for hiding this comment

dmvk commented Oct 22, 2019

mxm commented Oct 23, 2019 via email

sunjincheng121 left a comment •

edited

Loading

dmvk commented Sep 7, 2019 •

edited

Loading

tweise commented Sep 8, 2019 •

edited

Loading