From c8afb2d494f01ea1f89c993d00ce2f122cd25d56 Mon Sep 17 00:00:00 2001 From: Danny McCormick Date: Thu, 6 Aug 2026 14:31:53 +0000 Subject: [PATCH 1/7] Fix mobile gaming release validation background process cleanup for Java 21 compatibility In Java 20+, Thread.stop() unconditionally throws UnsupportedOperationException. This caused mobile gaming validation scripts to fail when stopping background injector and leaderboard threads, leaving the child processes running as orphans and causing the nightly snapshot validation workflow to hang and time out at 6 hours. This change introduces Process-based background execution and cleanup in TestScripts to properly terminate background processes across Java versions. --- release/src/main/groovy/TestScripts.groovy | 94 +++++++++++++++++++ .../groovy/mobilegaming-java-dataflow.groovy | 20 ++-- .../groovy/mobilegaming-java-direct.groovy | 12 +-- 3 files changed, 105 insertions(+), 21 deletions(-) diff --git a/release/src/main/groovy/TestScripts.groovy b/release/src/main/groovy/TestScripts.groovy index dc2438007ac1..c9d5ab989fe6 100644 --- a/release/src/main/groovy/TestScripts.groovy +++ b/release/src/main/groovy/TestScripts.groovy @@ -37,6 +37,7 @@ class TestScripts { static String bqDataset static String pubsubTopic static String mavenLocalPath + static List backgroundProcesses = Collections.synchronizedList(new ArrayList()) } def TestScripts(String[] args) { @@ -79,6 +80,10 @@ class TestScripts { var.mavenLocalPath = options.mavenLocalPath println "Maven local path: ${var.mavenLocalPath}" } + + Runtime.getRuntime().addShutdownHook(new Thread({ + stopAllBackgroundProcesses() + })) } def ver() { @@ -135,6 +140,37 @@ class TestScripts { } } + // Run a command in the background, returning the Process object. + public Process runBackground(String cmd) { + println cmd + if (cmd.startsWith("mvn ")) { + return _mvnBackground(cmd.substring(4)) + } else { + return _executeBackground(cmd) + } + } + + // Stop/kill a background process and all its descendants. + public void stopProcess(Process proc) { + if (proc != null && proc.isAlive()) { + try { + proc.descendants().forEach { it.destroyForcibly() } + } catch (Throwable ignored) { + } + proc.destroyForcibly() + proc.waitFor(10, java.util.concurrent.TimeUnit.SECONDS) + } + var.backgroundProcesses.remove(proc) + } + + // Stop all active background processes. + public void stopAllBackgroundProcesses() { + def procs = new ArrayList<>(var.backgroundProcesses) + procs.each { proc -> + stopProcess(proc) + } + } + // Check for expected results in actual stdout from previous command, if fails, log errors then exit. public void see(String expected, String actual) { if (!actual.contains(expected)) { @@ -159,6 +195,7 @@ class TestScripts { // Cleanup and print success public void done() { + stopAllBackgroundProcesses() var.startDir.deleteDir() println "[SUCCESS]" System.exit(0) @@ -187,6 +224,26 @@ class TestScripts { return output_text } + // Run a single command asynchronously in the background + private Process _executeBackground(String cmd) { + def shell = "sh -c cmd".split(' ') + shell[2] = cmd + def pb = new ProcessBuilder(shell) + pb.directory(var.curDir) + pb.redirectErrorStream(true) + def proc = pb.start() + var.backgroundProcesses.add(proc) + Thread.startDaemon { + try { + proc.inputStream.eachLine { + println it + } + } catch (Throwable ignored) { + } + } + return proc + } + // Change directory private void _chdir(String subdir) { var.curDir = new File(var.curDir.absolutePath, subdir) @@ -233,8 +290,45 @@ class TestScripts { return _execute(setPath + cmd) } + // Run a maven command in the background + private Process _mvnBackground(String args) { + String mvnlocalPath = var.mavenLocalPath + if (!(var.mavenLocalPath)) { + mvnlocalPath = var.startDir + } + def m2 = new File(mvnlocalPath, ".m2/repository") + m2.mkdirs() + def settings = new File(mvnlocalPath, "settings.xml") + if(!settings.exists()) { + settings.write """ + + ${m2.absolutePath} + + + testrel + + + test.release + ${var.repoUrl} + + + + + + """ + } + def cmd = "mvn ${args} -s ${settings.absolutePath} -Ptestrel -B" + String path = System.getenv("PATH"); + String maven_home = System.getenv("MAVEN_HOME") ?: '/usr/local/maven' + println "Using maven ${maven_home}" + def mvnPath = "${maven_home}/bin" + def setPath = "export PATH=\"${mvnPath}:${path}\" && " + return _executeBackground(setPath + cmd) + } + // Clean up and report error public void error(String text) { + stopAllBackgroundProcesses() var.startDir.deleteDir() println "[ERROR] $text" System.exit(1) diff --git a/release/src/main/groovy/mobilegaming-java-dataflow.groovy b/release/src/main/groovy/mobilegaming-java-dataflow.groovy index 51ea528a7638..96cd557562ce 100644 --- a/release/src/main/groovy/mobilegaming-java-dataflow.groovy +++ b/release/src/main/groovy/mobilegaming-java-dataflow.groovy @@ -138,19 +138,13 @@ class LeaderBoardRunner { } println "Tables ${userTable} and ${teamTable} created successfully." - def InjectorThread = Thread.start() { - t.run(mobileGamingCommands.createInjectorCommand()) - } + def injectorProcess = t.runBackground(mobileGamingCommands.createInjectorCommand()) String jobName = "leaderboard-validation-" + new Date().getTime() + "-" + new Random().nextInt(1000) - def LeaderBoardThread = Thread.start() { - if (useStreamingEngine) { - t.run(mobileGamingCommands.createPipelineCommand( - "LeaderBoardWithStreamingEngine", runner, jobName, "LeaderBoard")) - } else { - t.run(mobileGamingCommands.createPipelineCommand("LeaderBoard", runner, jobName)) - } - } + def leaderBoardProcess = useStreamingEngine ? + t.runBackground(mobileGamingCommands.createPipelineCommand( + "LeaderBoardWithStreamingEngine", runner, jobName, "LeaderBoard")) : + t.runBackground(mobileGamingCommands.createPipelineCommand("LeaderBoard", runner, jobName)) t.run("gcloud dataflow jobs list | grep pyflow-wordstream-candidate | grep Running | cut -d' ' -f1") @@ -175,8 +169,8 @@ class LeaderBoardRunner { println "Waiting for pipeline to produce more results..." sleep(60000) // wait for 1 min } - InjectorThread.stop() - LeaderBoardThread.stop() + t.stopProcess(injectorProcess) + t.stopProcess(leaderBoardProcess) t.run("""RUNNING_JOB=`gcloud dataflow jobs list | grep ${jobName} | grep Running | cut -d' ' -f1` if [ ! -z "\${RUNNING_JOB}" ] then diff --git a/release/src/main/groovy/mobilegaming-java-direct.groovy b/release/src/main/groovy/mobilegaming-java-direct.groovy index 34eab4c00768..3985c7e59634 100644 --- a/release/src/main/groovy/mobilegaming-java-direct.groovy +++ b/release/src/main/groovy/mobilegaming-java-direct.groovy @@ -98,14 +98,10 @@ while (!tables.contains(userTable) || !tables.contains(teamTable)) { } println "Tables ${userTable} and ${teamTable} created successfully." -def InjectorThread = Thread.start() { - t.run(mobileGamingCommands.createInjectorCommand()) -} +def injectorProcess = t.runBackground(mobileGamingCommands.createInjectorCommand()) jobName = "leaderboard-validation-" + new Date().getTime() + "-" + new Random().nextInt(1000) -def LeaderBoardThread = Thread.start() { - t.run(mobileGamingCommands.createPipelineCommand("LeaderBoard", runner, jobName)) -} +def leaderBoardProcess = t.runBackground(mobileGamingCommands.createPipelineCommand("LeaderBoard", runner, jobName)) // verify outputs in BQ tables def startTime = System.currentTimeMillis() @@ -128,8 +124,8 @@ while ((System.currentTimeMillis() - startTime)/60000 < mobileGamingCommands.EXE println "Waiting for pipeline to produce more results..." sleep(60000) // wait for 1 min } -InjectorThread.stop() -LeaderBoardThread.stop() +t.stopProcess(injectorProcess) +t.stopProcess(leaderBoardProcess) if(!isSuccess){ t.error("FAILED: Failed running LeaderBoard on DirectRunner") From 5d8fd71844797a469227d8265ba452b94859ec4c Mon Sep 17 00:00:00 2001 From: Danny McCormick Date: Thu, 6 Aug 2026 15:19:24 +0000 Subject: [PATCH 2/7] Fix release validation parameter defaults for workflow dispatch --- .../workflows/beam_PostRelease_NightlySnapshot.yml | 11 ++++++----- .../org/apache/beam/gradle/BeamModulePlugin.groovy | 2 +- release/build.gradle.kts | 2 +- 3 files changed, 8 insertions(+), 7 deletions(-) diff --git a/.github/workflows/beam_PostRelease_NightlySnapshot.yml b/.github/workflows/beam_PostRelease_NightlySnapshot.yml index d4e2e0690cfb..05bd075050e0 100644 --- a/.github/workflows/beam_PostRelease_NightlySnapshot.yml +++ b/.github/workflows/beam_PostRelease_NightlySnapshot.yml @@ -19,12 +19,13 @@ on: workflow_dispatch: inputs: RELEASE: - description: Beam version of current release (e.g. 2.XX.0) - required: true - default: '2.XX.0' + description: Beam version of current release (e.g. 2.XX.0, or leave empty for nightly SNAPSHOT) + required: false + default: '' SNAPSHOT_URL: - description: Location of the staged artifacts in Maven central (https://repository.apache.org/content/repositories/orgapachebeam-NNNN/). - required: true + description: Location of the staged artifacts in Maven central (https://repository.apache.org/content/repositories/orgapachebeam-NNNN/ or leave empty for snapshots). + required: false + default: '' schedule: - cron: '15 16 * * *' diff --git a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy index 225201a5c0d8..e10cd483ef52 100644 --- a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy +++ b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy @@ -2643,7 +2643,7 @@ class BeamModulePlugin implements Plugin { JavaExamplesArchetypeValidationConfiguration config = it as JavaExamplesArchetypeValidationConfiguration def taskName = "run${config.type}Java${config.runner}" - def releaseVersion = project.findProperty('ver') ?: project.version + def releaseVersion = (project.findProperty('ver') && project.findProperty('ver') != '2.XX.0') ? project.findProperty('ver') : project.version def releaseRepo = project.findProperty('repourl') ?: 'https://repository.apache.org/content/repositories/snapshots' // shared maven local path for maven archetype projects def sharedMavenLocal = project.findProperty('mavenLocalPath') ?: '' diff --git a/release/build.gradle.kts b/release/build.gradle.kts index 54165dc49654..6cf3540a7f88 100644 --- a/release/build.gradle.kts +++ b/release/build.gradle.kts @@ -41,7 +41,7 @@ task("runJavaExamplesValidationTask") { dependsOn(":runners:spark:3:runQuickstartJavaSpark") dependsOn(":runners:flink:2.2:runQuickstartJavaFlinkLocal") dependsOn(":runners:direct-java:runMobileGamingJavaDirect") - if (project.hasProperty("ver") || !project.version.toString().endsWith("SNAPSHOT")) { + if ((project.findProperty("ver")?.toString()?.isNotEmpty() == true && project.findProperty("ver") != "2.XX.0") || !project.version.toString().endsWith("SNAPSHOT")) { // only run one variant of MobileGaming on Dataflow for nightly dependsOn(":runners:google-cloud-dataflow-java:runMobileGamingJavaDataflow") } From 691dc6ae5e4c62788781a7f0533766c07923db4c Mon Sep 17 00:00:00 2001 From: Danny McCormick Date: Thu, 6 Aug 2026 15:51:04 +0000 Subject: [PATCH 3/7] Add required JVM add-opens flags for Spark quickstart on Java 17/21 --- .../main/groovy/quickstart-java-spark.groovy | 20 +++++++++++++++---- 1 file changed, 16 insertions(+), 4 deletions(-) diff --git a/release/src/main/groovy/quickstart-java-spark.groovy b/release/src/main/groovy/quickstart-java-spark.groovy index 3c5be754daab..7832a191738f 100644 --- a/release/src/main/groovy/quickstart-java-spark.groovy +++ b/release/src/main/groovy/quickstart-java-spark.groovy @@ -30,10 +30,22 @@ t.describe 'Run Apache Beam Java SDK Quickstart - Spark' t.intent 'Runs the WordCount Code with Spark runner' // Run the wordcount example with the spark runner - t.run """mvn compile exec:java -q \ - -Dexec.mainClass=org.apache.beam.examples.WordCount \ - -Dexec.args="--inputFile=pom.xml --output=counts \ - --runner=SparkRunner" -Pspark-runner""" + + // Retrieve classpath + def deps = t.run """mvn compile dependency:build-classpath -q \ + -Dmdep.outputFile=/dev/stdout \ + -Dmaven.wagon.http.retryHandler.class=default \ + -Dmaven.wagon.http.retryHandler.count=5 \ + -Dmaven.wagon.http.pool=false \ + -Dmaven.wagon.httpconnectionManager.ttlSeconds=120 \ + -Dhttp.keepAlive=false \ + -Pspark-runner""" + + def cp = "target/classes:${deps}" + def jvmArgs = "--add-opens=java.base/sun.nio.ch=ALL-UNNAMED --add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.base/java.lang.invoke=ALL-UNNAMED --add-opens=java.base/java.lang=ALL-UNNAMED" + t.run """mvn exec:exec -q -Dexec.executable=java \ + -Dexec.args="${jvmArgs} -cp ${cp} org.apache.beam.examples.WordCount \ + --inputFile=pom.xml --output=counts --runner=SparkRunner" """ // Verify text from the pom.xml input file String result = t.run "grep Foundation counts*" From 7eba486a8d3b24b7f4a5c797933ab09f3d055908 Mon Sep 17 00:00:00 2001 From: Danny McCormick Date: Thu, 6 Aug 2026 17:00:58 -0400 Subject: [PATCH 4/7] Modify RELEASE input description and default value Updated the default release version and modified the description for the RELEASE input. --- .github/workflows/beam_PostRelease_NightlySnapshot.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/beam_PostRelease_NightlySnapshot.yml b/.github/workflows/beam_PostRelease_NightlySnapshot.yml index 05bd075050e0..d119dfb7754a 100644 --- a/.github/workflows/beam_PostRelease_NightlySnapshot.yml +++ b/.github/workflows/beam_PostRelease_NightlySnapshot.yml @@ -19,9 +19,9 @@ on: workflow_dispatch: inputs: RELEASE: - description: Beam version of current release (e.g. 2.XX.0, or leave empty for nightly SNAPSHOT) + description: Beam version of current release (pass in empty string for nightly SNAPSHOT) required: false - default: '' + default: '2.XX.0' SNAPSHOT_URL: description: Location of the staged artifacts in Maven central (https://repository.apache.org/content/repositories/orgapachebeam-NNNN/ or leave empty for snapshots). required: false From 5102cf3ec13db4809b4bff9d9b8155113ae9151f Mon Sep 17 00:00:00 2001 From: Danny McCormick Date: Fri, 7 Aug 2026 10:39:52 +0000 Subject: [PATCH 5/7] Check background process exit status and clean up tables between runs --- release/src/main/groovy/TestScripts.groovy | 59 ++++++++++++++++--- .../groovy/mobilegaming-java-dataflow.groovy | 20 ++++--- .../groovy/mobilegaming-java-direct.groovy | 35 ++++++++--- 3 files changed, 92 insertions(+), 22 deletions(-) diff --git a/release/src/main/groovy/TestScripts.groovy b/release/src/main/groovy/TestScripts.groovy index c9d5ab989fe6..0fae2af01641 100644 --- a/release/src/main/groovy/TestScripts.groovy +++ b/release/src/main/groovy/TestScripts.groovy @@ -25,6 +25,11 @@ import groovy.util.CliBuilder */ class TestScripts { + class BackgroundProcessInfo { + Process process + String cmd + } + // Global state to maintain when running the steps class var { static File startDir @@ -38,6 +43,7 @@ class TestScripts { static String pubsubTopic static String mavenLocalPath static List backgroundProcesses = Collections.synchronizedList(new ArrayList()) + static Map backgroundProcessInfo = Collections.synchronizedMap(new HashMap()) } def TestScripts(String[] args) { @@ -150,24 +156,58 @@ class TestScripts { } } + // Check whether any background processes exited unexpectedly with a non-zero exit code + public void checkBackgroundProcesses() { + def procs = new ArrayList<>(var.backgroundProcesses) + for (Process proc : procs) { + if (proc != null && !proc.isAlive()) { + int exitVal = proc.exitValue() + if (exitVal != 0) { + def info = var.backgroundProcessInfo.get(proc) + String cmd = info ? info.cmd : "unknown command" + error("Background command failed with exit code ${exitVal}: ${cmd}") + } + } + } + } + // Stop/kill a background process and all its descendants. public void stopProcess(Process proc) { - if (proc != null && proc.isAlive()) { - try { - proc.descendants().forEach { it.destroyForcibly() } - } catch (Throwable ignored) { + if (proc != null) { + if (!proc.isAlive()) { + int exitVal = proc.exitValue() + var.backgroundProcesses.remove(proc) + def info = var.backgroundProcessInfo.remove(proc) + if (exitVal != 0) { + String cmd = info ? info.cmd : "unknown command" + error("Background command failed with exit code ${exitVal}: ${cmd}") + } + } else { + try { + proc.descendants().forEach { it.destroyForcibly() } + } catch (Throwable ignored) { + } + proc.destroyForcibly() + proc.waitFor(10, java.util.concurrent.TimeUnit.SECONDS) + var.backgroundProcesses.remove(proc) + var.backgroundProcessInfo.remove(proc) } - proc.destroyForcibly() - proc.waitFor(10, java.util.concurrent.TimeUnit.SECONDS) } - var.backgroundProcesses.remove(proc) } // Stop all active background processes. public void stopAllBackgroundProcesses() { def procs = new ArrayList<>(var.backgroundProcesses) procs.each { proc -> - stopProcess(proc) + if (proc != null && proc.isAlive()) { + try { + proc.descendants().forEach { it.destroyForcibly() } + } catch (Throwable ignored) { + } + proc.destroyForcibly() + } + var.backgroundProcesses.remove(proc) + var.backgroundProcessInfo.remove(proc) } } @@ -195,6 +235,7 @@ class TestScripts { // Cleanup and print success public void done() { + checkBackgroundProcesses() stopAllBackgroundProcesses() var.startDir.deleteDir() println "[SUCCESS]" @@ -203,6 +244,7 @@ class TestScripts { // Run a single command, capture output, verify return code is 0 private String _execute(String cmd) { + checkBackgroundProcesses() def shell = "sh -c cmd".split(' ') shell[2] = cmd def pb = new ProcessBuilder(shell) @@ -233,6 +275,7 @@ class TestScripts { pb.redirectErrorStream(true) def proc = pb.start() var.backgroundProcesses.add(proc) + var.backgroundProcessInfo.put(proc, new BackgroundProcessInfo(process: proc, cmd: cmd)) Thread.startDaemon { try { proc.inputStream.eachLine { diff --git a/release/src/main/groovy/mobilegaming-java-dataflow.groovy b/release/src/main/groovy/mobilegaming-java-dataflow.groovy index 96cd557562ce..bf3778da95ec 100644 --- a/release/src/main/groovy/mobilegaming-java-dataflow.groovy +++ b/release/src/main/groovy/mobilegaming-java-dataflow.groovy @@ -120,16 +120,22 @@ class LeaderBoardRunner { ].join(",") String tables = t.run("bq query --use_legacy_sql=false 'SELECT table_name FROM ${dataset}.INFORMATION_SCHEMA.TABLES'") - - if (!tables.contains(userTable)) { - t.intent("Creating table: ${userTable}") - t.run("bq mk --table ${dataset}.${userTable} ${userSchema}") + if (tables.contains(userTable)) { + t.run("bq rm -f -t ${dataset}.${userTable}") + } + if (tables.contains(teamTable)) { + t.run("bq rm -f -t ${dataset}.${teamTable}") } - if (!tables.contains(teamTable)) { - t.intent("Creating table: ${teamTable}") - t.run("bq mk --table ${dataset}.${teamTable} ${teamSchema}") + while (tables.contains(userTable) || tables.contains(teamTable)) { + sleep(3000) + tables = t.run("bq query --use_legacy_sql=false 'SELECT table_name FROM ${dataset}.INFORMATION_SCHEMA.TABLES'") } + t.intent("Creating table: ${userTable}") + t.run("bq mk --table ${dataset}.${userTable} ${userSchema}") + t.intent("Creating table: ${teamTable}") + t.run("bq mk --table ${dataset}.${teamTable} ${teamSchema}") + // Verify that the tables have been created successfully tables = t.run("bq query --use_legacy_sql=false 'SELECT table_name FROM ${dataset}.INFORMATION_SCHEMA.TABLES'") while (!tables.contains(userTable) || !tables.contains(teamTable)) { diff --git a/release/src/main/groovy/mobilegaming-java-direct.groovy b/release/src/main/groovy/mobilegaming-java-direct.groovy index 3985c7e59634..042865c91d2f 100644 --- a/release/src/main/groovy/mobilegaming-java-direct.groovy +++ b/release/src/main/groovy/mobilegaming-java-direct.groovy @@ -80,16 +80,25 @@ def teamSchema = [ ].join(",") String tables = t.run("bq query --use_legacy_sql=false 'SELECT table_name FROM ${dataset}.INFORMATION_SCHEMA.TABLES'") - -if (!tables.contains(userTable)) { - t.intent("Creating table: ${userTable}") - t.run("bq mk --table ${dataset}.${userTable} ${userSchema}") +if (tables.contains(userTable)) { + t.run("bq rm -f -t ${dataset}.${userTable}") } -if (!tables.contains(teamTable)) { - t.intent("Creating table: ${teamTable}") - t.run("bq mk --table ${dataset}.${teamTable} ${teamSchema}") +if (tables.contains(teamTable)) { + t.run("bq rm -f -t ${dataset}.${teamTable}") } +// Make sure old tables are completely deleted before recreating them +tables = t.run("bq query --use_legacy_sql=false 'SELECT table_name FROM ${dataset}.INFORMATION_SCHEMA.TABLES'") +while (tables.contains(userTable) || tables.contains(teamTable)) { + sleep(3000) + tables = t.run("bq query --use_legacy_sql=false 'SELECT table_name FROM ${dataset}.INFORMATION_SCHEMA.TABLES'") +} + +t.intent("Creating table: ${userTable}") +t.run("bq mk --table ${dataset}.${userTable} ${userSchema}") +t.intent("Creating table: ${teamTable}") +t.run("bq mk --table ${dataset}.${teamTable} ${teamSchema}") + // Verify that the tables have been created tables = t.run("bq query --use_legacy_sql=false 'SELECT table_name FROM ${dataset}.INFORMATION_SCHEMA.TABLES'") while (!tables.contains(userTable) || !tables.contains(teamTable)) { @@ -132,4 +141,16 @@ if(!isSuccess){ } t.success("LeaderBoard successfully run on DirectRunner.") +tables = t.run("bq query --use_legacy_sql=false 'SELECT table_name FROM ${dataset}.INFORMATION_SCHEMA.TABLES'") +if (tables.contains(userTable)) { + t.run("bq rm -f -t ${dataset}.${userTable}") +} +if (tables.contains(teamTable)) { + t.run("bq rm -f -t ${dataset}.${teamTable}") +} +while (tables.contains(userTable) || tables.contains(teamTable)) { + sleep(3000) + tables = t.run("bq query --use_legacy_sql=false 'SELECT table_name FROM ${dataset}.INFORMATION_SCHEMA.TABLES'") +} + t.done() From 080c9d18e0b6d09a3d5a230bc54d9facc3901556 Mon Sep 17 00:00:00 2001 From: Danny McCormick Date: Fri, 7 Aug 2026 10:42:59 +0000 Subject: [PATCH 6/7] Use bounded retry loops when checking BigQuery tables --- .../groovy/mobilegaming-java-dataflow.groovy | 39 ++++++++++++++---- .../groovy/mobilegaming-java-direct.groovy | 40 ++++++++++++++----- 2 files changed, 62 insertions(+), 17 deletions(-) diff --git a/release/src/main/groovy/mobilegaming-java-dataflow.groovy b/release/src/main/groovy/mobilegaming-java-dataflow.groovy index bf3778da95ec..31b4f0670f98 100644 --- a/release/src/main/groovy/mobilegaming-java-dataflow.groovy +++ b/release/src/main/groovy/mobilegaming-java-dataflow.groovy @@ -126,9 +126,18 @@ class LeaderBoardRunner { if (tables.contains(teamTable)) { t.run("bq rm -f -t ${dataset}.${teamTable}") } - while (tables.contains(userTable) || tables.contains(teamTable)) { - sleep(3000) + int retries = 10 + boolean deleted = false + for (int i = 0; i < retries; i++) { tables = t.run("bq query --use_legacy_sql=false 'SELECT table_name FROM ${dataset}.INFORMATION_SCHEMA.TABLES'") + if (!tables.contains(userTable) && !tables.contains(teamTable)) { + deleted = true + break + } + sleep(3000) + } + if (!deleted) { + t.error("Timed out waiting for tables ${userTable} / ${teamTable} to be deleted.") } t.intent("Creating table: ${userTable}") @@ -137,10 +146,17 @@ class LeaderBoardRunner { t.run("bq mk --table ${dataset}.${teamTable} ${teamSchema}") // Verify that the tables have been created successfully - tables = t.run("bq query --use_legacy_sql=false 'SELECT table_name FROM ${dataset}.INFORMATION_SCHEMA.TABLES'") - while (!tables.contains(userTable) || !tables.contains(teamTable)) { - sleep(3000) + boolean created = false + for (int i = 0; i < retries; i++) { tables = t.run("bq query --use_legacy_sql=false 'SELECT table_name FROM ${dataset}.INFORMATION_SCHEMA.TABLES'") + if (tables.contains(userTable) && tables.contains(teamTable)) { + created = true + break + } + sleep(3000) + } + if (!created) { + t.error("Timed out waiting for tables ${userTable} / ${teamTable} to be created.") } println "Tables ${userTable} and ${teamTable} created successfully." @@ -202,10 +218,17 @@ fi // It will take couple seconds to clean up tables. // This loop makes sure tables are completely deleted before running the pipeline - tables = t.run("bq query --use_legacy_sql=false 'SELECT table_name FROM ${dataset}.INFORMATION_SCHEMA.TABLES'") - while (tables.contains(userTable) || tables.contains(teamTable)) { - sleep(3000) + deleted = false + for (int i = 0; i < retries; i++) { tables = t.run("bq query --use_legacy_sql=false 'SELECT table_name FROM ${dataset}.INFORMATION_SCHEMA.TABLES'") + if (!tables.contains(userTable) && !tables.contains(teamTable)) { + deleted = true + break + } + sleep(3000) + } + if (!deleted) { + println "Warning: Timed out waiting for tables ${userTable} / ${teamTable} to be deleted." } } } diff --git a/release/src/main/groovy/mobilegaming-java-direct.groovy b/release/src/main/groovy/mobilegaming-java-direct.groovy index 042865c91d2f..398822a9a2ce 100644 --- a/release/src/main/groovy/mobilegaming-java-direct.groovy +++ b/release/src/main/groovy/mobilegaming-java-direct.groovy @@ -87,11 +87,18 @@ if (tables.contains(teamTable)) { t.run("bq rm -f -t ${dataset}.${teamTable}") } -// Make sure old tables are completely deleted before recreating them -tables = t.run("bq query --use_legacy_sql=false 'SELECT table_name FROM ${dataset}.INFORMATION_SCHEMA.TABLES'") -while (tables.contains(userTable) || tables.contains(teamTable)) { - sleep(3000) +int retries = 10 +boolean deleted = false +for (int i = 0; i < retries; i++) { tables = t.run("bq query --use_legacy_sql=false 'SELECT table_name FROM ${dataset}.INFORMATION_SCHEMA.TABLES'") + if (!tables.contains(userTable) && !tables.contains(teamTable)) { + deleted = true + break + } + sleep(3000) +} +if (!deleted) { + t.error("Timed out waiting for tables ${userTable} / ${teamTable} to be deleted.") } t.intent("Creating table: ${userTable}") @@ -100,10 +107,17 @@ t.intent("Creating table: ${teamTable}") t.run("bq mk --table ${dataset}.${teamTable} ${teamSchema}") // Verify that the tables have been created -tables = t.run("bq query --use_legacy_sql=false 'SELECT table_name FROM ${dataset}.INFORMATION_SCHEMA.TABLES'") -while (!tables.contains(userTable) || !tables.contains(teamTable)) { - sleep(3000) +boolean created = false +for (int i = 0; i < retries; i++) { tables = t.run("bq query --use_legacy_sql=false 'SELECT table_name FROM ${dataset}.INFORMATION_SCHEMA.TABLES'") + if (tables.contains(userTable) && tables.contains(teamTable)) { + created = true + break + } + sleep(3000) +} +if (!created) { + t.error("Timed out waiting for tables ${userTable} / ${teamTable} to be created.") } println "Tables ${userTable} and ${teamTable} created successfully." @@ -148,9 +162,17 @@ if (tables.contains(userTable)) { if (tables.contains(teamTable)) { t.run("bq rm -f -t ${dataset}.${teamTable}") } -while (tables.contains(userTable) || tables.contains(teamTable)) { - sleep(3000) +deleted = false +for (int i = 0; i < retries; i++) { tables = t.run("bq query --use_legacy_sql=false 'SELECT table_name FROM ${dataset}.INFORMATION_SCHEMA.TABLES'") + if (!tables.contains(userTable) && !tables.contains(teamTable)) { + deleted = true + break + } + sleep(3000) +} +if (!deleted) { + println "Warning: Timed out waiting for tables ${userTable} / ${teamTable} to be deleted." } t.done() From ddfa090716d7e3059998a56bf4e27f8217040e8b Mon Sep 17 00:00:00 2001 From: Danny McCormick Date: Fri, 7 Aug 2026 11:01:22 +0000 Subject: [PATCH 7/7] Address review feedback: add waitFor to stopAllBackgroundProcesses, deduplicate _mvn logic, trim and quote classpath --- release/src/main/groovy/TestScripts.groovy | 87 +++++++------------ .../groovy/quickstart-java-flinklocal.groovy | 4 +- .../main/groovy/quickstart-java-spark.groovy | 4 +- 3 files changed, 36 insertions(+), 59 deletions(-) diff --git a/release/src/main/groovy/TestScripts.groovy b/release/src/main/groovy/TestScripts.groovy index 0fae2af01641..e0e9cf454495 100644 --- a/release/src/main/groovy/TestScripts.groovy +++ b/release/src/main/groovy/TestScripts.groovy @@ -26,7 +26,6 @@ import groovy.util.CliBuilder class TestScripts { class BackgroundProcessInfo { - Process process String cmd } @@ -205,6 +204,10 @@ class TestScripts { } catch (Throwable ignored) { } proc.destroyForcibly() + try { + proc.waitFor(10, java.util.concurrent.TimeUnit.SECONDS) + } catch (Throwable ignored) { + } } var.backgroundProcesses.remove(proc) var.backgroundProcessInfo.remove(proc) @@ -275,7 +278,7 @@ class TestScripts { pb.redirectErrorStream(true) def proc = pb.start() var.backgroundProcesses.add(proc) - var.backgroundProcessInfo.put(proc, new BackgroundProcessInfo(process: proc, cmd: cmd)) + var.backgroundProcessInfo.put(proc, new BackgroundProcessInfo(cmd: cmd)) Thread.startDaemon { try { proc.inputStream.eachLine { @@ -295,8 +298,8 @@ class TestScripts { } } - // Run a maven command, setting up a new local repository and a settings.xml with a custom repository if needed - private String _mvn(String args) { + // Build the maven command string with custom repository and settings.xml + private String _buildMvnCmd(String args) { String mvnlocalPath = var.mavenLocalPath if (!(var.mavenLocalPath)) { mvnlocalPath = var.startDir @@ -304,69 +307,43 @@ class TestScripts { def m2 = new File(mvnlocalPath, ".m2/repository") m2.mkdirs() def settings = new File(mvnlocalPath, "settings.xml") - if(!settings.exists()) { - settings.write """ - - ${m2.absolutePath} - - - testrel - - - test.release - ${var.repoUrl} - - - - - - """ + if (!settings.exists()) { + settings.write """ + + ${m2.absolutePath} + + + testrel + + + test.release + ${var.repoUrl} + + + + + + """ } def cmd = "mvn ${args} -s ${settings.absolutePath} -Ptestrel -B" - String path = System.getenv("PATH"); + String path = System.getenv("PATH") // Set the path on jenkins executors to use a recent maven // MAVEN_HOME is not set on some executors, so default to 3.5.2 String maven_home = System.getenv("MAVEN_HOME") ?: '/usr/local/maven' println "Using maven ${maven_home}" def mvnPath = "${maven_home}/bin" def setPath = "export PATH=\"${mvnPath}:${path}\" && " - return _execute(setPath + cmd) + return setPath + cmd + } + + // Run a maven command, setting up a new local repository and a settings.xml with a custom repository if needed + private String _mvn(String args) { + return _execute(_buildMvnCmd(args)) } // Run a maven command in the background private Process _mvnBackground(String args) { - String mvnlocalPath = var.mavenLocalPath - if (!(var.mavenLocalPath)) { - mvnlocalPath = var.startDir - } - def m2 = new File(mvnlocalPath, ".m2/repository") - m2.mkdirs() - def settings = new File(mvnlocalPath, "settings.xml") - if(!settings.exists()) { - settings.write """ - - ${m2.absolutePath} - - - testrel - - - test.release - ${var.repoUrl} - - - - - - """ - } - def cmd = "mvn ${args} -s ${settings.absolutePath} -Ptestrel -B" - String path = System.getenv("PATH"); - String maven_home = System.getenv("MAVEN_HOME") ?: '/usr/local/maven' - println "Using maven ${maven_home}" - def mvnPath = "${maven_home}/bin" - def setPath = "export PATH=\"${mvnPath}:${path}\" && " - return _executeBackground(setPath + cmd) + return _executeBackground(_buildMvnCmd(args)) } // Clean up and report error diff --git a/release/src/main/groovy/quickstart-java-flinklocal.groovy b/release/src/main/groovy/quickstart-java-flinklocal.groovy index 36c6ddd38354..3cd59270c04d 100644 --- a/release/src/main/groovy/quickstart-java-flinklocal.groovy +++ b/release/src/main/groovy/quickstart-java-flinklocal.groovy @@ -41,9 +41,9 @@ t.describe 'Run Apache Beam Java SDK Quickstart - Flink Local' -Dhttp.keepAlive=false \ -Pflink-runner""" - def cp = "target/classes:${deps}" + def cp = "target/classes:${deps.trim()}" t.run """mvn exec:exec -q -Dexec.executable=java \ - -Dexec.args="-cp ${cp} org.apache.beam.examples.WordCount \ + -Dexec.args="-cp '${cp}' org.apache.beam.examples.WordCount \ --inputFile=pom.xml --output=counts --runner=FlinkRunner" """ // Verify text from the pom.xml input file diff --git a/release/src/main/groovy/quickstart-java-spark.groovy b/release/src/main/groovy/quickstart-java-spark.groovy index 7832a191738f..248e85dbcc59 100644 --- a/release/src/main/groovy/quickstart-java-spark.groovy +++ b/release/src/main/groovy/quickstart-java-spark.groovy @@ -41,10 +41,10 @@ t.describe 'Run Apache Beam Java SDK Quickstart - Spark' -Dhttp.keepAlive=false \ -Pspark-runner""" - def cp = "target/classes:${deps}" + def cp = "target/classes:${deps.trim()}" def jvmArgs = "--add-opens=java.base/sun.nio.ch=ALL-UNNAMED --add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.base/java.lang.invoke=ALL-UNNAMED --add-opens=java.base/java.lang=ALL-UNNAMED" t.run """mvn exec:exec -q -Dexec.executable=java \ - -Dexec.args="${jvmArgs} -cp ${cp} org.apache.beam.examples.WordCount \ + -Dexec.args="${jvmArgs} -cp '${cp}' org.apache.beam.examples.WordCount \ --inputFile=pom.xml --output=counts --runner=SparkRunner" """ // Verify text from the pom.xml input file