From 8f33cb292238cbbbbb9b37fd73c436bc9c47e4ec Mon Sep 17 00:00:00 2001 From: aibrahiim Date: Wed, 30 Sep 2026 12:38:11 +0300 Subject: [PATCH] delay LeaderBoardIT Pub/Sub inject --- .../trigger_files/beam_PostCommit_Python.json | 4 +-- .../complete/game/leader_board_it_test.py | 28 ++++++++++++------- 2 files changed, 20 insertions(+), 12 deletions(-) diff --git a/.github/trigger_files/beam_PostCommit_Python.json b/.github/trigger_files/beam_PostCommit_Python.json index e11d2260913f..271390048a60 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": "40241", - "modification": 63 + "pr": "40241-followup", + "modification": 64 } 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 218d4e4a7787..27074b3d8788 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 @@ -34,6 +34,7 @@ # pytype: skip-file import logging +import threading import time import unittest import uuid @@ -65,8 +66,8 @@ class LeaderBoardIT(unittest.TestCase): DEFAULT_INPUT_COUNT = 500 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 + PUBLISH_DELAY_SECS = 2 * 60 def setUp(self): self.test_pipeline = TestPipeline(is_integration_test=True) @@ -86,7 +87,6 @@ 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 @@ -167,14 +167,22 @@ def test_leader_board_it(self): self.addCleanup(self._cleanup_pubsub) self.addCleanup(utils.delete_bq_dataset, self.project, self.dataset_ref) - # Generate input data and inject to PubSub. - self._inject_pubsub_game_events(self.input_topic, self.DEFAULT_INPUT_COUNT) - - # Get pipeline options from command argument: --test-pipeline-options, - # and start pipeline job by calling pipeline main function. - leader_board.run( - self.test_pipeline.get_full_options_as_args(**extra_opts), - save_main_session=False) + def _publish_after_delay(): + time.sleep(self.PUBLISH_DELAY_SECS) + self._inject_pubsub_game_events( + self.input_topic, self.DEFAULT_INPUT_COUNT) + + publish_thread = threading.Thread(target=_publish_after_delay, daemon=True) + publish_thread.start() + + try: + # Get pipeline options from command argument: --test-pipeline-options, + # and start pipeline job by calling pipeline main function. + leader_board.run( + self.test_pipeline.get_full_options_as_args(**extra_opts), + save_main_session=False) + finally: + publish_thread.join(timeout=self.PUBLISH_DELAY_SECS + 60) if __name__ == '__main__':