Add support for Flink 1.20

.github/actions/setup-default-test-properties/test-properties.json

@@ -14,7 +14,8 @@
   },
   "JavaTestProperties": {
     "SUPPORTED_VERSIONS": ["8", "11", "17", "21"],
-    "FLINK_VERSIONS": ["1.17", "1.18", "1.19"],
+    "FLINK_VERSIONS": ["1.17", "1.18", "1.19", "1.20"],
+    "DEFAULT_FLINK_VERSION": "1.20",
     "SPARK_VERSIONS": ["2", "3"]
   },
   "GoTestProperties": {

.github/trigger_files/beam_PostCommit_Go_VR_Flink.json

@@ -1,5 +1,3 @@
 {
-  "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "modification": 1,
-  "https://github.com/apache/beam/pull/32648": "testing addition of Flink 1.19 support"
+  "": "testing addition of Flink 1.20 support"
 }

.github/trigger_files/beam_PostCommit_Java_Examples_Flink.json

@@ -1,3 +1,3 @@
 {
-    "https://github.com/apache/beam/pull/32648": "testing flink 1.19 support"
+    "": "testing Flink 1.20 support"
 }

.github/trigger_files/beam_PostCommit_Java_Jpms_Flink_Java11.json

@@ -1,3 +1,3 @@
 {
-    "https://github.com/apache/beam/pull/32648": "testing flink 1.19 support"
+    "": "Testing Flink 1.20 support"
 }

.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Flink.json

@@ -1,5 +1,3 @@
 {
-  "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "https://github.com/apache/beam/pull/31156": "noting that PR #31156 should run this test",
-  "https://github.com/apache/beam/pull/32648": "testing addition of Flink 1.19 support"
+  "": "testing addition of Flink 1.20 support"
 }

.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Flink_Java11.json

@@ -1,5 +1,3 @@
 {
-  "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "https://github.com/apache/beam/pull/31156": "noting that PR #31156 should run this test",
-  "https://github.com/apache/beam/pull/32648": "testing addition of Flink 1.19 support"
+  "": "testing addition of Flink 1.20 support"
 }

.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Flink_Java8.json

@@ -1,4 +1,3 @@
 {
-
-  "https://github.com/apache/beam/pull/32648": "testing addition of Flink 1.19 support"
+  "": "testing addition of Flink 1.20 support"
 }

.github/trigger_files/beam_PostCommit_Python_ValidatesRunner_Flink.json

@@ -1,3 +1,3 @@
 {
-  "https://github.com/apache/beam/pull/32648": "testing addition of Flink 1.19 support"
+  "": "testing addition of Flink 1.20 support"
 }

.github/trigger_files/beam_PostCommit_XVR_Flink.json

@@ -1,3 +1,3 @@
 {
-  "https://github.com/apache/beam/pull/32648": "testing addition of Flink 1.19 support"
+  "": "testing addition of Flink 1.20 support"
 }

.github/workflows/beam_LoadTests_Java_GBK_Smoke.yml

@@ -106,7 +106,7 @@ jobs:
           arguments: |
             --info \
             -PloadTest.mainClass=org.apache.beam.sdk.loadtests.GroupByKeyLoadTest \
-            -Prunner=:runners:flink:1.19 \
+            -Prunner=:runners:flink:1.20 \
             '-PloadTest.args=${{ env.beam_LoadTests_Java_GBK_Smoke_test_arguments_3 }}' \
       - name: run GroupByKey load test Spark
         uses: ./.github/actions/gradle-command-self-hosted-action

.github/workflows/beam_PostCommit_Java_Examples_Flink.yml

@@ -80,7 +80,7 @@ jobs:
       - name: run examplesIntegrationTest script
         uses: ./.github/actions/gradle-command-self-hosted-action
         with:
-          gradle-command: :runners:flink:1.19:examplesIntegrationTest
+          gradle-command: :runners:flink:1.20:examplesIntegrationTest
       - name: Archive JUnit Test Results
         uses: actions/upload-artifact@v4
         if: ${{ !success() }}

.github/workflows/beam_PostCommit_Java_Nexmark_Flink.yml

@@ -102,13 +102,13 @@ jobs:
         with:
           gradle-command: :sdks:java:testing:nexmark:run
           arguments: |
-            -Pnexmark.runner=:runners:flink:1.19 \
+            -Pnexmark.runner=:runners:flink:1.20 \
              "${{ env.GRADLE_COMMAND_ARGUMENTS }} --streaming=${{ matrix.streaming }} --queryLanguage=${{ matrix.queryLanguage }}" \
       - name: run PostCommit Java Nexmark Flink (${{ matrix.streaming }}) script
         if: matrix.queryLanguage == 'none'
         uses: ./.github/actions/gradle-command-self-hosted-action
         with:
           gradle-command: :sdks:java:testing:nexmark:run
           arguments: |
-            -Pnexmark.runner=:runners:flink:1.19 \
+            -Pnexmark.runner=:runners:flink:1.20 \
             "${{ env.GRADLE_COMMAND_ARGUMENTS }}--streaming=${{ matrix.streaming }}"

.github/workflows/beam_PostCommit_Java_PVR_Flink_Streaming.yml

@@ -77,7 +77,7 @@ jobs:
       - name: run PostCommit Java Flink PortableValidatesRunner Streaming script
         uses: ./.github/actions/gradle-command-self-hosted-action
         with:
-            gradle-command: runners:flink:1.19:job-server:validatesPortableRunnerStreaming
+            gradle-command: runners:flink:1.20:job-server:validatesPortableRunnerStreaming
       - name: Archive JUnit Test Results
         uses: actions/upload-artifact@v4
         if: ${{ !success() }}

.github/workflows/beam_PostCommit_Java_Tpcds_Flink.yml

@@ -101,5 +101,5 @@ jobs:
         with:
           gradle-command: :sdks:java:testing:tpcds:run
           arguments: |
-            -Ptpcds.runner=:runners:flink:1.19 \
+            -Ptpcds.runner=:runners:flink:1.20 \
             "-Ptpcds.args=${{env.tpcdsBigQueryArgs}} ${{env.tpcdsInfluxDBArgs}} ${{ env.GRADLE_COMMAND_ARGUMENTS }} --queries=${{env.tpcdsQueriesArg}}" \

.github/workflows/beam_PostCommit_Java_ValidatesRunner_Flink.yml

@@ -78,7 +78,7 @@ jobs:
       - name: run validatesRunner script
         uses: ./.github/actions/gradle-command-self-hosted-action
         with:
-          gradle-command: :runners:flink:1.19:validatesRunner
+          gradle-command: :runners:flink:1.20:validatesRunner
       - name: Archive JUnit Test Results
         uses: actions/upload-artifact@v4
         if: ${{ !success() }}

.github/workflows/beam_PostCommit_Java_ValidatesRunner_Flink_Java8.yml

@@ -80,11 +80,11 @@ jobs:
             11
       - name: run jar Java8 script
         run: |
-          ./gradlew :runners:flink:1.19:jar :runners:flink:1.19:testJar
+          ./gradlew :runners:flink:1.20:jar :runners:flink:1.20:testJar
       - name: run validatesRunner Java8 script
         uses: ./.github/actions/gradle-command-self-hosted-action
         with:
-          gradle-command: :runners:flink:1.19:validatesRunner
+          gradle-command: :runners:flink:1.20:validatesRunner
           arguments: |
             -x shadowJar \
             -x shadowTestJar \

.github/workflows/beam_PostCommit_XVR_Flink.yml

@@ -47,7 +47,7 @@ env:
   DEVELOCITY_ACCESS_KEY: ${{ secrets.GE_ACCESS_TOKEN }}
   GRADLE_ENTERPRISE_CACHE_USERNAME: ${{ secrets.GE_CACHE_USERNAME }}
   GRADLE_ENTERPRISE_CACHE_PASSWORD: ${{ secrets.GE_CACHE_PASSWORD }}
-  FlinkVersion: 1.19
+  FlinkVersion: 1.20
 
 jobs:
   beam_PostCommit_XVR_Flink:

.github/workflows/beam_PreCommit_Java_PVR_Flink_Batch.yml

@@ -94,7 +94,7 @@ jobs:
       - name: run validatesPortableRunnerBatch script
         uses: ./.github/actions/gradle-command-self-hosted-action
         with:
-          gradle-command: :runners:flink:1.19:job-server:validatesPortableRunnerBatch
+          gradle-command: :runners:flink:1.20:job-server:validatesPortableRunnerBatch
         env:
           CLOUDSDK_CONFIG: ${{ env.KUBELET_GCLOUD_CONFIG_PATH }}
       - name: Archive JUnit Test Results

.github/workflows/beam_PreCommit_Java_PVR_Flink_Docker.yml

@@ -99,7 +99,7 @@ jobs:
       - name: run PreCommit Java PVR Flink Docker script
         uses: ./.github/actions/gradle-command-self-hosted-action
         with:
-          gradle-command: :runners:flink:1.19:job-server:validatesPortableRunnerDocker
+          gradle-command: :runners:flink:1.20:job-server:validatesPortableRunnerDocker
         env:
           CLOUDSDK_CONFIG: ${{ env.KUBELET_GCLOUD_CONFIG_PATH}}
       - name: Archive JUnit Test Results

.github/workflows/beam_Publish_Docker_Snapshots.yml

@@ -83,7 +83,7 @@ jobs:
       - name: run Publish Docker Snapshots script for Flink
         uses: ./.github/actions/gradle-command-self-hosted-action
         with:
-          gradle-command: :runners:flink:1.17:job-server-container:dockerPush
+          gradle-command: :runners:flink:1.20:job-server-container:dockerPush
           arguments: |
             -Pdocker-repository-root=gcr.io/apache-beam-testing/beam_portability \
-            -Pdocker-tag-list=latest
+            -Pdocker-tag-list=latest

CHANGES.md

@@ -61,6 +61,7 @@
 * New highly anticipated feature Y added to Java SDK ([#Y](https://github.com/apache/beam/issues/Y)).
 * [Python] Introduce Managed Transforms API ([#31495](https://github.com/apache/beam/pull/31495))
 * Flink 1.19 support added ([#32648](https://github.com/apache/beam/pull/32648))
+* Flink 1.20 support added ([#32863](https://github.com/apache/beam/pull/32863))
 
 ## I/Os
 

contributor-docs/release-guide.md

@@ -887,7 +887,7 @@ write to BigQuery, and create a cluster of machines for running containers (for
   ```
   **Flink Local Runner**
   ```
-  ./gradlew :runners:flink:1.19:runQuickstartJavaFlinkLocal \
+  ./gradlew :runners:flink:1.20:runQuickstartJavaFlinkLocal \
   -Prepourl=https://repository.apache.org/content/repositories/orgapachebeam-${KEY} \
   -Pver=${RELEASE_VERSION}
   ```

gradle.properties

@@ -39,6 +39,6 @@ docker_image_default_repo_root=apache
 docker_image_default_repo_prefix=beam_
 
 # supported flink versions
-flink_versions=1.17,1.18,1.19
+flink_versions=1.17,1.18,1.19,1.20
 # supported python versions
 python_versions=3.9,3.10,3.11,3.12

learning/tour-of-beam/learning-content/introduction/introduction-concepts/runner-concepts/description.md

@@ -191,8 +191,8 @@ $ wordcount --input gs://dataflow-samples/shakespeare/kinglear.txt \
 {{if (eq .Sdk "java")}}
 
 ##### Portable
-1. Starting with Beam 2.18.0, pre-built Flink Job Service Docker images are available at Docker Hub: `Flink 1.17`, `Flink 1.18`, `Flink 1.19`.
-2. Start the JobService endpoint: `docker run --net=host apache/beam_flink1.19_job_server:latest`
+1. Starting with Beam 2.18.0, pre-built Flink Job Service Docker images are available at Docker Hub: `Flink 1.17`, `Flink 1.18`, `Flink `1.19`, `Flink 1.20`.
+2. Start the JobService endpoint: `docker run --net=host apache/beam_flink1.20_job_server:latest`
 3. Submit the pipeline to the above endpoint by using the PortableRunner, job_endpoint set to localhost:8099 (this is the default address of the JobService). Optionally set environment_type set to LOOPBACK. For example:
 
 ```
@@ -233,8 +233,8 @@ mvn exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount \
 {{end}}
 
 {{if (eq .Sdk "python")}}
-1. Starting with Beam 2.18.0, pre-built Flink Job Service Docker images are available at Docker Hub: `Flink 1.17`, `Flink 1.18`, `Flink 1.19`.
-2. Start the JobService endpoint: `docker run --net=host apache/beam_flink1.19_job_server:latest`
+1. Starting with Beam 2.18.0, pre-built Flink Job Service Docker images are available at Docker Hub: `Flink 1.17`, `Flink 1.18`, `Flink 1.19`, `Flink 1.20`.
+2. Start the JobService endpoint: `docker run --net=host apache/beam_flink1.20_job_server:latest`
 3. Submit the pipeline to the above endpoint by using the PortableRunner, job_endpoint set to localhost:8099 (this is the default address of the JobService). Optionally set environment_type set to LOOPBACK. For example:
 
 ```

release/build.gradle.kts

@@ -39,7 +39,7 @@ task("runJavaExamplesValidationTask") {
   dependsOn(":runners:direct-java:runQuickstartJavaDirect")
   dependsOn(":runners:google-cloud-dataflow-java:runQuickstartJavaDataflow")
   dependsOn(":runners:spark:3:runQuickstartJavaSpark")
-  dependsOn(":runners:flink:1.19:runQuickstartJavaFlinkLocal")
+  dependsOn(":runners:flink:1.20:runQuickstartJavaFlinkLocal")
   dependsOn(":runners:direct-java:runMobileGamingJavaDirect")
   dependsOn(":runners:google-cloud-dataflow-java:runMobileGamingJavaDataflow")
   dependsOn(":runners:twister2:runQuickstartJavaTwister2")

runners/flink/1.20/build.gradle

@@ -0,0 +1,25 @@
+/*
+ * 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.
+ */
+
+project.ext {
+  flink_major = '1.20'
+  flink_version = '1.20.0'
+}
+
+// Load the main build script which contains all build logic.
+apply from: "../flink_runner.gradle"

runners/flink/1.20/job-server-container/build.gradle

@@ -0,0 +1,26 @@
+/*
+ * 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"

runners/flink/1.20/job-server/build.gradle

@@ -0,0 +1,31 @@
+/*
+ * 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.20-job-server'
+}
+
+// Load the main build script which contains all build logic.
+apply from: "$basePath/flink_job_server.gradle"

runners/flink/flink_runner.gradle

@@ -191,6 +191,10 @@ dependencies {
   implementation "org.apache.flink:flink-metrics-core:$flink_version"
   implementation "org.apache.flink:flink-java:$flink_version"
 
+  if (flink_version.compareTo("1.20") >= 0) {
+    implementation "org.apache.flink:flink-core-api:$flink_version"
+  }
+
   implementation "org.apache.flink:flink-runtime:$flink_version"
   implementation "org.apache.flink:flink-metrics-core:$flink_version"
   testImplementation "org.apache.flink:flink-runtime:$flink_version:tests"

sdks/go/examples/stringsplit/stringsplit.go

@@ -21,7 +21,7 @@
 // 1. From a command line, navigate to the top-level beam/ directory and run
 // the Flink job server:
 //
-//	./gradlew :runners:flink:1.19:job-server:runShadow -Djob-host=localhost -Dflink-master=local
+//	./gradlew :runners:flink:1.20:job-server:runShadow -Djob-host=localhost -Dflink-master=local
 //
 // 2. The job server is ready to receive jobs once it outputs a log like the
 // following: `JobService started on localhost:8099`. Take note of the endpoint

sdks/go/examples/wasm/README.md

@@ -68,13 +68,13 @@ cd $BEAM_HOME
 Expected output should include the following, from which you acquire the latest flink runner version.
 
 ```shell
-'flink_versions: 1.17,1.18,1.19'
+'flink_versions: 1.17,1.18,1.19,1.20'
 ```
 
-#### 2. Set to the latest flink runner version i.e. 1.16
+#### 2. Set to the latest flink runner version i.e. 1.20
 
 ```shell
-FLINK_VERSION=1.16
+FLINK_VERSION=1.20
 ```
 
 #### 3. In a separate terminal, start the flink runner (It should take a few minutes on the first execution)

sdks/python/apache_beam/options/pipeline_options.py

@@ -1679,7 +1679,7 @@ def _add_argparse_args(cls, parser):
 class FlinkRunnerOptions(PipelineOptions):
 
   # These should stay in sync with gradle.properties.
-  PUBLISHED_FLINK_VERSIONS = ['1.17', '1.18', '1.19']
+  PUBLISHED_FLINK_VERSIONS = ['1.17', '1.18', '1.19', '1.20']
 
   @classmethod
   def _add_argparse_args(cls, parser):

sdks/typescript/src/apache_beam/runners/flink.ts

@@ -28,7 +28,7 @@ import { JavaJarService } from "../utils/service";
 const MAGIC_HOST_NAMES = ["[local]", "[auto]"];
 
 // These should stay in sync with gradle.properties.
-const PUBLISHED_FLINK_VERSIONS = ["1.17", "1.18", "1.19"];
+const PUBLISHED_FLINK_VERSIONS = ["1.17", "1.18", "1.19", "1.20"];
 
 const defaultOptions = {
   flinkMaster: "[local]",

settings.gradle.kts

@@ -137,6 +137,10 @@ include(":runners:flink:1.18:job-server-container")
 include(":runners:flink:1.19")
 include(":runners:flink:1.19:job-server")
 include(":runners:flink:1.19:job-server-container")
+// Flink 1.20
+include(":runners:flink:1.20")
+include(":runners:flink:1.20:job-server")
+include(":runners:flink:1.20:job-server-container")
 /* End Flink Runner related settings */
 include(":runners:twister2")
 include(":runners:google-cloud-dataflow-java")