[BEAM-9295] Add Flink 1.10 build target and Make FlinkRunner compatible with Flink 1.10 #10945

sunjincheng121 · 2020-02-24T03:16:54Z

Add Flink 1.10 build target and Make FlinkRunner compatible with Flink 1.10

Post-Commit Tests Status (on master branch)

Lang	SDK	Apex	Dataflow	Gearpump	Samza
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.

sunjincheng121 · 2020-02-24T06:46:17Z

Run CommunityMetrics PreCommit

sunjincheng121 · 2020-02-24T07:06:14Z

Run Python2_PVR_Flink PreCommit

mxm

There a large amount of code duplication. Is that really necessary? AFAIK the changes between 1.9 and 1.10 should be minimal. Could you please revise this or explain why it is necessary?

sunjincheng121 · 2020-02-25T14:49:17Z

There a large amount of code duplication. Is that really necessary? AFAIK the changes between 1.9 and 1.10 should be minimal. Could you please revise this or explain why it is necessary?

There are a lot of changes related to the job client API(https://issues.apache.org/jira/browse/FLINK-14392, https://issues.apache.org/jira/browse/FLINK-14376) in 1.10. I have added comments at the header of each test case copied to 1.10 for the reason making a copy of them. Could you help to take a look at if the reason makes sense for you? :)

sunjincheng121 · 2020-02-25T14:51:51Z

Hi @mxm I found that the test case test_large_elements failed with execption:

java.io.IOException: Cannot write record to fresh sort buffer. Record too large.
        at org.apache.flink.runtime.operators.chaining.SynchronousChainedCombineDriver.collect(SynchronousChainedCombineDriver.java:176)
        at org.apache.flink.runtime.operators.util.metrics.CountingCollector.collect(CountingCollector.java:35)
        at org.apache.flink.runtime.operators.MapDriver.run(MapDriver.java:103)
        at org.apache.flink.runtime.operators.BatchTask.run(BatchTask.java:504)
        at org.apache.flink.runtime.operators.BatchTask.invoke(BatchTask.java:369)
        at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:707)
        at org.apache.flink.runtime.taskmanager.Task.run(Task.java:532)

I guess it's related to the feature "Unified Memory Configuration for TaskExecutors"(https://issues.apache.org/jira/browse/FLINK-13980) which is introduced in Flink 1.10. Before 1.10, the memory managed by the Flink's MemoryManager is calculated dynamically if not configured and I have checked that it will be about 2500 MB in my local machine. Since 1.10, it will be 128 MB if not configured (taskmanager.memory.managed.size).

I have performed a simple test and the failed test test_large_elements could pass after adding the following code at

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

Line 77 in 7b3a3fa

:

flinkConfiguration.set(TaskManagerOptions.MANAGED_MEMORY_SIZE, MemorySize.parse("2048m"));

I'm still investigating the best way to address this issue at Beam side. And appreciate if you have any suggestion on this :)

sunjincheng121 · 2020-02-26T05:41:33Z

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

@@ -67,6 +69,7 @@ static ExecutionEnvironment createBatchExecutionEnvironment(

    // depending on the master, create the right environment.
    if ("[local]".equals(flinkMasterHostPort)) {
+      flinkConfiguration.set(TaskManagerOptions.MANAGED_MEMORY_SIZE, MemorySize.parse("2048m"));


It has sets the default managed memory size to 128MB for MiniCluster in https://issues.apache.org/jira/browse/FLINK-15763. Have set it to a large value when the master host is [local]. Appreciate for any suggestions on a better way to address this issue.

sunjincheng121 · 2020-02-27T06:44:05Z

Run Java PreCommit

sunjincheng121 · 2020-02-27T07:47:52Z

Run Python PreCommit

sunjincheng121 · 2020-02-28T00:18:37Z

Test failure is not caused by the current PR.

mxm

How about removing 1.7 first and then revising this PR? I don't feel comfortable now with the amount of code duplication. Also I suspect that this PR will break support for 1.7 to the RemoteEnvironment removal.

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

runners/flink/1.10/src/test/java/org/apache/beam/runners/flink/FlinkSubmissionTest.java

sunjincheng121 · 2020-02-29T05:20:44Z

@mxm Thanks for the review. I have updated the PR according to your comments. Currently it only copied 3 tests and I think even we drop 1.7 support, we still need to copy these tests. What's your thought?

sunjincheng121 · 2020-02-29T05:26:22Z

Run Spotless PreCommit

sunjincheng121 · 2020-02-29T05:26:36Z

Run CommunityMetrics PreCommit

mxm

Thank you @sunjincheng121. It generally looks good to me. IMHO the test code duplication is not optimal and I would like to change that but this could also be done in a follow-up. Could you squash the commits?

...s/flink/1.10/src/test/java/org/apache/beam/runners/flink/FlinkExecutionEnvironmentsTest.java

.../1.10/src/test/java/org/apache/beam/runners/flink/FlinkPipelineExecutionEnvironmentTest.java

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

mxm · 2020-03-09T18:46:34Z

retest this please

sunjincheng121 · 2020-03-10T08:28:06Z

Thanks for the review @mxm ! And the suggestion about test case is great :) , I have update the PR accordingly :)

mxm · 2020-03-10T11:06:23Z

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

@@ -77,6 +67,7 @@ static ExecutionEnvironment createBatchExecutionEnvironment(

    // depending on the master, create the right environment.
    if ("[local]".equals(flinkMasterHostPort)) {
+      flinkConfiguration.setString("taskmanager.memory.managed.size", "2048m");


I have the feeling this won't be reliable enough. Why not instead taskmanager.memory.managed.fraction?

It will set the taskmanager.memory.managed.size as 128MB for MiniCluster if it's not set. I think set taskmanager.memory.managed.fraction" doesn't take effect here. Thoughts? :)

I'm hesitant with this default because it will always pre-allocate 2GB of memory which won't be used most of the time, except for the one large record test case you mentioned.

We could set I'd go for something like https://github.com/apache/flink/blob/42a56f4c75693773e21fa2dea45df640c2d7f9da/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/TaskExecutorProcessUtils.java#L287 based on the memory available.

Actually, that is what the Flink 1.8 code used to do: https://github.com/apache/flink/blob/60d9b96456f142f8d18d5882016840a00159403e/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java#L296

So let's just check the free memory and use a fraction for memory managed memory by default. What do you think?

Thanks @mxm, Sounds good to me ;)

Cool, thanks for the changes.

runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkExecutionEnvironmentsTest.java

...flink/src/test/java/org/apache/beam/runners/flink/FlinkPipelineExecutionEnvironmentTest.java

mxm · 2020-03-10T11:11:10Z

Thanks for making the adjustments. A couple more comments but otherwise it looks very good! Great work.

sunjincheng121 · 2020-03-10T14:23:36Z

Run Python PreCommit

mxm · 2020-03-10T16:31:09Z

Run Python PreCommit

…le with Flink 1.10

sunjincheng121 · 2020-03-11T05:16:24Z

Run Python2_PVR_Flink PreCommit

sunjincheng121 · 2020-03-11T05:43:57Z

Run Python PreCommit

sunjincheng121 · 2020-03-11T05:44:11Z

Run Python2_PVR_Flink PreCommit

sunjincheng121 · 2020-03-11T08:28:28Z

Run Python2_PVR_Flink PreCommit

sunjincheng121 · 2020-03-11T08:28:37Z

Run Java PreCommit

mxm · 2020-03-11T13:25:48Z

Run Python2_PVR_Flink PreCommit

mxm · 2020-03-11T13:25:53Z

Run Java PreCommit

mxm · 2020-03-11T13:26:19Z

Run Flink ValidatesRunner

mxm · 2020-03-11T13:26:40Z

Run Java Flink PortableValidatesRunner Streaming

mxm · 2020-03-11T13:26:44Z

Run Java Flink PortableValidatesRunner Batch

mxm · 2020-03-11T16:57:31Z

Run Java PreCommit

mxm

Thank you @sunjincheng121. I'll merge this once the tests pass.

probot-autolabeler bot added build examples flink infra java kotlin runners labels Feb 24, 2020

sunjincheng121 mentioned this pull request Feb 24, 2020

[BEAM-9298] Drop support for Flink 1.7 #10884

Merged

4 tasks

iemejia requested a review from mxm February 24, 2020 08:29

mxm reviewed Feb 25, 2020

View reviewed changes

sunjincheng121 commented Feb 26, 2020

View reviewed changes

sunjincheng121 force-pushed the BEAM-9285-PR branch from 2c7680b to 82b781c Compare February 27, 2020 03:51

mxm reviewed Feb 28, 2020

View reviewed changes

runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkExecutionEnvironments.java Show resolved Hide resolved

runners/flink/1.10/src/test/java/org/apache/beam/runners/flink/FlinkSubmissionTest.java Outdated Show resolved Hide resolved

sunjincheng121 force-pushed the BEAM-9285-PR branch from 82b781c to 3657ca4 Compare February 29, 2020 05:19

mxm self-requested a review March 5, 2020 14:29

mxm reviewed Mar 9, 2020

View reviewed changes

sunjincheng121 force-pushed the BEAM-9285-PR branch from 7d87a6d to 1f513ad Compare March 10, 2020 08:18

mxm reviewed Mar 10, 2020

View reviewed changes

sunjincheng121 force-pushed the BEAM-9285-PR branch from 1f513ad to 4cc537a Compare March 10, 2020 12:45

[BEAM-9295] Add Flink 1.10 build target and Make FlinkRunner compatib…

f91b390

…le with Flink 1.10

sunjincheng121 force-pushed the BEAM-9285-PR branch from 4cc537a to f91b390 Compare March 11, 2020 04:59

mxm approved these changes Mar 11, 2020

View reviewed changes

mxm merged commit be8b4ff into apache:master Mar 11, 2020

sunjincheng121 deleted the BEAM-9285-PR branch March 12, 2020 01:29

mxm mentioned this pull request Apr 1, 2020

[BEAM-9550] Increase JVM Metaspace size for the TaskExecutors. #11193

Merged

4 tasks

[BEAM-9295] Add Flink 1.10 build target and Make FlinkRunner compatible with Flink 1.10 #10945

[BEAM-9295] Add Flink 1.10 build target and Make FlinkRunner compatible with Flink 1.10 #10945

Conversation

sunjincheng121 commented Feb 24, 2020

Post-Commit Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

sunjincheng121 commented Feb 24, 2020

sunjincheng121 commented Feb 24, 2020

mxm left a comment

Choose a reason for hiding this comment

sunjincheng121 commented Feb 25, 2020

sunjincheng121 commented Feb 25, 2020

sunjincheng121 Feb 26, 2020

Choose a reason for hiding this comment

sunjincheng121 commented Feb 27, 2020

sunjincheng121 commented Feb 27, 2020

sunjincheng121 commented Feb 28, 2020

mxm left a comment

Choose a reason for hiding this comment

sunjincheng121 commented Feb 29, 2020

sunjincheng121 commented Feb 29, 2020

sunjincheng121 commented Feb 29, 2020

mxm left a comment

Choose a reason for hiding this comment

mxm commented Mar 9, 2020

sunjincheng121 commented Mar 10, 2020 • edited

mxm Mar 10, 2020

Choose a reason for hiding this comment

sunjincheng121 Mar 10, 2020

Choose a reason for hiding this comment

mxm Mar 10, 2020

Choose a reason for hiding this comment

sunjincheng121 Mar 11, 2020

Choose a reason for hiding this comment

mxm Mar 11, 2020

Choose a reason for hiding this comment

mxm commented Mar 10, 2020

sunjincheng121 commented Mar 10, 2020

mxm commented Mar 10, 2020

sunjincheng121 commented Mar 11, 2020

sunjincheng121 commented Mar 11, 2020

sunjincheng121 commented Mar 11, 2020

sunjincheng121 commented Mar 11, 2020

sunjincheng121 commented Mar 11, 2020

mxm commented Mar 11, 2020

mxm commented Mar 11, 2020

mxm commented Mar 11, 2020

mxm commented Mar 11, 2020

mxm commented Mar 11, 2020

mxm commented Mar 11, 2020

mxm left a comment

Choose a reason for hiding this comment

sunjincheng121 commented Mar 10, 2020 •

edited