From c6c69eaf8538ebf8eb4dd3b9cc95b91abee80ce0 Mon Sep 17 00:00:00 2001 From: Danny McCormick Date: Wed, 11 Nov 2026 12:19:47 +0000 Subject: [PATCH 1/2] Sleep between create_stream retry attempts when waiting for LocalStack in Kinesis ITs --- .../python/apache_beam/io/external/xlang_kinesisio_it_test.py | 4 ++-- sdks/python/apache_beam/yaml/integration_tests.py | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/sdks/python/apache_beam/io/external/xlang_kinesisio_it_test.py b/sdks/python/apache_beam/io/external/xlang_kinesisio_it_test.py index 6a86248571e3..884760c60928 100644 --- a/sdks/python/apache_beam/io/external/xlang_kinesisio_it_test.py +++ b/sdks/python/apache_beam/io/external/xlang_kinesisio_it_test.py @@ -273,19 +273,19 @@ def __init__(self, access_key, secret_key, region, service_endpoint): def create_stream(self, stream_name): # localstack could not have initialized in the container yet so repeat - retries = 10 + retries = 15 for i in range(retries): try: self.kinesis_client.create_stream( StreamName=stream_name, ShardCount=1, ) - time.sleep(2) break except Exception as e: if i == retries - 1: logging.error('Could not create kinesis stream') raise e + time.sleep(2) # Wait for the stream to be active self.get_first_shard_id(stream_name) diff --git a/sdks/python/apache_beam/yaml/integration_tests.py b/sdks/python/apache_beam/yaml/integration_tests.py index 75db6642866e..1aed39b1582c 100644 --- a/sdks/python/apache_beam/yaml/integration_tests.py +++ b/sdks/python/apache_beam/yaml/integration_tests.py @@ -1006,16 +1006,16 @@ def temp_iceberg_table_with_pk(table_data): def _create_kinesis_stream(kinesis_client, stream_name): """Creates a Kinesis stream and waits until it is active.""" - retries = 10 + retries = 15 for attempt in range(retries): try: kinesis_client.create_stream(StreamName=stream_name, ShardCount=1) - time.sleep(2) break except Exception as exn: if attempt == retries - 1: _LOGGER.error('Could not create kinesis stream') raise exn + time.sleep(2) for attempt in range(retries): stream = kinesis_client.describe_stream(StreamName=stream_name) From 8cc143c170e8a8bf502bf9ec885d3424e2b76f18 Mon Sep 17 00:00:00 2001 From: Danny McCormick Date: Fri, 13 Nov 2026 05:11:20 +0000 Subject: [PATCH 2/2] Keep retries=10 and post-create sleep, add comment on except sleep --- .../apache_beam/io/external/xlang_kinesisio_it_test.py | 8 ++++++-- sdks/python/apache_beam/yaml/integration_tests.py | 8 ++++++-- 2 files changed, 12 insertions(+), 4 deletions(-) diff --git a/sdks/python/apache_beam/io/external/xlang_kinesisio_it_test.py b/sdks/python/apache_beam/io/external/xlang_kinesisio_it_test.py index 884760c60928..2d625b9337fc 100644 --- a/sdks/python/apache_beam/io/external/xlang_kinesisio_it_test.py +++ b/sdks/python/apache_beam/io/external/xlang_kinesisio_it_test.py @@ -273,19 +273,23 @@ def __init__(self, access_key, secret_key, region, service_endpoint): def create_stream(self, stream_name): # localstack could not have initialized in the container yet so repeat - retries = 15 + retries = 10 for i in range(retries): try: self.kinesis_client.create_stream( StreamName=stream_name, ShardCount=1, ) + time.sleep(2) break except Exception as e: if i == retries - 1: logging.error('Could not create kinesis stream') raise e - time.sleep(2) + # LocalStack's HTTP endpoint may not be ready to accept connections + # immediately after container startup; sleep before retrying so all + # attempts are not exhausted immediately on ConnectionClosedError. + time.sleep(3) # Wait for the stream to be active self.get_first_shard_id(stream_name) diff --git a/sdks/python/apache_beam/yaml/integration_tests.py b/sdks/python/apache_beam/yaml/integration_tests.py index 1aed39b1582c..01a699c6fe44 100644 --- a/sdks/python/apache_beam/yaml/integration_tests.py +++ b/sdks/python/apache_beam/yaml/integration_tests.py @@ -1006,16 +1006,20 @@ def temp_iceberg_table_with_pk(table_data): def _create_kinesis_stream(kinesis_client, stream_name): """Creates a Kinesis stream and waits until it is active.""" - retries = 15 + retries = 10 for attempt in range(retries): try: kinesis_client.create_stream(StreamName=stream_name, ShardCount=1) + time.sleep(2) break except Exception as exn: if attempt == retries - 1: _LOGGER.error('Could not create kinesis stream') raise exn - time.sleep(2) + # LocalStack's HTTP endpoint may not be ready to accept connections + # immediately after container startup; sleep before retrying so all + # attempts are not exhausted immediately on ConnectionClosedError. + time.sleep(3) for attempt in range(retries): stream = kinesis_client.describe_stream(StreamName=stream_name)