From ce27e30b074be60bbf394047ffecdfbbe5e4b5ed Mon Sep 17 00:00:00 2001 From: Maximilian Michels Date: Fri, 20 Sep 2019 18:42:35 -0700 Subject: [PATCH 1/2] [BEAM-7962] Drop support for Flink 1.5 and 1.6 Flink 1.9 is now released and Beam 2.17.0 is going to support it. Since the Flink community only supports the last two Flink releases, it is now time to drop at least 1.5 and 1.6. As a follow-up, we will get rid of Flink 1.5/1.6 specific workarounds, e.g. make use of Flink's preSnapshotBarrier in AbstractStreamOperator which removes the needs to buffer elements during a snapshot. Build time should decrease by several minutes. --- .../jenkins/CommonTestProperties.groovy | 2 +- .../job_PostCommit_Java_Nexmark_Flink.groovy | 8 ++--- ...PortableValidatesRunner_Flink_Batch.groovy | 2 +- ...ableValidatesRunner_Flink_Streaming.groovy | 2 +- ...stCommit_Java_ValidatesRunner_Flink.groovy | 2 +- .../beam/gradle/BeamModulePlugin.groovy | 6 ++-- examples/java/build.gradle | 2 +- examples/kotlin/build.gradle | 2 +- project-mappings | 6 ++-- release/build.gradle | 2 +- release/src/main/scripts/run_rc_validation.sh | 2 +- runners/flink/1.5/build.gradle | 34 ------------------- .../1.5/job-server-container/build.gradle | 26 -------------- runners/flink/1.5/job-server/build.gradle | 31 ----------------- runners/flink/1.6/build.gradle | 34 ------------------- .../1.6/job-server-container/build.gradle | 26 -------------- runners/flink/1.6/job-server/build.gradle | 31 ----------------- runners/flink/1.7/build.gradle | 4 +-- .../types/CoderTypeSerializer.java | 0 .../types/EncodedValueSerializer.java | 0 .../FlinkBroadcastStateInternalsTest.java | 0 .../streaming/FlinkStateInternalsTest.java | 0 .../types/CoderTypeSerializerTest.java | 0 runners/flink/flink_runner.gradle | 11 ++---- sdks/go/test/build.gradle | 4 +-- sdks/java/testing/nexmark/build.gradle | 2 +- .../runners/portability/flink_runner.py | 2 +- .../test-suites/portable/py2/build.gradle | 14 ++++---- .../test-suites/portable/py35/build.gradle | 2 +- .../test-suites/portable/py36/build.gradle | 2 +- .../test-suites/portable/py37/build.gradle | 2 +- settings.gradle | 8 ----- website/src/contribute/release-guide.md | 2 +- website/src/documentation/dsls/sql/shell.md | 6 ++-- website/src/documentation/runners/flink.md | 4 +-- website/src/documentation/sdks/nexmark.md | 6 ++-- 36 files changed, 45 insertions(+), 242 deletions(-) delete mode 100644 runners/flink/1.5/build.gradle delete mode 100644 runners/flink/1.5/job-server-container/build.gradle delete mode 100644 runners/flink/1.5/job-server/build.gradle delete mode 100644 runners/flink/1.6/build.gradle delete mode 100644 runners/flink/1.6/job-server-container/build.gradle delete mode 100644 runners/flink/1.6/job-server/build.gradle rename runners/flink/{1.5 => 1.7}/src/main/java/org/apache/beam/runners/flink/translation/types/CoderTypeSerializer.java (100%) rename runners/flink/{1.5 => 1.7}/src/main/java/org/apache/beam/runners/flink/translation/types/EncodedValueSerializer.java (100%) rename runners/flink/{1.5 => 1.7}/src/test/java/org/apache/beam/runners/flink/streaming/FlinkBroadcastStateInternalsTest.java (100%) rename runners/flink/{1.5 => 1.7}/src/test/java/org/apache/beam/runners/flink/streaming/FlinkStateInternalsTest.java (100%) rename runners/flink/{1.5 => 1.7}/src/test/java/org/apache/beam/runners/flink/translation/types/CoderTypeSerializerTest.java (100%) diff --git a/.test-infra/jenkins/CommonTestProperties.groovy b/.test-infra/jenkins/CommonTestProperties.groovy index 14dd7b075fed..203e3981b850 100644 --- a/.test-infra/jenkins/CommonTestProperties.groovy +++ b/.test-infra/jenkins/CommonTestProperties.groovy @@ -35,7 +35,7 @@ class CommonTestProperties { JAVA: [ DATAFLOW: ":runners:google-cloud-dataflow-java", SPARK: ":runners:spark", - FLINK: ":runners:flink:1.5", + FLINK: ":runners:flink:1.8", DIRECT: ":runners:direct-java" ], PYTHON: [ diff --git a/.test-infra/jenkins/job_PostCommit_Java_Nexmark_Flink.groovy b/.test-infra/jenkins/job_PostCommit_Java_Nexmark_Flink.groovy index 065f74d57f3c..038d6b3ac70c 100644 --- a/.test-infra/jenkins/job_PostCommit_Java_Nexmark_Flink.groovy +++ b/.test-infra/jenkins/job_PostCommit_Java_Nexmark_Flink.groovy @@ -40,7 +40,7 @@ NoPhraseTriggeringPostCommitBuilder.postCommitJob('beam_PostCommit_Java_Nexmark_ rootBuildScriptDir(commonJobProperties.checkoutDir) tasks(':sdks:java:testing:nexmark:run') commonJobProperties.setGradleSwitches(delegate) - switches('-Pnexmark.runner=":runners:flink:1.5"' + + switches('-Pnexmark.runner=":runners:flink:1.8"' + ' -Pnexmark.args="' + [NexmarkBigqueryProperties.nexmarkBigQueryArgs, '--streaming=false', @@ -55,7 +55,7 @@ NoPhraseTriggeringPostCommitBuilder.postCommitJob('beam_PostCommit_Java_Nexmark_ rootBuildScriptDir(commonJobProperties.checkoutDir) tasks(':sdks:java:testing:nexmark:run') commonJobProperties.setGradleSwitches(delegate) - switches('-Pnexmark.runner=":runners:flink:1.5"' + + switches('-Pnexmark.runner=":runners:flink:1.8"' + ' -Pnexmark.args="' + [NexmarkBigqueryProperties.nexmarkBigQueryArgs, '--streaming=true', @@ -70,7 +70,7 @@ NoPhraseTriggeringPostCommitBuilder.postCommitJob('beam_PostCommit_Java_Nexmark_ rootBuildScriptDir(commonJobProperties.checkoutDir) tasks(':sdks:java:testing:nexmark:run') commonJobProperties.setGradleSwitches(delegate) - switches('-Pnexmark.runner=":runners:flink:1.5"' + + switches('-Pnexmark.runner=":runners:flink:1.8"' + ' -Pnexmark.args="' + [NexmarkBigqueryProperties.nexmarkBigQueryArgs, '--queryLanguage=sql', @@ -85,7 +85,7 @@ NoPhraseTriggeringPostCommitBuilder.postCommitJob('beam_PostCommit_Java_Nexmark_ rootBuildScriptDir(commonJobProperties.checkoutDir) tasks(':sdks:java:testing:nexmark:run') commonJobProperties.setGradleSwitches(delegate) - switches('-Pnexmark.runner=":runners:flink:1.5"' + + switches('-Pnexmark.runner=":runners:flink:1.8"' + ' -Pnexmark.args="' + [NexmarkBigqueryProperties.nexmarkBigQueryArgs, '--queryLanguage=sql', diff --git a/.test-infra/jenkins/job_PostCommit_Java_PortableValidatesRunner_Flink_Batch.groovy b/.test-infra/jenkins/job_PostCommit_Java_PortableValidatesRunner_Flink_Batch.groovy index b8a59b3702c9..4da75f9171a0 100644 --- a/.test-infra/jenkins/job_PostCommit_Java_PortableValidatesRunner_Flink_Batch.groovy +++ b/.test-infra/jenkins/job_PostCommit_Java_PortableValidatesRunner_Flink_Batch.groovy @@ -36,7 +36,7 @@ PostcommitJobBuilder.postCommitJob('beam_PostCommit_Java_PVR_Flink_Batch', steps { gradle { rootBuildScriptDir(commonJobProperties.checkoutDir) - tasks(':runners:flink:1.5:job-server:validatesPortableRunnerBatch') + tasks(':runners:flink:1.8:job-server:validatesPortableRunnerBatch') commonJobProperties.setGradleSwitches(delegate) } } diff --git a/.test-infra/jenkins/job_PostCommit_Java_PortableValidatesRunner_Flink_Streaming.groovy b/.test-infra/jenkins/job_PostCommit_Java_PortableValidatesRunner_Flink_Streaming.groovy index 6a48e31a3c85..612c1542dfbf 100644 --- a/.test-infra/jenkins/job_PostCommit_Java_PortableValidatesRunner_Flink_Streaming.groovy +++ b/.test-infra/jenkins/job_PostCommit_Java_PortableValidatesRunner_Flink_Streaming.groovy @@ -36,7 +36,7 @@ PostcommitJobBuilder.postCommitJob('beam_PostCommit_Java_PVR_Flink_Streaming', steps { gradle { rootBuildScriptDir(commonJobProperties.checkoutDir) - tasks(':runners:flink:1.5:job-server:validatesPortableRunnerStreaming') + tasks(':runners:flink:1.8:job-server:validatesPortableRunnerStreaming') commonJobProperties.setGradleSwitches(delegate) } } diff --git a/.test-infra/jenkins/job_PostCommit_Java_ValidatesRunner_Flink.groovy b/.test-infra/jenkins/job_PostCommit_Java_ValidatesRunner_Flink.groovy index 2c8b21269ad0..d5e6da9a6d3c 100644 --- a/.test-infra/jenkins/job_PostCommit_Java_ValidatesRunner_Flink.groovy +++ b/.test-infra/jenkins/job_PostCommit_Java_ValidatesRunner_Flink.groovy @@ -37,7 +37,7 @@ PostcommitJobBuilder.postCommitJob('beam_PostCommit_Java_ValidatesRunner_Flink', steps { gradle { rootBuildScriptDir(commonJobProperties.checkoutDir) - tasks(':runners:flink:1.5:validatesRunner') + tasks(':runners:flink:1.8:validatesRunner') commonJobProperties.setGradleSwitches(delegate) } } diff --git a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy index 3d5e2d6e4021..c93cc6e80e55 100644 --- a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy +++ b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy @@ -1336,7 +1336,7 @@ class BeamModulePlugin implements Plugin { } if (runner?.equalsIgnoreCase('flink')) { - testRuntime it.project(path: ":runners:flink:1.5", configuration: 'testRuntime') + testRuntime it.project(path: ":runners:flink:1.8", configuration: 'testRuntime') } if (runner?.equalsIgnoreCase('spark')) { @@ -1808,7 +1808,7 @@ class BeamModulePlugin implements Plugin { dependsOn setupTask // We need flink-job-server-container dependency since Python PortableRunner automatically // brings the flink-job-server-container up when --job_endpoint is not specified. - dependsOn ':runners:flink:1.5:job-server-container:docker' + dependsOn ':runners:flink:1.8:job-server-container:docker' } mainTask.dependsOn pythonTask cleanupTask.mustRunAfter pythonTask @@ -1990,7 +1990,7 @@ class BeamModulePlugin implements Plugin { project.task('portableWordCount' + (isStreaming ? 'Streaming' : 'Batch')) { dependsOn = ['installGcpTest'] mustRunAfter = [ - ':runners:flink:1.5:job-server-container:docker', + ':runners:flink:1.8:job-server-container:docker', ':sdks:python:container:py2:docker', ':sdks:python:container:py35:docker', ':sdks:python:container:py36:docker', diff --git a/examples/java/build.gradle b/examples/java/build.gradle index 7b817bf05e3f..87586889a097 100644 --- a/examples/java/build.gradle +++ b/examples/java/build.gradle @@ -78,7 +78,7 @@ dependencies { // https://issues.apache.org/jira/browse/BEAM-3583 // apexRunnerPreCommit project(":runners:apex") directRunnerPreCommit project(path: ":runners:direct-java", configuration: "shadow") - flinkRunnerPreCommit project(":runners:flink:1.5") + flinkRunnerPreCommit project(":runners:flink:1.8") // TODO: Make the netty version used configurable, we add netty-all 4.1.17.Final so it appears on the classpath // before 4.1.8.Final defined by Apache Beam sparkRunnerPreCommit "io.netty:netty-all:4.1.17.Final" diff --git a/examples/kotlin/build.gradle b/examples/kotlin/build.gradle index b0fa7f377f90..b3fcc4cdbac2 100644 --- a/examples/kotlin/build.gradle +++ b/examples/kotlin/build.gradle @@ -81,7 +81,7 @@ dependencies { // https://issues.apache.org/jira/browse/BEAM-3583 // apexRunnerPreCommit project(":runners:apex") directRunnerPreCommit project(path: ":runners:direct-java", configuration: "shadow") - flinkRunnerPreCommit project(":runners:flink:1.5") + flinkRunnerPreCommit project(":runners:flink:1.8") // TODO: Make the netty version used configurable, we add netty-all 4.1.17.Final so it appears on the classpath // before 4.1.8.Final defined by Apache Beam sparkRunnerPreCommit "io.netty:netty-all:4.1.17.Final" diff --git a/project-mappings b/project-mappings index db6165343b2f..f8ac258aa415 100644 --- a/project-mappings +++ b/project-mappings @@ -94,9 +94,9 @@ :beam-runners-google-cloud-dataflow-java-examples :runners:google-cloud-dataflow-java:examples :beam-runners-google-cloud-dataflow-java :runners:google-cloud-dataflow-java :beam-runners-gearpump :runners:gearpump -:beam-runners-flink_2.11-job-server-container :runners:flink:1.5:job-server-container -:beam-runners-flink_2.11-job-server :runners:flink:1.5:job-server -:beam-runners-flink_2.11 :runners:flink:1.5 +:beam-runners-flink_2.11-job-server-container :runners:flink:1.8:job-server-container +:beam-runners-flink_2.11-job-server :runners:flink:1.8:job-server +:beam-runners-flink_2.11 :runners:flink:1.8 :beam-runners-flink-1.7-job-server-container :runners:flink:1.7:job-server-container :beam-runners-flink-1.7-job-server :runners:flink:1.7:job-server :beam-runners-flink-1.7 :runners:flink:1.7 diff --git a/release/build.gradle b/release/build.gradle index c5228babf52e..44e9f986ed48 100644 --- a/release/build.gradle +++ b/release/build.gradle @@ -34,7 +34,7 @@ task runJavaExamplesValidationTask { dependsOn ":runners:google-cloud-dataflow-java:runQuickstartJavaDataflow" dependsOn ":runners:apex:runQuickstartJavaApex" dependsOn ":runners:spark:runQuickstartJavaSpark" - dependsOn ":runners:flink:1.5:runQuickstartJavaFlinkLocal" + dependsOn ":runners:flink:1.8:runQuickstartJavaFlinkLocal" dependsOn ":runners:direct-java:runMobileGamingJavaDirect" dependsOn ":runners:google-cloud-dataflow-java:runMobileGamingJavaDataflow" } diff --git a/release/src/main/scripts/run_rc_validation.sh b/release/src/main/scripts/run_rc_validation.sh index 02464e0d97ca..0057f24c30f5 100755 --- a/release/src/main/scripts/run_rc_validation.sh +++ b/release/src/main/scripts/run_rc_validation.sh @@ -209,7 +209,7 @@ if [[ "$java_quickstart_flink_local" = true ]]; then echo "*************************************************************" echo "* Running Java Quickstart with Flink local runner" echo "*************************************************************" - ./gradlew :runners:flink:1.5:runQuickstartJavaFlinkLocal \ + ./gradlew :runners:flink:1.8:runQuickstartJavaFlinkLocal \ -Prepourl=${REPO_URL} \ -Pver=${RELEASE_VER} else diff --git a/runners/flink/1.5/build.gradle b/runners/flink/1.5/build.gradle deleted file mode 100644 index b06339534ddc..000000000000 --- a/runners/flink/1.5/build.gradle +++ /dev/null @@ -1,34 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * License); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an AS IS BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -def basePath = '..' - -/* All properties required for loading the Flink build script. */ -project.ext { - // Set the version of all Flink-related dependencies here. - flink_version = '1.5.6' - // Main source directory and Flink version specific code. - 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"] - archives_base_name = 'beam-runners-flink_2.11' -} - -// Load the main build script which contains all build logic. -apply from: "$basePath/flink_runner.gradle" diff --git a/runners/flink/1.5/job-server-container/build.gradle b/runners/flink/1.5/job-server-container/build.gradle deleted file mode 100644 index afdb68a0fc91..000000000000 --- a/runners/flink/1.5/job-server-container/build.gradle +++ /dev/null @@ -1,26 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * License); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an AS IS BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -def basePath = '../../job-server-container' - -project.ext { - resource_path = basePath -} - -// Load the main build script which contains all build logic. -apply from: "$basePath/flink_job_server_container.gradle" diff --git a/runners/flink/1.5/job-server/build.gradle b/runners/flink/1.5/job-server/build.gradle deleted file mode 100644 index fbba7a34ac54..000000000000 --- a/runners/flink/1.5/job-server/build.gradle +++ /dev/null @@ -1,31 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * License); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an AS IS BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -def basePath = '../../job-server' - -project.ext { - // Look for the source code in the parent module - main_source_dirs = ["$basePath/src/main/java"] - test_source_dirs = ["$basePath/src/test/java"] - main_resources_dirs = ["$basePath/src/main/resources"] - test_resources_dirs = ["$basePath/src/test/resources"] - archives_base_name = 'beam-runners-flink_2.11-job-server' -} - -// Load the main build script which contains all build logic. -apply from: "$basePath/flink_job_server.gradle" diff --git a/runners/flink/1.6/build.gradle b/runners/flink/1.6/build.gradle deleted file mode 100644 index e8541cc4a4cd..000000000000 --- a/runners/flink/1.6/build.gradle +++ /dev/null @@ -1,34 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * License); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an AS IS BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -def basePath = '..' - -/* All properties required for loading the Flink build script */ -project.ext { - // Set the version of all Flink-related dependencies here. - flink_version = '1.6.4' - // Main source directory and Flink version specific code. - main_source_dirs = ["$basePath/src/main/java", "../1.5/src/main/java"] - test_source_dirs = ["$basePath/src/test/java", "../1.5/src/test/java"] - main_resources_dirs = ["$basePath/src/main/resources"] - test_resources_dirs = ["$basePath/src/test/resources"] - archives_base_name = 'beam-runners-flink-1.6' -} - -// Load the main build script which contains all build logic. -apply from: "$basePath/flink_runner.gradle" diff --git a/runners/flink/1.6/job-server-container/build.gradle b/runners/flink/1.6/job-server-container/build.gradle deleted file mode 100644 index afdb68a0fc91..000000000000 --- a/runners/flink/1.6/job-server-container/build.gradle +++ /dev/null @@ -1,26 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * License); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an AS IS BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -def basePath = '../../job-server-container' - -project.ext { - resource_path = basePath -} - -// Load the main build script which contains all build logic. -apply from: "$basePath/flink_job_server_container.gradle" diff --git a/runners/flink/1.6/job-server/build.gradle b/runners/flink/1.6/job-server/build.gradle deleted file mode 100644 index 39f1810da65b..000000000000 --- a/runners/flink/1.6/job-server/build.gradle +++ /dev/null @@ -1,31 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * License); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an AS IS BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -def basePath = '../../job-server' - -project.ext { - // Look for the source code in the parent module - main_source_dirs = ["$basePath/src/main/java"] - test_source_dirs = ["$basePath/src/test/java"] - main_resources_dirs = ["$basePath/src/main/resources"] - test_resources_dirs = ["$basePath/src/test/resources"] - archives_base_name = 'beam-runners-flink-1.6-job-server' -} - -// Load the main build script which contains all build logic. -apply from: "$basePath/flink_job_server.gradle" diff --git a/runners/flink/1.7/build.gradle b/runners/flink/1.7/build.gradle index 9029153aef44..40132477c391 100644 --- a/runners/flink/1.7/build.gradle +++ b/runners/flink/1.7/build.gradle @@ -23,8 +23,8 @@ project.ext { // Set the version of all Flink-related dependencies here. flink_version = '1.7.2' // Main source directory and Flink version specific code. - main_source_dirs = ["$basePath/src/main/java", "../1.5/src/main/java"] - test_source_dirs = ["$basePath/src/test/java", "../1.5/src/test/java"] + 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"] archives_base_name = 'beam-runners-flink-1.7' diff --git a/runners/flink/1.5/src/main/java/org/apache/beam/runners/flink/translation/types/CoderTypeSerializer.java b/runners/flink/1.7/src/main/java/org/apache/beam/runners/flink/translation/types/CoderTypeSerializer.java similarity index 100% rename from runners/flink/1.5/src/main/java/org/apache/beam/runners/flink/translation/types/CoderTypeSerializer.java rename to runners/flink/1.7/src/main/java/org/apache/beam/runners/flink/translation/types/CoderTypeSerializer.java diff --git a/runners/flink/1.5/src/main/java/org/apache/beam/runners/flink/translation/types/EncodedValueSerializer.java b/runners/flink/1.7/src/main/java/org/apache/beam/runners/flink/translation/types/EncodedValueSerializer.java similarity index 100% rename from runners/flink/1.5/src/main/java/org/apache/beam/runners/flink/translation/types/EncodedValueSerializer.java rename to runners/flink/1.7/src/main/java/org/apache/beam/runners/flink/translation/types/EncodedValueSerializer.java diff --git a/runners/flink/1.5/src/test/java/org/apache/beam/runners/flink/streaming/FlinkBroadcastStateInternalsTest.java b/runners/flink/1.7/src/test/java/org/apache/beam/runners/flink/streaming/FlinkBroadcastStateInternalsTest.java similarity index 100% rename from runners/flink/1.5/src/test/java/org/apache/beam/runners/flink/streaming/FlinkBroadcastStateInternalsTest.java rename to runners/flink/1.7/src/test/java/org/apache/beam/runners/flink/streaming/FlinkBroadcastStateInternalsTest.java diff --git a/runners/flink/1.5/src/test/java/org/apache/beam/runners/flink/streaming/FlinkStateInternalsTest.java b/runners/flink/1.7/src/test/java/org/apache/beam/runners/flink/streaming/FlinkStateInternalsTest.java similarity index 100% rename from runners/flink/1.5/src/test/java/org/apache/beam/runners/flink/streaming/FlinkStateInternalsTest.java rename to runners/flink/1.7/src/test/java/org/apache/beam/runners/flink/streaming/FlinkStateInternalsTest.java diff --git a/runners/flink/1.5/src/test/java/org/apache/beam/runners/flink/translation/types/CoderTypeSerializerTest.java b/runners/flink/1.7/src/test/java/org/apache/beam/runners/flink/translation/types/CoderTypeSerializerTest.java similarity index 100% rename from runners/flink/1.5/src/test/java/org/apache/beam/runners/flink/translation/types/CoderTypeSerializerTest.java rename to runners/flink/1.7/src/test/java/org/apache/beam/runners/flink/translation/types/CoderTypeSerializerTest.java diff --git a/runners/flink/flink_runner.gradle b/runners/flink/flink_runner.gradle index 9b3d9674a26f..7d1c9aabdf56 100644 --- a/runners/flink/flink_runner.gradle +++ b/runners/flink/flink_runner.gradle @@ -83,14 +83,7 @@ test { } // TODO Running tests of all Flink versions in parallel can be too harsh on Jenkins memory // Run them serially for now, to avoid "Exit code 137", i.e. Jenkins host killing the Gradle test process - if (project.path == ":runners:flink:1.6") { - mustRunAfter(":runners:flink:1.5:test") - } else if (project.path == ":runners:flink:1.7") { - mustRunAfter(":runners:flink:1.5:test") - mustRunAfter(":runners:flink:1.6:test") - } else if (project.path == ":runners:flink:1.8") { - mustRunAfter(":runners:flink:1.5:test") - mustRunAfter(":runners:flink:1.6:test") + if (project.path == ":runners:flink:1.8") { mustRunAfter(":runners:flink:1.7:test") } } @@ -190,5 +183,5 @@ task validatesRunner { dependsOn validatesRunnerStreaming } -// Generates :runners:flink:1.5:runQuickstartJavaFlinkLocal +// Generates :runners:flink:1.8:runQuickstartJavaFlinkLocal createJavaExamplesArchetypeValidationTask(type: 'Quickstart', runner: 'FlinkLocal') diff --git a/sdks/go/test/build.gradle b/sdks/go/test/build.gradle index b12d3f6f8094..77fd3bef6d0d 100644 --- a/sdks/go/test/build.gradle +++ b/sdks/go/test/build.gradle @@ -49,12 +49,12 @@ golang { task flinkValidatesRunner { dependsOn ":sdks:go:test:goBuild" - dependsOn ":runners:flink:1.5:job-server:shadowJar" + dependsOn ":runners:flink:1.8:job-server:shadowJar" doLast { def options = [ "--runner flink", "--parallel 1", // prevent memory overuse - "--flink_job_server_jar ${project(":runners:flink:1.5:job-server").shadowJar.archivePath}", + "--flink_job_server_jar ${project(":runners:flink:1.8:job-server").shadowJar.archivePath}", ] exec { executable "sh" diff --git a/sdks/java/testing/nexmark/build.gradle b/sdks/java/testing/nexmark/build.gradle index d44ccbe1286b..a6378cf0b0cd 100644 --- a/sdks/java/testing/nexmark/build.gradle +++ b/sdks/java/testing/nexmark/build.gradle @@ -102,7 +102,7 @@ if (shouldProvideSpark) { // // Parameters: // -Pnexmark.runner -// Specify a runner subproject, such as ":runners:spark" or ":runners:flink:1.5" +// Specify a runner subproject, such as ":runners:spark" or ":runners:flink:1.8" // Defaults to ":runners:direct-java" // // -Pnexmark.args diff --git a/sdks/python/apache_beam/runners/portability/flink_runner.py b/sdks/python/apache_beam/runners/portability/flink_runner.py index a904a3a6a390..97691da39a37 100644 --- a/sdks/python/apache_beam/runners/portability/flink_runner.py +++ b/sdks/python/apache_beam/runners/portability/flink_runner.py @@ -24,7 +24,7 @@ from apache_beam.runners.portability import job_server from apache_beam.runners.portability import portable_runner -PUBLISHED_FLINK_VERSIONS = ['1.6', '1.7', '1.8'] +PUBLISHED_FLINK_VERSIONS = ['1.7', '1.8'] class FlinkRunner(portable_runner.PortableRunner): diff --git a/sdks/python/test-suites/portable/py2/build.gradle b/sdks/python/test-suites/portable/py2/build.gradle index 9a87f7036206..53cd7287fcdf 100644 --- a/sdks/python/test-suites/portable/py2/build.gradle +++ b/sdks/python/test-suites/portable/py2/build.gradle @@ -28,7 +28,7 @@ def pythonRootDir = "${rootDir}/sdks/python" addPortableWordCountTasks() task preCommitPy2() { - dependsOn ':runners:flink:1.5:job-server-container:docker' + dependsOn ':runners:flink:1.8:job-server-container:docker' dependsOn ':sdks:python:container:py2:docker' dependsOn portableWordCountBatch dependsOn portableWordCountStreaming @@ -43,12 +43,12 @@ task preCommitPy2() { // 2. Either a) or b) // a) If you want the Job Server to run in a Docker container: // -// ./gradlew :runners:flink:1.5:job-server-container:docker +// ./gradlew :runners:flink:1.8:job-server-container:docker // // b) Otherwise, start a local JobService, for example, the Portable Flink runner // (in a separate shell since it continues to run): // -// ./gradlew :runners:flink:1.5:job-server:runShadow +// ./gradlew :runners:flink:1.8:job-server:runShadow // // Then you can run this example: // @@ -86,7 +86,7 @@ task crossLanguagePythonJavaDirect { task crossLanguagePythonJavaFlink { dependsOn 'setupVirtualenv' - dependsOn ':runners:flink:1.5:job-server-container:docker' + dependsOn ':runners:flink:1.8:job-server-container:docker' dependsOn ':sdks:python:container:py2:docker' dependsOn ':sdks:java:container:docker' dependsOn ':sdks:java:testing:expansion-service:buildTestExpansionServiceJar' @@ -111,7 +111,7 @@ task crossLanguagePythonJavaFlink { task crossLanguagePortableWordCount { dependsOn 'setupVirtualenv' - dependsOn ':runners:flink:1.5:job-server-container:docker' + dependsOn ':runners:flink:1.8:job-server-container:docker' dependsOn ':sdks:python:container:py2:docker' dependsOn ':sdks:java:container:docker' dependsOn ':sdks:java:testing:expansion-service:buildTestExpansionServiceJar' @@ -217,7 +217,7 @@ def flinkCompatibilityMatrix = { extra_experiments.add('pre_optimize=all') tasks.create(name: name) { dependsOn 'setupVirtualenv' - dependsOn ':runners:flink:1.5:job-server:shadowJar' + dependsOn ':runners:flink:1.8:job-server:shadowJar' if (workerType.toLowerCase() == 'docker') dependsOn ':sdks:python:container:py2:docker' else if (workerType.toLowerCase() == 'process') @@ -225,7 +225,7 @@ def flinkCompatibilityMatrix = { doLast { exec { executable 'sh' - args '-c', ". ${envdir}/bin/activate && cd ${pythonRootDir} && pip install -e .[test] && python -m apache_beam.runners.portability.flink_runner_test --flink_job_server_jar=${project(":runners:flink:1.5:job-server:").shadowJar.archivePath} --environment_type=${workerType} ${environment_config} ${streaming ? '--streaming' : ''} ${extra_experiments ? '--extra_experiments=' + extra_experiments.join(',') : ''}" + args '-c', ". ${envdir}/bin/activate && cd ${pythonRootDir} && pip install -e .[test] && python -m apache_beam.runners.portability.flink_runner_test --flink_job_server_jar=${project(":runners:flink:1.8:job-server:").shadowJar.archivePath} --environment_type=${workerType} ${environment_config} ${streaming ? '--streaming' : ''} ${extra_experiments ? '--extra_experiments=' + extra_experiments.join(',') : ''}" } } } diff --git a/sdks/python/test-suites/portable/py35/build.gradle b/sdks/python/test-suites/portable/py35/build.gradle index 389fbb4d70f5..2b38af963e14 100644 --- a/sdks/python/test-suites/portable/py35/build.gradle +++ b/sdks/python/test-suites/portable/py35/build.gradle @@ -24,7 +24,7 @@ pythonVersion = '3.5' addPortableWordCountTasks() task preCommitPy35() { - dependsOn ':runners:flink:1.5:job-server-container:docker' + dependsOn ':runners:flink:1.8:job-server-container:docker' dependsOn ':sdks:python:container:py35:docker' dependsOn portableWordCountBatch dependsOn portableWordCountStreaming diff --git a/sdks/python/test-suites/portable/py36/build.gradle b/sdks/python/test-suites/portable/py36/build.gradle index 8a4d947c4eaa..8dd7a72fa2b5 100644 --- a/sdks/python/test-suites/portable/py36/build.gradle +++ b/sdks/python/test-suites/portable/py36/build.gradle @@ -24,7 +24,7 @@ pythonVersion = '3.6' addPortableWordCountTasks() task preCommitPy36() { - dependsOn ':runners:flink:1.5:job-server-container:docker' + dependsOn ':runners:flink:1.8:job-server-container:docker' dependsOn ':sdks:python:container:py36:docker' dependsOn portableWordCountBatch dependsOn portableWordCountStreaming diff --git a/sdks/python/test-suites/portable/py37/build.gradle b/sdks/python/test-suites/portable/py37/build.gradle index 3bb103874250..72220bcadced 100644 --- a/sdks/python/test-suites/portable/py37/build.gradle +++ b/sdks/python/test-suites/portable/py37/build.gradle @@ -24,7 +24,7 @@ pythonVersion = '3.7' addPortableWordCountTasks() task preCommitPy37() { - dependsOn ':runners:flink:1.5:job-server-container:docker' + dependsOn ':runners:flink:1.8:job-server-container:docker' dependsOn ':sdks:python:container:py37:docker' dependsOn portableWordCountBatch dependsOn portableWordCountStreaming diff --git a/settings.gradle b/settings.gradle index 08303742b438..6a5c36119e53 100644 --- a/settings.gradle +++ b/settings.gradle @@ -31,14 +31,6 @@ include ":runners:core-java" include ":runners:direct-java" include ":runners:extensions-java:metrics" /* Begin Flink Runner related settings */ -// Flink 1.5 (with Scala 2.11 suffix) -include ":runners:flink:1.5" -include ":runners:flink:1.5:job-server" -include ":runners:flink:1.5:job-server-container" -// Flink 1.6 -include ":runners:flink:1.6" -include ":runners:flink:1.6:job-server" -include ":runners:flink:1.6:job-server-container" // Flink 1.7 include ":runners:flink:1.7" include ":runners:flink:1.7:job-server" diff --git a/website/src/contribute/release-guide.md b/website/src/contribute/release-guide.md index ddf9e4ae13e6..03fb6ccba04c 100644 --- a/website/src/contribute/release-guide.md +++ b/website/src/contribute/release-guide.md @@ -993,7 +993,7 @@ _Note_: -Prepourl and -Pver can be found in the RC vote email sent by Release Ma ``` Flink Local Runner ``` - ./gradlew :runners:flink:1.5:runQuickstartJavaFlinkLocal \ + ./gradlew :runners:flink:1.8:runQuickstartJavaFlinkLocal \ -Prepourl=https://repository.apache.org/content/repositories/orgapachebeam-${KEY} \ -Pver=${RELEASE_VERSION} ``` diff --git a/website/src/documentation/dsls/sql/shell.md b/website/src/documentation/dsls/sql/shell.md index 69326e5699fa..1317575030bd 100644 --- a/website/src/documentation/dsls/sql/shell.md +++ b/website/src/documentation/dsls/sql/shell.md @@ -31,7 +31,7 @@ This page describes how to work with the shell, but does not focus on specific f To use Beam SQL shell, you must first clone the [Beam SDK repository](https://github.com/apache/beam). Then, from the root of the repository clone, execute the following commands to run the shell: ``` -./gradlew -p sdks/java/extensions/sql/shell -Pbeam.sql.shell.bundled=':runners:flink:1.5,:sdks:java:io:kafka' installDist +./gradlew -p sdks/java/extensions/sql/shell -Pbeam.sql.shell.bundled=':runners:flink:1.8,:sdks:java:io:kafka' installDist ./sdks/java/extensions/sql/shell/build/install/shell/bin/shell ``` @@ -119,7 +119,7 @@ By default, Beam uses the `DirectRunner` to run the pipeline on the machine wher 1. Make sure the SQL shell includes the desired runner. Add the corresponding project id to the `-Pbeam.sql.shell.bundled` parameter of the Gradle invocation ([source code](https://github.com/apache/beam/blob/master/sdks/java/extensions/sql/shell/build.gradle), [project ids](https://github.com/apache/beam/blob/master/settings.gradle)). For example, use the following command to include Flink runner and KafkaIO: ``` - ./gradlew -p sdks/java/extensions/sql/shell -Pbeam.sql.shell.bundled=':runners:flink:1.5,:sdks:java:io:kafka' installDist + ./gradlew -p sdks/java/extensions/sql/shell -Pbeam.sql.shell.bundled=':runners:flink:1.8,:sdks:java:io:kafka' installDist ``` _Note: You can bundle multiple runners (using a comma-separated list) or other additional components in the same manner. For example, you can add support for more I/Os._ @@ -145,7 +145,7 @@ To configure the runner, you must specify `PipelineOptions` by using the `SET` c You can also build your own standalone package for SQL shell using `distZip` or `distTar` tasks. For example: ``` -./gradlew -p sdks/java/extensions/sql/shell -Pbeam.sql.shell.bundled=':runners:flink:1.5,:sdks:java:io:kafka' distZip +./gradlew -p sdks/java/extensions/sql/shell -Pbeam.sql.shell.bundled=':runners:flink:1.8,:sdks:java:io:kafka' distZip ls ./sdks/java/extensions/sql/shell/build/distributions/ beam-sdks-java-extensions-sql-shell-2.6.0-SNAPSHOT.tar beam-sdks-java-extensions-sql-shell-2.6.0-SNAPSHOT.zip diff --git a/website/src/documentation/runners/flink.md b/website/src/documentation/runners/flink.md index 9bcb149454e2..6ceb92b244ca 100644 --- a/website/src/documentation/runners/flink.md +++ b/website/src/documentation/runners/flink.md @@ -250,7 +250,7 @@ download it on the [Downloads page]({{ site.baseurl available. To run a pipeline on an embedded Flink cluster: -1. Start the JobService endpoint: `./gradlew :runners:flink:1.5:job-server:runShadow` +1. Start the JobService endpoint: `./gradlew :runners:flink:1.8:job-server:runShadow` @@ -283,7 +283,7 @@ To run on a separate [Flink cluster](https://ci.apache.org/projects/flink/flink- 1. Start a Flink cluster which exposes the Rest interface on `localhost:8081` by default. -2. Start JobService with Flink Rest endpoint: `./gradlew :runners:flink:1.5:job-server:runShadow -PflinkMasterUrl=localhost:8081`. +2. Start JobService with Flink Rest endpoint: `./gradlew :runners:flink:1.8:job-server:runShadow -PflinkMasterUrl=localhost:8081`. 3. Submit the pipeline as above. diff --git a/website/src/documentation/sdks/nexmark.md b/website/src/documentation/sdks/nexmark.md index 22b381a10a7b..d5230da4fc08 100644 --- a/website/src/documentation/sdks/nexmark.md +++ b/website/src/documentation/sdks/nexmark.md @@ -149,7 +149,7 @@ When running via Gradle, the following two parameters control the execution: -P nexmark.runner The Gradle project name of the runner, such as ":runners:direct-java" or - ":runners:flink:1.5. The project names can be found in the root + ":runners:flink:1.8. The project names can be found in the root `settings.gradle`. Test data is deterministically synthesized on demand. The test @@ -557,7 +557,7 @@ Streaming Mode: Batch Mode: ./gradlew :sdks:java:testing:nexmark:run \ - -Pnexmark.runner=":runners:flink:1.5" \ + -Pnexmark.runner=":runners:flink:1.8" \ -Pnexmark.args=" --runner=FlinkRunner --suite=SMOKE @@ -570,7 +570,7 @@ Batch Mode: Streaming Mode: ./gradlew :sdks:java:testing:nexmark:run \ - -Pnexmark.runner=":runners:flink:1.5" \ + -Pnexmark.runner=":runners:flink:1.8" \ -Pnexmark.args=" --runner=FlinkRunner --suite=SMOKE From 661237c71ac7991f72d4a4dbce14e9ccf53a3d80 Mon Sep 17 00:00:00 2001 From: Maximilian Michels Date: Sat, 21 Sep 2019 15:00:54 -0700 Subject: [PATCH 2/2] [BEAM-7962] Update version compatibility section on Flink Runner page --- website/src/documentation/runners/flink.md | 25 ++++++++++++++++++++-- 1 file changed, 23 insertions(+), 2 deletions(-) diff --git a/website/src/documentation/runners/flink.md b/website/src/documentation/runners/flink.md index 6ceb92b244ca..cd354d6cc9d2 100644 --- a/website/src/documentation/runners/flink.md +++ b/website/src/documentation/runners/flink.md @@ -103,12 +103,33 @@ To find out which version of Flink is compatible with Beam please see the table Artifact Id - >=2.13.0 + 2.17.0 1.8.x beam-runners-flink-1.8 - >=2.10.0 + 1.7.x + beam-runners-flink-1.7 + + + 2.13.0 - 2.16.0 + 1.8.x + beam-runners-flink-1.8 + + + 1.7.x + beam-runners-flink-1.7 + + + 1.6.x + beam-runners-flink-1.6 + + + 1.5.x + beam-runners-flink_2.11 + + + 2.10.0 - 2.16.0 1.7.x beam-runners-flink-1.7