Go VR Flink test on Flink 2.0

Abacn · Abacn · commit df0450622e93 · 2026-04-24T12:27:10.000-04:00
diff --git a/.github/trigger_files/beam_PostCommit_Go_VR_Flink.json b/.github/trigger_files/beam_PostCommit_Go_VR_Flink.json
@@ -2,5 +2,4 @@
   "comment": "Modify this file in a trivial way to cause this test suite to run",
   "modification": 2,
   "https://github.com/apache/beam/pull/32440": "testing datastream optimizations",
-  "https://github.com/apache/beam/pull/32648": "testing addition of Flink 1.19 support"
 }
diff --git a/runners/flink/2.0/src/main/java/org/apache/beam/runners/flink/FlinkExecutionEnvironments.java b/runners/flink/2.0/src/main/java/org/apache/beam/runners/flink/FlinkExecutionEnvironments.java
@@ -271,6 +271,13 @@ public static StreamExecutionEnvironment createStreamExecutionEnvironment(
     }
     configureWebUIOptions(flinkStreamEnv.getConfig(), options.as(PipelineOptions.class));
 
+    // Register Kryo serializer for ImmutableBiMap to fix serialization issues in Flink 2
+    ((org.apache.flink.api.common.serialization.SerializerConfigImpl)
+            flinkStreamEnv.getConfig().getSerializerConfig())
+        .registerTypeWithKryoSerializer(
+            org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap.class,
+            org.apache.beam.runners.flink.translation.types.ImmutableMapSerializer.class);
+
     return flinkStreamEnv;
   }
 
diff --git a/sdks/go/container/build.gradle b/sdks/go/container/build.gradle
@@ -33,8 +33,10 @@ docker {
   name containerImageName(
           name: project.docker_image_default_repo_prefix + "go_sdk",
           root: project.rootProject.hasProperty(["docker-repository-root"]) ?
-                  project.rootProject["docker-repository-root"] :
-                  project.docker_image_default_repo_root)
+              project.rootProject["docker-repository-root"] :
+              project.docker_image_default_repo_root,
+          tag: project.rootProject.hasProperty(["docker-tag"]) ?
+              project.rootProject["docker-tag"] : project.sdk_version)
   // tags used by dockerTag task
   tags containerImageTags()
   files "./build/"
diff --git a/sdks/go/run_with_go_version.sh b/sdks/go/run_with_go_version.sh
@@ -83,12 +83,19 @@ if [ -z "$GOCMD" ] ; then
     # The download command isn't concurrency safe so we get an exclusive lock, without wait.
     # If we're first, we ensure the command is downloaded, releasing the lock afterwards.
     # This operation is cached on system and won't be re-downloaded at least.
-    flock --exclusive --nonblock --conflict-exit-code 0 $LOCKFILE $GOBIN/$GOVERS download
+    if command -v flock &> /dev/null; then
+        flock --exclusive --nonblock --conflict-exit-code 0 $LOCKFILE $GOBIN/$GOVERS download
 
-    # Execute the script with the remaining arguments.
-    # We get a shared lock for the ordinary go command execution.
-    echo $GOBIN/$GOVERS $@
-    flock --shared --timeout=10 $LOCKFILE $GOBIN/$GOVERS $@
+        # Execute the script with the remaining arguments.
+        # We get a shared lock for the ordinary go command execution.
+        echo $GOBIN/$GOVERS $@
+        flock --shared --timeout=10 $LOCKFILE $GOBIN/$GOVERS $@
+    else
+        echo "WARNING: flock not found, running without lock."
+        $GOBIN/$GOVERS download
+        echo $GOBIN/$GOVERS $@
+        $GOBIN/$GOVERS $@
+    fi
 else
     # Minor TODO: Figure out if we can pull out the GOCMD env variable after goPrepare
     # completion, and avoid this brittle GOBIN substitution.
diff --git a/sdks/go/test/build.gradle b/sdks/go/test/build.gradle
@@ -79,8 +79,7 @@ task dataflowValidatesRunnerARM64() {
 task flinkValidatesRunner {
   group = "Verification"
 
-  // TODO(https://github.com/apache/beam/issues/37600) use project.ext.latestFlinkVersion after resolved
-  def flinkVersion = '1.20'
+  def flinkVersion = project.ext.latestFlinkVersion
 
   dependsOn ":sdks:go:test:goBuild"
   dependsOn ":sdks:go:container:docker"
@@ -105,6 +104,7 @@ task flinkValidatesRunner {
         "--runner flink",
         "--flink_job_server_jar ${project(":runners:flink:${flinkVersion}:job-server").shadowJar.archivePath}",
         "--pipeline_opts \"${pipelineOptions.join(' ')}\"",
+        "--prebuild_go_docker_tag ${project(":sdks:go:container").containerImageTags().first()}",
     ]
     exec {
       if (fork_java_home) {
@@ -133,6 +133,7 @@ task samzaValidatesRunner {
         "--runner samza",
         "--samza_job_server_jar ${project(":runners:samza:job-server").shadowJar.archivePath}",
         "--pipeline_opts \"${pipelineOptions.join(' ')}\"",
+        "--prebuild_go_docker_tag ${project(":sdks:go:container").containerImageTags().first()}",
     ]
     exec {
       if (fork_java_home) {
@@ -150,6 +151,7 @@ task sparkValidatesRunner {
   group = "Verification"
 
   dependsOn ":sdks:go:test:goBuild"
+  dependsOn ":sdks:go:container:docker"
   dependsOn ":sdks:java:container:${project.ext.currentJavaVersion}:docker"
   dependsOn ":runners:spark:3:job-server:shadowJar"
   dependsOn ":sdks:java:testing:expansion-service:buildTestExpansionServiceJar"
@@ -161,6 +163,7 @@ task sparkValidatesRunner {
         "--runner spark",
         "--spark_job_server_jar ${project(":runners:spark:3:job-server").shadowJar.archivePath}",
         "--pipeline_opts \"${pipelineOptions.join(' ')}\"",
+        "--prebuild_go_docker_tag ${project(":sdks:go:container").containerImageTags().first()}",
     ]
     exec {
       if (fork_java_home) {
@@ -195,6 +198,7 @@ tasks.register("ulrValidatesRunner") {
     def options = [
         "--runner portable",
         "--pipeline_opts \"${pipelineOptions.join(' ')}\"",
+        "--prebuild_go_docker_tag ${project(":sdks:go:container").containerImageTags().first()}",
     ]
     exec {
       executable "sh"
@@ -227,6 +231,7 @@ task prismValidatesRunner {
     def options = [
         "--runner prism",
         "--pipeline_opts \"${pipelineOptions.join(' ')}\"",
+        "--prebuild_go_docker_tag ${project(":sdks:go:container").containerImageTags().first()}",
     ]
     exec {
       if (fork_java_home) {
@@ -265,11 +270,12 @@ ext.goIoValidatesRunnerTask = { proj, name, scriptOpts, pipelineOpts ->
           "--expansion_jar=debeziumio:${debeziumIoExpJar}",
           "--expansion_jar=gcpio:${gcpIoExpJar}",
           "--bq_dataset=apache-beam-testing.beam_bigquery_io_test_temp",
-          "--bt_instance=projects/apache-beam-testing/instances/beam-test"
+          "--bt_instance=projects/apache-beam-testing/instances/beam-test",
       ]
       pipelineOptions.addAll(pipelineOpts)
       def options = [
           "--pipeline_opts \"${pipelineOptions.join(' ')}\"",
+          "--prebuild_go_docker_tag ${project(":sdks:go:container").containerImageTags().first()}",
       ]
       options.addAll(scriptOpts)
       logger.info("Running the command: sh -c ./run_validatesrunner_tests.sh ${options.join(' ')}")
diff --git a/sdks/go/test/run_validatesrunner_tests.sh b/sdks/go/test/run_validatesrunner_tests.sh
@@ -32,6 +32,7 @@
 #    --tests -> A space-seperated list of targets for "go test", written with
 #        beam/sdks/go as the working directory. Defaults to all packages in the
 #        integration and regression directories.
+#    --run -> To select which tests to run
 #    --timeout -> Timeout for the go test command, on a per-package level.
 #    --simultaneous -> Number of simultaneous packages to test.
 #        Controls the -p flag for the go test command.
@@ -129,6 +130,11 @@ case $key in
         shift # past argument
         shift # past value
         ;;
+    --run)
+        TEST_RUN="$2"
+        shift # past argument
+        shift # past value
+        ;;
     --timeout)
         TIMEOUT="$2"
         shift # past argument
@@ -229,8 +235,8 @@ case $key in
         shift # past argument
         shift # past value
         ;;
-    --java11_home)
-        JAVA11_HOME="$2"
+    --prebuild_go_docker_tag)
+        PREBUILD_GO_DOCKER_TAG="$2"
         shift # past argument
         shift # past value
         ;;
@@ -427,8 +433,12 @@ if [[ "$RUNNER" == "dataflow" ]]; then
     fi
   fi
 else
-  TAG=dev
-  ./gradlew :sdks:go:container:docker -Pdocker-tag=$TAG
+  if [[ -n {PREBUILD_GO_DOCKER_TAG} ]]; then
+    TAG=$PREBUILD_GO_DOCKER_TAG
+  else
+    TAG=dev
+    ./gradlew :sdks:go:container:docker -Pdocker-tag=$TAG
+  fi
   CONTAINER=apache/beam_go_sdk
 fi
 
@@ -447,6 +457,9 @@ ARGS="$ARGS --environment_config=$CONTAINER:$TAG"
 ARGS="$ARGS --staging_location=$GCS_LOCATION/staging-validatesrunner-test/$GCS_SUBFOLDER"
 ARGS="$ARGS --temp_location=$GCS_LOCATION/temp-validatesrunner-test/$GCS_SUBFOLDER"
 ARGS="$ARGS --endpoint=$ENDPOINT"
+if [[ -n "$TEST_RUN" ]]; then
+  ARGS="$ARGS -run $TEST_RUN"
+fi
 if [[ -n "$TEST_EXPANSION_ADDR" ]]; then
   ARGS="$ARGS --test_expansion_addr=$TEST_EXPANSION_ADDR"
 fi

Original file line number	Diff line number	Diff line change
`@@ -2,5 +2,4 @@`
`2`	`2`	`"comment": "Modify this file in a trivial way to cause this test suite to run",`
`3`	`3`	`"modification": 2,`
`4`	`4`	`"https://github.com/apache/beam/pull/32440": "testing datastream optimizations",`
`5`		`- "https://github.com/apache/beam/pull/32648": "testing addition of Flink 1.19 support"`
`6`	`5`	`}`
Original file line number	Diff line number	Diff line change
`@@ -271,6 +271,13 @@ public static StreamExecutionEnvironment createStreamExecutionEnvironment(`
`271`	`271`	`}`
`272`	`272`	`configureWebUIOptions(flinkStreamEnv.getConfig(), options.as(PipelineOptions.class));`
`273`	`273`
	`274`	`+ // Register Kryo serializer for ImmutableBiMap to fix serialization issues in Flink 2`
	`275`	`+ ((org.apache.flink.api.common.serialization.SerializerConfigImpl)`
	`276`	`+ flinkStreamEnv.getConfig().getSerializerConfig())`
	`277`	`+ .registerTypeWithKryoSerializer(`
	`278`	`+ org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap.class,`
	`279`	`+ org.apache.beam.runners.flink.translation.types.ImmutableMapSerializer.class);`
	`280`	`+`
`274`	`281`	`return flinkStreamEnv;`
`275`	`282`	`}`
`276`	`283`