Exercise Flink 2.0 Python Validates Runner tests

Author: Abacn

.github/trigger_files/beam_PostCommit_PortableJar_Flink.json

@@ -0,0 +1 @@
+{}

.github/trigger_files/beam_PostCommit_Python_Examples_Flink.json

@@ -0,0 +1 @@
+{}

.github/workflows/beam_PostCommit_Python_Portable_Flink.yml

@@ -64,7 +64,8 @@ jobs:
         job_phrase: ["Run Python Portable Flink"]
         # TODO: Enable PROCESS https://github.com/apache/beam/issues/35702
         # environment_type: ['DOCKER', 'LOOPBACK', 'PROCESS']
-        environment_type: ['DOCKER', 'LOOPBACK']
+        # mode not covered by PreCommit_Python_PVR_Flink (i.e. 'LOOPBACK')
+        environment_type: ['DOCKER', 'PROCESS']
     steps:
       - uses: actions/checkout@v4
       - name: Setup repository
@@ -83,7 +84,7 @@ jobs:
           CLOUDSDK_CONFIG: ${{ env.KUBELET_GCLOUD_CONFIG_PATH}}
         uses: ./.github/actions/gradle-command-self-hosted-action
         with:
-          gradle-command: :sdks:python:test-suites:portable:py310:flinkCompatibilityMatrix${{ matrix.environment_type }}
+          gradle-command: :sdks:python:test-suites:portable:py310:flink2CompatibilityMatrix${{ matrix.environment_type }}
           arguments: |
             -PpythonVersion=3.10 \
       - name: Archive Python Test Results

.github/workflows/beam_PreCommit_Python_PVR_Flink.yml

@@ -106,7 +106,8 @@ jobs:
         env:
           CLOUDSDK_CONFIG: ${{ env.KUBELET_GCLOUD_CONFIG_PATH}}
         with:
-          gradle-command: :sdks:python:test-suites:portable:py313:flinkValidatesRunner
+          # Run Flink 2 tests. Flink 1.20 is covered by PostCommit_Python_ValidatesRunner_Flink
+          gradle-command: :sdks:python:test-suites:portable:py313:flink2ValidatesRunner
           arguments: |
             -PpythonVersion=3.13 \
       - name: Archive Python Test Results

sdks/python/test-suites/portable/common.gradle

@@ -22,25 +22,43 @@ import org.apache.tools.ant.taskdefs.condition.Os
 
 def pythonRootDir = "${rootDir}/sdks/python"
 def pythonVersionSuffix = project.ext.pythonVersion.replace('.', '')
+// TODO(https://github.com/apache/beam/issues/36947): Remove when dropping Flink 1.x support
 def latestFlinkVersion = project.ext.latestFlinkVersion
+def latestFlink2Version = '2.0'
 def currentJavaVersion = project.ext.currentJavaVersion
 
 ext {
   pythonContainerTask = ":sdks:python:container:py${pythonVersionSuffix}:docker"
 }
 
-def createFlinkRunnerTestTask(String workerType) {
-  def taskName = "flinkCompatibilityMatrix${workerType}"
-  // project(":runners:flink:${latestFlinkVersion}:job-server").shadowJar.archivePath is not resolvable until runtime, so hard-code it here.
-  def jobServerJar = "${rootDir}/runners/flink/${latestFlinkVersion}/job-server/build/libs/beam-runners-flink-${latestFlinkVersion}-job-server-${version}.jar"
+def createFlink2ConfigDir = tasks.register("createFlink2ConfigDir", Copy) {
+  from file("${project.rootDir}/runners/flink/2.0/src/test/resources/flink-test-config.yaml")
+  into file("${project.buildDir}/flinkConf")
+  rename { String fileName -> 'config.yaml' }
+}
+
+def createFlinkRunnerTestTask(String workerType, String flinkVersion) {
+  String taskName
+
+  // project(":runners:flink:${flinkVersion}:job-server").shadowJar.archivePath is not resolvable until runtime, so hard-code it here.
+  def jobServerJar = "${rootDir}/runners/flink/${flinkVersion}/job-server/build/libs/beam-runners-flink-${flinkVersion}-job-server-${version}.jar"
   def options = "--flink_job_server_jar=${jobServerJar} --environment_type=${workerType}"
+  if flinkVersion.startsWith('2') {
+    taskName = "flink2CompatibilityMatrix${workerType}"
+    options += " --flink_conf_dir=${buildDir.absolutePath}/flinkConf"
+  } else {
+    taskName = "flinkCompatibilityMatrix${workerType}"
+  }
   if (workerType == 'PROCESS') {
     options += " --environment_options=process_command=${buildDir.absolutePath}/sdk_worker.sh"
   }
   def task = toxTask(taskName, 'flink-runner-test', options)
   // Through the Flink job server, we transitively add dependencies on the expansion services needed in tests.
   task.configure {
-    dependsOn ":runners:flink:${latestFlinkVersion}:job-server:shadowJar"
+    if flinkVersion.startsWith('2') {
+      dependsOn createFlink2ConfigDir
+    }
+    dependsOn ":runners:flink:${flinkVersion}:job-server:shadowJar"
     // The Java SDK worker is required to execute external transforms.
     def suffix = getSupportedJavaVersion()
     dependsOn ":sdks:java:container:${suffix}:docker"
@@ -53,31 +71,19 @@ def createFlinkRunnerTestTask(String workerType) {
   return task
 }
 
-createFlinkRunnerTestTask('DOCKER')
-createFlinkRunnerTestTask('PROCESS')
-createFlinkRunnerTestTask('LOOPBACK')
+createFlinkRunnerTestTask('DOCKER', latestFlinkVersion)
+createFlinkRunnerTestTask('PROCESS', latestFlinkVersion)
+createFlinkRunnerTestTask('LOOPBACK', latestFlinkVersion)
+createFlinkRunnerTestTask('DOCKER', latestFlink2Version)
+createFlinkRunnerTestTask('PROCESS', latestFlink2Version)
+createFlinkRunnerTestTask('LOOPBACK', latestFlink2Version)
 
 task flinkValidatesRunner() {
   dependsOn 'flinkCompatibilityMatrixLOOPBACK'
 }
 
-// TODO(https://github.com/apache/beam/issues/19962): Enable on pre-commit.
-tasks.register("flinkTriggerTranscript") {
-  dependsOn 'setupVirtualenv'
-  dependsOn ":runners:flink:${latestFlinkVersion}:job-server:shadowJar"
-  doLast {
-    exec {
-      executable 'sh'
-      args '-c', """
-          . ${envdir}/bin/activate \\
-          && cd ${pythonRootDir} \\
-          && pip install -e .[test] \\
-          && pytest \\
-              apache_beam/transforms/trigger_test.py::WeakTestStreamTranscriptTest \\
-              --test-pipeline-options='--runner=FlinkRunner --environment_type=LOOPBACK --flink_job_server_jar=${project(":runners:flink:${latestFlinkVersion}:job-server:").shadowJar.archivePath}'
-          """
-    }
-  }
+task flink2ValidatesRunner() {
+  dependsOn 'flink2CompatibilityMatrixLOOPBACK'
 }
 
 // Verifies BEAM-10702.