diff --git a/.github/workflows/beam_PostRelease_NightlySnapshot.yml b/.github/workflows/beam_PostRelease_NightlySnapshot.yml index d4e2e0690cfb..d119dfb7754a 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 + description: Beam version of current release (pass in empty string for nightly SNAPSHOT) + required: false default: '2.XX.0' 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") } diff --git a/release/src/main/groovy/TestScripts.groovy b/release/src/main/groovy/TestScripts.groovy index dc2438007ac1..e0e9cf454495 100644 --- a/release/src/main/groovy/TestScripts.groovy +++ b/release/src/main/groovy/TestScripts.groovy @@ -25,6 +25,10 @@ import groovy.util.CliBuilder */ class TestScripts { + class BackgroundProcessInfo { + String cmd + } + // Global state to maintain when running the steps class var { static File startDir @@ -37,6 +41,8 @@ class TestScripts { static String bqDataset static String pubsubTopic static String mavenLocalPath + static List backgroundProcesses = Collections.synchronizedList(new ArrayList()) + static Map backgroundProcessInfo = Collections.synchronizedMap(new HashMap()) } def TestScripts(String[] args) { @@ -79,6 +85,10 @@ class TestScripts { var.mavenLocalPath = options.mavenLocalPath println "Maven local path: ${var.mavenLocalPath}" } + + Runtime.getRuntime().addShutdownHook(new Thread({ + stopAllBackgroundProcesses() + })) } def ver() { @@ -135,6 +145,75 @@ 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) + } + } + + // 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) { + 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) + } + } + } + + // Stop all active background processes. + public void stopAllBackgroundProcesses() { + def procs = new ArrayList<>(var.backgroundProcesses) + procs.each { proc -> + if (proc != null && proc.isAlive()) { + try { + proc.descendants().forEach { it.destroyForcibly() } + } 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) + } + } + // 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 +238,8 @@ class TestScripts { // Cleanup and print success public void done() { + checkBackgroundProcesses() + stopAllBackgroundProcesses() var.startDir.deleteDir() println "[SUCCESS]" System.exit(0) @@ -166,6 +247,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) @@ -187,6 +269,27 @@ 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) + var.backgroundProcessInfo.put(proc, new BackgroundProcessInfo(cmd: cmd)) + 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) @@ -195,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 @@ -204,37 +307,48 @@ 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) { + return _executeBackground(_buildMvnCmd(args)) } // 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..31b4f0670f98 100644 --- a/release/src/main/groovy/mobilegaming-java-dataflow.groovy +++ b/release/src/main/groovy/mobilegaming-java-dataflow.groovy @@ -120,37 +120,53 @@ 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}") + 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}") + 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)) { - 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." - 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 +191,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 @@ -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 34eab4c00768..398822a9a2ce 100644 --- a/release/src/main/groovy/mobilegaming-java-direct.groovy +++ b/release/src/main/groovy/mobilegaming-java-direct.groovy @@ -80,32 +80,51 @@ 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.run("bq rm -f -t ${dataset}.${userTable}") +} +if (tables.contains(teamTable)) { + t.run("bq rm -f -t ${dataset}.${teamTable}") +} -if (!tables.contains(userTable)) { - t.intent("Creating table: ${userTable}") - t.run("bq mk --table ${dataset}.${userTable} ${userSchema}") +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 (!tables.contains(teamTable)) { - t.intent("Creating table: ${teamTable}") - t.run("bq mk --table ${dataset}.${teamTable} ${teamSchema}") +if (!deleted) { + t.error("Timed out waiting for tables ${userTable} / ${teamTable} to be deleted.") } +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)) { - 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." -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,12 +147,32 @@ 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") } 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}") +} +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() 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 3c5be754daab..248e85dbcc59 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.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 \ + --inputFile=pom.xml --output=counts --runner=SparkRunner" """ // Verify text from the pom.xml input file String result = t.run "grep Foundation counts*"