Go VR Flink test on Flink 2.0

Author: Abacn

.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"
 }

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/"

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(' ')}")

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