Skip to content

delay LeaderBoardIT Pub/Sub inject #40350

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
aIbrahiim wants to merge 1 commit into from
Closed

delay LeaderBoardIT Pub/Sub inject #40350

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions .github/trigger_files/beam_PostCommit_Python.json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
}
28 changes: 18 additions & 10 deletions sdks/python/apache_beam/examples/complete/game/leader_board_it_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
# pytype: skip-file

import logging
import threading
import time
import unittest
import uuid
Expand Down Expand Up @@ -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)
Expand All @@ -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
Expand Down Expand Up @@ -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__':
Expand Down
Loading