From c9782d8cac42ac06de37ae69978641991f38bb3a Mon Sep 17 00:00:00 2001 From: Baz <22987674+bazarnov@users.noreply.github.com> Date: Tue, 6 Oct 2026 00:05:03 +0300 Subject: [PATCH] docs(concurrent-cdk): mark `timeout_seconds` as deprecated and ignored `ConcurrentSource.timeout_seconds` has not been read since the queue timeout it fed was removed, yet the docstring still promised the read stops when no record arrives within it. Document it as deprecated and ignored and drop the unread attribute. The parameter and `DEFAULT_TIMEOUT_SECONDS` stay for existing callers; the constant gets a comment saying it is only that parameter's default. The `PartitionEnqueuer` and `PartitionReader` docstrings also said the main thread waits for a timeout when a worker exception never reaches the queue. It blocks on the queue forever. --- airbyte_cdk/sources/concurrent_source/concurrent_source.py | 5 ++--- airbyte_cdk/sources/streams/concurrent/partition_enqueuer.py | 2 +- airbyte_cdk/sources/streams/concurrent/partition_reader.py | 2 +- 3 files changed, 4 insertions(+), 5 deletions(-) diff --git a/airbyte_cdk/sources/concurrent_source/concurrent_source.py b/airbyte_cdk/sources/concurrent_source/concurrent_source.py index 474780bcc3..03aa3ab2f6 100644 --- a/airbyte_cdk/sources/concurrent_source/concurrent_source.py +++ b/airbyte_cdk/sources/concurrent_source/concurrent_source.py @@ -35,7 +35,7 @@ class ConcurrentSource: The read is done when all partitions for all streams w ere generated and read. """ - DEFAULT_TIMEOUT_SECONDS = 900 + DEFAULT_TIMEOUT_SECONDS = 900 # Deprecated: default of the ignored `timeout_seconds`. @staticmethod def create( @@ -90,14 +90,13 @@ def __init__( :param slice_logger: The slice logger used to create messages on new slices :param message_repository: The repository to emit messages to :param initial_number_partitions_to_generate: The initial number of concurrent partition generation tasks. Limiting this number ensures will limit the latency of the first records emitted. While the latency is not critical, emitting the records early allows the platform and the destination to process them as early as possible. - :param timeout_seconds: The maximum number of seconds to wait for a record to be read from the queue. If no record is read within this time, the source will stop reading and return. + :param timeout_seconds: Deprecated and ignored. Kept so existing callers do not break. """ self._threadpool = threadpool self._logger = logger self._slice_logger = slice_logger self._message_repository = message_repository self._initial_number_partitions_to_generate = initial_number_partitions_to_generate - self._timeout_seconds = timeout_seconds # We set a maxsize to for the main thread to process record items when the queue size grows. This assumes that there are less # threads generating partitions that than are max number of workers. If it weren't the case, we could have threads only generating diff --git a/airbyte_cdk/sources/streams/concurrent/partition_enqueuer.py b/airbyte_cdk/sources/streams/concurrent/partition_enqueuer.py index a4dd81f29a..fa1d1c5f0f 100644 --- a/airbyte_cdk/sources/streams/concurrent/partition_enqueuer.py +++ b/airbyte_cdk/sources/streams/concurrent/partition_enqueuer.py @@ -38,7 +38,7 @@ def generate_partitions(self, stream: AbstractStream) -> None: When all the partitions are added to the queue, a sentinel is added to the queue to indicate that all the partitions have been generated. If an exception is encountered, the exception will be caught and put in the queue. This is very important because if we don't, the - main thread will have no way to know that something when wrong and will wait until the timeout is reached + main thread will have no way to know that something went wrong and will block on the queue forever. This method is meant to be called in a separate thread. """ diff --git a/airbyte_cdk/sources/streams/concurrent/partition_reader.py b/airbyte_cdk/sources/streams/concurrent/partition_reader.py index 0edc5056a9..a2e7c70999 100644 --- a/airbyte_cdk/sources/streams/concurrent/partition_reader.py +++ b/airbyte_cdk/sources/streams/concurrent/partition_reader.py @@ -66,7 +66,7 @@ def process_partition(self, partition: Partition, cursor: Cursor) -> None: When all the partitions are added to the queue, a sentinel is added to the queue to indicate that all the partitions have been generated. If an exception is encountered, the exception will be caught and put in the queue. This is very important because if we don't, the - main thread will have no way to know that something when wrong and will wait until the timeout is reached + main thread will have no way to know that something went wrong and will block on the queue forever. This method is meant to be called from a thread. :param partition: The partition to read data from