Add Flink 2.2.1 runner support

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

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

.github/workflows/beam_PostCommit_Java_PVR_Flink_Streaming.yml

@@ -56,11 +56,11 @@ jobs:
     runs-on: [self-hosted, ubuntu-24.04, main]
     timeout-minutes: 120
     strategy:
-      matrix: 
+      matrix:
         job_name: [beam_PostCommit_Java_PVR_Flink_Streaming]
         job_phrase: [Run Java Flink PortableValidatesRunner Streaming]
         # every major version
-        flink_version: [ '1.20', '2.0' ]
+        flink_version: [ '1.20', '2.0', '2.2']
     if: |
       github.event_name == 'workflow_dispatch' ||
       github.event_name == 'pull_request_target' ||

.github/workflows/beam_PostCommit_Java_ValidatesRunner_Flink.yml

@@ -49,7 +49,7 @@ env:
   GRADLE_ENTERPRISE_CACHE_PASSWORD: ${{ secrets.GE_CACHE_PASSWORD }}
 
 jobs:
-  beam_PostCommit_Java_ValidatesRunner_Flink:    
+  beam_PostCommit_Java_ValidatesRunner_Flink:
     name: ${{ matrix.job_name }} (${{ matrix.flink_version }})
     runs-on: [self-hosted, ubuntu-24.04, main]
     timeout-minutes: 100
@@ -58,7 +58,7 @@ jobs:
         job_name: [beam_PostCommit_Java_ValidatesRunner_Flink]
         job_phrase: [Run Flink ValidatesRunner]
         # every major version
-        flink_version: ['1.20', '2.0']
+        flink_version: ['1.20', '2.0', '2.2']
     if: |
       github.event_name == 'workflow_dispatch' ||
       github.event_name == 'pull_request_target' ||

.github/workflows/beam_Publish_Docker_Snapshots.yml

@@ -84,10 +84,10 @@ jobs:
             -Pdocker-repository-root=gcr.io/apache-beam-testing/beam_portability \
             -Pdocker-tag-list=${{ github.sha }}${LATEST_TAG} \
             -Pdocker-pull-licenses
-      - name: run Publish Docker Snapshots script for Flink 2.0
+      - name: run Publish Docker Snapshots script for Flink 2.2
         uses: ./.github/actions/gradle-command-self-hosted-action
         with:
-          gradle-command: :runners:flink:2.0:job-server-container:dockerPush
+          gradle-command: :runners:flink:2.2:job-server-container:dockerPush
           arguments: |
             -Pdocker-repository-root=gcr.io/apache-beam-testing/beam_portability \
             -Pdocker-tag-list=${{ github.sha }}${LATEST_TAG} \

.github/workflows/run_rc_validation_java_quickstart.yml

@@ -88,7 +88,7 @@ jobs:
       - name: Run QuickStart Java Flink Runner
         uses: ./.github/actions/gradle-command-self-hosted-action
         with:
-          gradle-command: :runners:flink:2.0:runQuickstartJavaFlinkLocal
+          gradle-command: :runners:flink:2.2:runQuickstartJavaFlinkLocal
           arguments: |
             -Prepourl=${{ env.APACHE_REPO_URL }} \
             -Pver=${{ env.RELEASE_VERSION }}

build.gradle.kts

@@ -1055,3 +1055,17 @@ if (project.hasProperty("testJavaVersion")) {
     }
   }
 }
+
+// Flink 2.2.1 (FLINK-39139) replaced org.lz4:lz4-java with the fork at.yawk.lz4:lz4-java,
+// which declares the same 'org.lz4:lz4-java' Gradle capability. Any subproject that depends
+// on both Flink 2.2+ and another library using org.lz4 (Kafka, Spark, etc.) hits a capability
+// conflict. This resolution applies globally so that individual subprojects do not each need
+// to declare it independently.
+allprojects {
+  configurations.all {
+    resolutionStrategy.capabilitiesResolution.withCapability("org.lz4:lz4-java") {
+      val yawk = candidates.find { it.id.displayName.startsWith("at.yawk.lz4:") }
+      if (yawk != null) select(yawk) else selectHighestVersion()
+    }
+  }
+}

gradle.properties

@@ -39,7 +39,7 @@ docker_image_default_repo_root=apache
 docker_image_default_repo_prefix=beam_
 
 # supported flink versions
-flink_versions=1.17,1.18,1.19,1.20,2.0
+flink_versions=1.17,1.18,1.19,1.20,2.0,2.2
 # supported spark versions
 spark_versions=3,4
 # supported python versions

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:2.0:runQuickstartJavaFlinkLocal")
+  dependsOn(":runners:flink:2.2:runQuickstartJavaFlinkLocal")
   dependsOn(":runners:direct-java:runMobileGamingJavaDirect")
   if (project.hasProperty("ver") || !project.version.toString().endsWith("SNAPSHOT")) {
     // only run one variant of MobileGaming on Dataflow for nightly

runners/flink/2.2/build.gradle

@@ -0,0 +1,58 @@
+/*
+ * 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.
+ */
+
+// Flink 2.2 API changes addressed in this module's source overrides:
+// - CheckpointingMode moved (Flink 2.2): org.apache.flink.streaming.api.CheckpointingMode
+//   is now @Deprecated; canonical path is org.apache.flink.core.execution.CheckpointingMode.
+//   FlinkExecutionEnvironments.java and FlinkPipelineOptionsTest.java are overridden here
+//   to use the non-deprecated import.
+//
+// Flink 2.2 breaking changes that are SQL/cluster-layer only (no runner code change needed):
+// - TLS cipher suite default changed (FLINK-39022): JDK 11.0.30+/17.0.18+/21.0.10+/24+
+//   disabled TLS_RSA_* ciphers; new default uses ECDHE suites. Verify cipher support
+//   on TLS-enabled clusters before upgrading.
+// - StreamingMultiJoinOperator state format change (FLINK-38209): savepoints from Flink 2.1
+//   with table.optimizer.multi-join.enabled=true are NOT compatible with 2.2.
+// - SQL row NOT NULL now enforced (FLINK-38181): use table.legacy-nested-row-nullability
+//   to restore prior (silent-ignore) behavior if needed.
+
+project.ext {
+  flink_major = '2.2'
+  flink_version = '2.2.1'
+  excluded_files = [
+    'main': [
+        // Used by DataSet API only
+        "org/apache/beam/runners/flink/adapter/BeamFlinkDataSetAdapter.java",
+        "org/apache/beam/runners/flink/FlinkBatchPipelineTranslator.java",
+        "org/apache/beam/runners/flink/FlinkBatchPortablePipelineTranslator.java",
+        "org/apache/beam/runners/flink/FlinkBatchTransformTranslators.java",
+        "org/apache/beam/runners/flink/translation/functions/FlinkNonMergingReduceFunction.java",
+        // Moved to org.apache.flink.runtime.state.StateBackendFactory
+        "org/apache/beam/runners/flink/FlinkStateBackendFactory.java",
+    ],
+    'test': [
+        // Used by DataSet API only
+        "org/apache/beam/runners/flink/adapter/BeamFlinkDataSetAdapterTest.java",
+        "org/apache/beam/runners/flink/batch/NonMergingGroupByKeyTest.java",
+        "org/apache/beam/runners/flink/batch/ReshuffleTest.java",
+    ]
+  ]
+}
+
+// Load the main build script which contains all build logic.
+apply from: "../flink_runner.gradle"

runners/flink/2.2/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/2.2/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-2.2-job-server'
+}
+
+// Load the main build script which contains all build logic.
+apply from: "$basePath/flink_job_server.gradle"