From 72d87622fbd4cd4025896997c85db9614465ca86 Mon Sep 17 00:00:00 2001 From: aibrahiim Date: Wed, 23 Sep 2026 18:56:12 +0300 Subject: [PATCH 1/6] reduce LeaderBoardIT flakiness --- .github/trigger_files/beam_PostCommit_Python.json | 2 +- .../examples/complete/game/leader_board_it_test.py | 11 +++++++++-- 2 files changed, 10 insertions(+), 3 deletions(-) diff --git a/.github/trigger_files/beam_PostCommit_Python.json b/.github/trigger_files/beam_PostCommit_Python.json index 9bfa92ff5106..9f1daca01a0f 100644 --- a/.github/trigger_files/beam_PostCommit_Python.json +++ b/.github/trigger_files/beam_PostCommit_Python.json @@ -1,5 +1,5 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run.", "pr": "40139", - "modification": 59 + "modification": 60 } diff --git a/sdks/python/apache_beam/examples/complete/game/leader_board_it_test.py b/sdks/python/apache_beam/examples/complete/game/leader_board_it_test.py index 4cc13171fe92..0daf414193b2 100644 --- a/sdks/python/apache_beam/examples/complete/game/leader_board_it_test.py +++ b/sdks/python/apache_beam/examples/complete/game/leader_board_it_test.py @@ -65,6 +65,7 @@ class LeaderBoardIT(unittest.TestCase): DEFAULT_INPUT_COUNT = 500 WAIT_UNTIL_FINISH_DURATION = 10 * 60 * 1000 # in milliseconds + BQ_MATCHER_TIMEOUT_SECS = 5 * 60 def setUp(self): self.test_pipeline = TestPipeline(is_integration_test=True) @@ -124,7 +125,10 @@ def test_leader_board_it(self): self.OUTPUT_TABLE_USERS, success_condition)) bq_users_verifier = BigqueryMatcher( - self.project, users_query, self.DEFAULT_EXPECTED_CHECKSUM) + self.project, + users_query, + self.DEFAULT_EXPECTED_CHECKSUM, + timeout_secs=self.BQ_MATCHER_TIMEOUT_SECS) teams_query = ( 'SELECT total_score FROM `%s.%s.%s` ' @@ -134,7 +138,10 @@ def test_leader_board_it(self): self.OUTPUT_TABLE_TEAMS, success_condition)) bq_teams_verifier = BigqueryMatcher( - self.project, teams_query, self.DEFAULT_EXPECTED_CHECKSUM) + self.project, + teams_query, + self.DEFAULT_EXPECTED_CHECKSUM, + timeout_secs=self.BQ_MATCHER_TIMEOUT_SECS) extra_opts = { 'allow_unsafe_triggers': True, From 106a047706e96775c20b4ee6d2cf7cdce3e38c00 Mon Sep 17 00:00:00 2001 From: aibrahiim Date: Thu, 24 Sep 2026 11:14:37 +0300 Subject: [PATCH 2/6] increase timeout --- .../examples/complete/game/leader_board_it_test.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/sdks/python/apache_beam/examples/complete/game/leader_board_it_test.py b/sdks/python/apache_beam/examples/complete/game/leader_board_it_test.py index 0daf414193b2..1d0f1afacdef 100644 --- a/sdks/python/apache_beam/examples/complete/game/leader_board_it_test.py +++ b/sdks/python/apache_beam/examples/complete/game/leader_board_it_test.py @@ -64,8 +64,9 @@ class LeaderBoardIT(unittest.TestCase): OUTPUT_TABLE_TEAMS = 'leader_board_teams' DEFAULT_INPUT_COUNT = 500 - WAIT_UNTIL_FINISH_DURATION = 10 * 60 * 1000 # in milliseconds - BQ_MATCHER_TIMEOUT_SECS = 5 * 60 + WAIT_UNTIL_FINISH_DURATION = 12 * 60 * 1000 # in milliseconds + # Poll BigQuery after the pipeline wait; streaming inserts can lag. + BQ_MATCHER_TIMEOUT_SECS = 10 * 60 def setUp(self): self.test_pipeline = TestPipeline(is_integration_test=True) From eb0dc9f06c0d43cc61a43ed839e2dea682124168 Mon Sep 17 00:00:00 2001 From: aibrahiim Date: Thu, 24 Sep 2026 15:51:58 +0300 Subject: [PATCH 3/6] add wait for PubSub publish --- .../examples/complete/game/leader_board_it_test.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/sdks/python/apache_beam/examples/complete/game/leader_board_it_test.py b/sdks/python/apache_beam/examples/complete/game/leader_board_it_test.py index 1d0f1afacdef..1fe23d76309b 100644 --- a/sdks/python/apache_beam/examples/complete/game/leader_board_it_test.py +++ b/sdks/python/apache_beam/examples/complete/game/leader_board_it_test.py @@ -99,9 +99,14 @@ def _inject_pubsub_game_events(self, topic, message_count): logging.debug( 'Injecting %d game events to topic %s', message_count, topic.name) + publish_futures = [] for _ in range(message_count): - self.pub_client.publish( - topic.name, (self.INPUT_EVENT % self._test_timestamp).encode('utf-8')) + publish_futures.append( + self.pub_client.publish( + topic.name, + (self.INPUT_EVENT % self._test_timestamp).encode('utf-8'))) + for future in publish_futures: + future.result() def _cleanup_pubsub(self): test_utils.cleanup_subscriptions(self.sub_client, [self.input_sub]) From f184031ac63fba35ee61a5ec0c691b79c98d1f4d Mon Sep 17 00:00:00 2001 From: aibrahiim Date: Thu, 24 Sep 2026 20:35:20 +0300 Subject: [PATCH 4/6] trigger workflow --- .github/trigger_files/beam_PostCommit_Python.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/trigger_files/beam_PostCommit_Python.json b/.github/trigger_files/beam_PostCommit_Python.json index 9f1daca01a0f..b8989e344590 100644 --- a/.github/trigger_files/beam_PostCommit_Python.json +++ b/.github/trigger_files/beam_PostCommit_Python.json @@ -1,5 +1,5 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run.", "pr": "40139", - "modification": 60 + "modification": 61 } From a6f9865f703a6875e58932274671da2336131b1c Mon Sep 17 00:00:00 2001 From: aibrahiim Date: Mon, 28 Sep 2026 20:12:34 +0300 Subject: [PATCH 5/6] trigger workflow --- .github/trigger_files/beam_PostCommit_Python.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/trigger_files/beam_PostCommit_Python.json b/.github/trigger_files/beam_PostCommit_Python.json index b8989e344590..678a4b063fe7 100644 --- a/.github/trigger_files/beam_PostCommit_Python.json +++ b/.github/trigger_files/beam_PostCommit_Python.json @@ -1,5 +1,5 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run.", "pr": "40139", - "modification": 61 + "modification": 62 } From 4a88188a58dd97401bc77065039ca0b6580e8ab9 Mon Sep 17 00:00:00 2001 From: aibrahiim Date: Tue, 29 Sep 2026 11:03:08 +0300 Subject: [PATCH 6/6] wait for PubSub subscription readiness --- .github/trigger_files/beam_PostCommit_Python.json | 4 ++-- .../examples/complete/game/leader_board_it_test.py | 4 +++- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/.github/trigger_files/beam_PostCommit_Python.json b/.github/trigger_files/beam_PostCommit_Python.json index 678a4b063fe7..e11d2260913f 100644 --- a/.github/trigger_files/beam_PostCommit_Python.json +++ b/.github/trigger_files/beam_PostCommit_Python.json @@ -1,5 +1,5 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run.", - "pr": "40139", - "modification": 62 + "pr": "40241", + "modification": 63 } diff --git a/sdks/python/apache_beam/examples/complete/game/leader_board_it_test.py b/sdks/python/apache_beam/examples/complete/game/leader_board_it_test.py index 1fe23d76309b..218d4e4a7787 100644 --- a/sdks/python/apache_beam/examples/complete/game/leader_board_it_test.py +++ b/sdks/python/apache_beam/examples/complete/game/leader_board_it_test.py @@ -64,7 +64,7 @@ class LeaderBoardIT(unittest.TestCase): OUTPUT_TABLE_TEAMS = 'leader_board_teams' DEFAULT_INPUT_COUNT = 500 - WAIT_UNTIL_FINISH_DURATION = 12 * 60 * 1000 # in milliseconds + WAIT_UNTIL_FINISH_DURATION = 15 * 60 * 1000 # in milliseconds # Poll BigQuery after the pipeline wait; streaming inserts can lag. BQ_MATCHER_TIMEOUT_SECS = 10 * 60 @@ -86,6 +86,8 @@ def setUp(self): name=self.sub_client.subscription_path( self.project, self.INPUT_SUB + _unique_id), topic=self.input_topic.name) + # New subscriptions can miss messages published immediately after create. + time.sleep(30) # Set up BigQuery environment self.dataset_ref = utils.create_bq_dataset(