Skip to content

Commit 077589a

Browse files
Amar3ttoAmar3tto
authored
Fix PostRelease Nightly Snapshot job (#34055)
* Refactor mobilegaming groovy scripts * Add retry policy
1 parent 22f2fbe commit 077589a

4 files changed

Lines changed: 90 additions & 28 deletions

File tree

‎examples/java/src/main/java/org/apache/beam/examples/complete/game/utils/WriteWindowedToBigQuery.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
2323
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition;
2424
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.WriteDisposition;
25+
import org.apache.beam.sdk.io.gcp.bigquery.InsertRetryPolicy;
2526
import org.apache.beam.sdk.transforms.DoFn;
2627
import org.apache.beam.sdk.transforms.ParDo;
2728
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
@@ -64,7 +65,8 @@ public PDone expand(PCollection<T> teamAndScore) {
6465
.to(getTable(projectId, datasetId, tableName))
6566
.withSchema(getSchema())
6667
.withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
67-
.withWriteDisposition(WriteDisposition.WRITE_APPEND));
68+
.withWriteDisposition(WriteDisposition.WRITE_APPEND)
69+
.withFailedInsertRetryPolicy(InsertRetryPolicy.retryTransientErrors()));
6870
return PDone.in(teamAndScore.getPipeline());
6971
}
7072
}

‎release/src/main/groovy/MobileGamingCommands.groovy‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,7 @@ class MobileGamingCommands {
3030
SparkRunner: "spark-runner",
3131
FlinkRunner: "flink-runner"]
3232

33-
public static final EXECUTION_TIMEOUT_IN_MINUTES = 60
33+
public static final EXECUTION_TIMEOUT_IN_MINUTES = 80
3434

3535
// Lists used to verify team names generated in the LeaderBoard example.
3636
// This list should be kept sync with COLORS in org.apache.beam.examples.complete.game.injector.Injector.

‎release/src/main/groovy/mobilegaming-java-dataflow.groovy‎

Lines changed: 47 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -66,16 +66,53 @@ class LeaderBoardRunner {
6666
def run(runner, TestScripts t, MobileGamingCommands mobileGamingCommands, boolean useStreamingEngine) {
6767
t.intent("Running: LeaderBoard example on DataflowRunner" +
6868
(useStreamingEngine ? " with Streaming Engine" : ""))
69-
t.run("bq rm -f -t ${t.bqDataset()}.leaderboard_DataflowRunner_user")
70-
t.run("bq rm -f -t ${t.bqDataset()}.leaderboard_DataflowRunner_team")
69+
70+
def dataset = t.bqDataset()
71+
def userTable = "leaderboard_DataflowRunner_user"
72+
def teamTable = "leaderboard_DataflowRunner_team"
73+
def userSchema = [
74+
"user:STRING",
75+
"total_score:INTEGER",
76+
"processing_time:STRING"
77+
].join(",")
78+
def teamSchema = [
79+
"team:STRING",
80+
"total_score:INTEGER",
81+
"window_start:STRING",
82+
"processing_time:STRING",
83+
"timing:STRING"
84+
].join(",")
85+
86+
// Remove existing tables if they exist
87+
String tables = t.run("bq query --use_legacy_sql=false 'SELECT table_name FROM ${dataset}.INFORMATION_SCHEMA.TABLES'")
88+
89+
if (tables.contains(userTable)) {
90+
t.run("bq rm -f -t ${dataset}.${userTable}")
91+
}
92+
if (tables.contains(teamTable)) {
93+
t.run("bq rm -f -t ${dataset}.${teamTable}")
94+
}
95+
7196
// It will take couple seconds to clean up tables.
7297
// This loop makes sure tables are completely deleted before running the pipeline
73-
String tables = ""
74-
while ({
98+
tables = t.run("bq query --use_legacy_sql=false 'SELECT table_name FROM ${dataset}.INFORMATION_SCHEMA.TABLES'")
99+
while (tables.contains(userTable) || tables.contains(teamTable)) {
75100
sleep(3000)
76-
tables = t.run("bq query SELECT table_id FROM ${t.bqDataset()}.__TABLES_SUMMARY__")
77-
tables.contains("leaderboard_${}_user") || tables.contains("leaderboard_${runner}_team")
78-
}());
101+
tables = t.run("bq query --use_legacy_sql=false 'SELECT table_name FROM ${dataset}.INFORMATION_SCHEMA.TABLES'")
102+
}
103+
104+
t.intent("Creating table: ${userTable}")
105+
t.run("bq mk --table ${dataset}.${userTable} ${userSchema}")
106+
t.intent("Creating table: ${teamTable}")
107+
t.run("bq mk --table ${dataset}.${teamTable} ${teamSchema}")
108+
109+
// Verify that the tables have been created successfully
110+
tables = t.run("bq query --use_legacy_sql=false 'SELECT table_name FROM ${dataset}.INFORMATION_SCHEMA.TABLES'")
111+
while (!tables.contains(userTable) || !tables.contains(teamTable)) {
112+
sleep(3000)
113+
tables = t.run("bq query --use_legacy_sql=false 'SELECT table_name FROM ${dataset}.INFORMATION_SCHEMA.TABLES'")
114+
}
115+
println "Tables ${userTable} and ${teamTable} created successfully."
79116

80117
def InjectorThread = Thread.start() {
81118
t.run(mobileGamingCommands.createInjectorCommand())
@@ -99,11 +136,9 @@ class LeaderBoardRunner {
99136
String query_result = ""
100137
while ((System.currentTimeMillis() - startTime) / 60000 < mobileGamingCommands.EXECUTION_TIMEOUT_IN_MINUTES) {
101138
try {
102-
tables = t.run "bq query --use_legacy_sql=false SELECT table_name FROM ${t.bqDataset()}.INFORMATION_SCHEMA.TABLES"
103-
if (tables.contains("leaderboard_${runner}_user") && tables.contains("leaderboard_${runner}_team")) {
104-
query_result = t.run """bq query --batch "SELECT user FROM [${t.gcpProject()}:${
105-
t.bqDataset()
106-
}.leaderboard_${runner}_user] LIMIT 10\""""
139+
tables = t.run "bq query --use_legacy_sql=false SELECT table_name FROM ${dataset}.INFORMATION_SCHEMA.TABLES"
140+
if (tables.contains(userTable) && tables.contains(teamTable)) {
141+
query_result = t.run """bq query --batch "SELECT user FROM [${dataset}.${userTable}] LIMIT 10\""""
107142
if (t.seeAnyOf(mobileGamingCommands.COLORS, query_result)) {
108143
isSuccess = true
109144
break

‎release/src/main/groovy/mobilegaming-java-direct.groovy‎

Lines changed: 39 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -62,16 +62,41 @@ t.success("HourlyTeamScore successfully run on DirectRunners.")
6262
* */
6363

6464
t.intent("Running: LeaderBoard example on DirectRunner")
65-
t.run("bq rm -f -t ${t.bqDataset()}.leaderboard_DirectRunner_user")
66-
t.run("bq rm -f -t ${t.bqDataset()}.leaderboard_DirectRunner_team")
67-
// It will take couple seconds to clean up tables.
68-
// This loop makes sure tables are completely deleted before running the pipeline
69-
String tables = ""
70-
while({
65+
66+
def dataset = t.bqDataset()
67+
def userTable = "leaderboard_DirectRunner_user"
68+
def teamTable = "leaderboard_DirectRunner_team"
69+
def userSchema = [
70+
"user:STRING",
71+
"total_score:INTEGER",
72+
"processing_time:STRING"
73+
].join(",")
74+
def teamSchema = [
75+
"team:STRING",
76+
"total_score:INTEGER",
77+
"window_start:STRING",
78+
"processing_time:STRING",
79+
"timing:STRING"
80+
].join(",")
81+
82+
String tables = t.run("bq query --use_legacy_sql=false 'SELECT table_name FROM ${dataset}.INFORMATION_SCHEMA.TABLES'")
83+
84+
if (!tables.contains(userTable)) {
85+
t.intent("Creating table: ${userTable}")
86+
t.run("bq mk --table ${dataset}.${userTable} ${userSchema}")
87+
}
88+
if (!tables.contains(teamTable)) {
89+
t.intent("Creating table: ${teamTable}")
90+
t.run("bq mk --table ${dataset}.${teamTable} ${teamSchema}")
91+
}
92+
93+
// Verify that the tables have been created
94+
tables = t.run("bq query --use_legacy_sql=false 'SELECT table_name FROM ${dataset}.INFORMATION_SCHEMA.TABLES'")
95+
while (!tables.contains(userTable) || !tables.contains(teamTable)) {
7196
sleep(3000)
72-
tables = t.run ("bq query SELECT table_id FROM ${t.bqDataset()}.__TABLES_SUMMARY__")
73-
tables.contains("leaderboard_${runner}_user") || tables.contains("leaderboard_${runner}_team")
74-
}());
97+
tables = t.run("bq query --use_legacy_sql=false 'SELECT table_name FROM ${dataset}.INFORMATION_SCHEMA.TABLES'")
98+
}
99+
println "Tables ${userTable} and ${teamTable} created successfully."
75100

76101
def InjectorThread = Thread.start() {
77102
t.run(mobileGamingCommands.createInjectorCommand())
@@ -86,12 +111,12 @@ def LeaderBoardThread = Thread.start() {
86111
def startTime = System.currentTimeMillis()
87112
def isSuccess = false
88113
String query_result = ""
89-
while((System.currentTimeMillis() - startTime)/60000 < mobileGamingCommands.EXECUTION_TIMEOUT_IN_MINUTES) {
114+
while ((System.currentTimeMillis() - startTime)/60000 < mobileGamingCommands.EXECUTION_TIMEOUT_IN_MINUTES) {
90115
try {
91-
tables = t.run "bq query --use_legacy_sql=false SELECT table_name FROM ${t.bqDataset()}.INFORMATION_SCHEMA.TABLES"
92-
if(tables.contains("leaderboard_${runner}_user") && tables.contains("leaderboard_${runner}_team")) {
93-
query_result = t.run """bq query --batch "SELECT user FROM [${t.gcpProject()}.${t.bqDataset()}.leaderboard_${runner}_user] LIMIT 10\""""
94-
if(t.seeAnyOf(mobileGamingCommands.COLORS, query_result)){
116+
tables = t.run "bq query --use_legacy_sql=false SELECT table_name FROM ${dataset}.INFORMATION_SCHEMA.TABLES"
117+
if (tables.contains(userTable) && tables.contains(teamTable)) {
118+
query_result = t.run """bq query --batch "SELECT user FROM [${dataset}.${userTable}] LIMIT 10\""""
119+
if (t.seeAnyOf(mobileGamingCommands.COLORS, query_result)){
95120
isSuccess = true
96121
break
97122
}

0 commit comments

Comments
 (0)