docs(concurrent-cdk): mark timeout_seconds as deprecated and ignored

[Baz (bazarnov)]

Overview

👉 TL;DR: Corrects the docs for an old option that promised a sync would stop by itself when no data arrived for a while. The option has done nothing since 2024; it stays only so existing code that passes it keeps working. Specifically, ConcurrentSource.__init__ now documents timeout_seconds as deprecated and ignored, and the never-read self._timeout_seconds assignment is removed. timeout_seconds on create() and __init__() and DEFAULT_TIMEOUT_SECONDS stay in the public signature. Two worker docstrings that said the main thread waits for a timeout now say it blocks on the queue forever.

Pairs with #1168:

• feat(low-code): opt-in TCP keepalive and request timeouts on HttpRequester #1168

Changes

• concurrent_source.py: the timeout_seconds docstring says "Deprecated and ignored" instead of promising the read stops when no record arrives from the queue within it. _consume_from_queue blocks on a bare queue.get(), so that promise has been false since Fix concurrent deadlock airbyte#34454.
• concurrent_source.py: drop self._timeout_seconds = timeout_seconds, the attribute's only occurrence in the repository.
• concurrent_source.py: DEFAULT_TIMEOUT_SECONDS gets a trailing comment saying it is only the default of the ignored timeout_seconds.
• partition_enqueuer.py, partition_reader.py: the docstrings no longer say the main thread waits for a timeout when a worker exception never reaches the queue; it blocks on the queue forever. Both lines were added by Fix concurrent deadlock airbyte#34454, the change that removed that timeout. Also fixes "when wrong" to "went wrong".

Review Spotlight

Reviewers with limited time, please review first:

• concurrent_source.py#L85-L99: the kept parameter, the new docstring line and the constructor body without the assignment
• concurrent_source.py#L145-L150: the bare queue.get() the old docstring contradicted

Risks

• Behaviour change: none at runtime. Nothing has read timeout_seconds since Fix concurrent deadlock airbyte#34454 removed ThrottledQueue, whose get() applied the timeout. The signature is unchanged: timeout_seconds: int = DEFAULT_TIMEOUT_SECONDS on create() and __init__(), DEFAULT_TIMEOUT_SECONDS = 900.
• Who sees it: only code that reads the private _timeout_seconds attribute, which now raises AttributeError. Readers found: 0.
• How it was counted:
  • CDK main: git grep -n _timeout_seconds returns only the removed line. git grep -n -w timeout_seconds finds no caller that passes it (the other hit is the unrelated ConnectorTestScenario.timeout_seconds field); ConcurrentDeclarativeSource and FileBasedSource call ConcurrentSource.create() without it.
  • airbyte master: git grep -n "\._timeout_seconds" origin/master matches only source-google-ads's own RunAsThread._timeout_seconds, unrelated. git grep -n -E "ConcurrentSource\.create\(|ConcurrentSource\(" origin/master -- airbyte-integrations/connectors finds 1 direct caller, source-salesforce, which does not pass timeout_seconds.

Test plan

No new test: nothing changes at runtime.

• python -m pytest unit_tests/sources/concurrent_source unit_tests/sources/streams/concurrent unit_tests/sources/test_source_read.py unit_tests/sources/declarative/async_job/test_integration.py -q -p no:cacheprovider
  • this branch: 272 passed
  • main: 272 passed
• Broader callers on this branch: unit_tests/sources/declarative/test_concurrent_declarative_source.py 86 passed (3 test_read_with_concurrent_and_synchronous_streams* tests deselected, they can crash on the sqlite cache race in Cached parent stream with more than one partition corrupts the shared sqlite request cache under concurrent reads (no such table: responses, file is not a database, no more rows available) #1150); unit_tests/sources/file_based/test_file_based_scenarios.py 253 passed.
• inspect.signature on both refs: timeout_seconds: int = 900 on ConcurrentSource.create and ConcurrentSource.__init__, and DEFAULT_TIMEOUT_SECONDS == 900. hasattr(source, "_timeout_seconds") on a source built with ConcurrentSource.create(..., timeout_seconds=1) is True on main and False here.
• ruff format --check, ruff check and mypy --config-file mypy.ini on concurrent_source.py, partition_enqueuer.py and partition_reader.py: clean.

Design notes: why no queue watchdog or page-token guard

Show/Hide Content

timeout_seconds is not wired back up as a queue timeout:

• It used to be one. Fix concurrent deadlock airbyte#34454 removed the 900 s queue.get timeout it fed, after sources died on queue.Empty.
• A queue timeout cannot stop a worker blocked in a socket read. ThreadPoolExecutor workers are joined at interpreter exit, so after the main thread raises, the process stays alive until the platform heartbeat ends the attempt. Checked with a worker blocked on a socket recv with no timeout: the process was still alive 6 s after the main thread raised.
• Async-job streams legitimately put nothing on the queue for long periods. AsyncJobOrchestrator polls with time.sleep and logs outside the queue, for up to the 60 min default polling_job_timeout (24 h for source-salesforce Bulk). A queue timeout would fail those healthy syncs.

The stall #1148 reports is an HTTP response that never finishes. #1168 bounds it at the request with default connect and read timeouts (layer 1).

A repeated-page-token guard (layer 3) is intentionally not added. Scroll APIs can return the same token on every page: Intercom does, and a custom paginator that stopped on a repeated token truncated its Companies stream to about 200 records (airbytehq/airbyte#70335). A cursor that never advances is ended with the manifest's stop_condition, which sees the raw response.

Part of #1148

[github-actions]

👋 Greetings, Airbyte Team Member!

Here are some helpful tips and reminders for your convenience.

💡 Show Tips and Tricks

Testing This CDK Version

You can test this version of the CDK using the following:

# Run the CLI from this branch:
uvx 'git+https://github.com/airbytehq/airbyte-python-cdk.git@bazarnov/docs-concurrent-source-timeout-seconds#egg=airbyte-python-cdk[dev]' --help

# Update a connector to use the CDK from this branch ref:
cd airbyte-integrations/connectors/source-example
poe use-cdk-branch bazarnov/docs-concurrent-source-timeout-seconds

PR Slash Commands

Airbyte Maintainers can execute the following slash commands on your PR:

• /autofix - Fixes most formatting and linting issues
• /poetry-lock - Updates poetry.lock file
• /test - Runs connector tests with the updated CDK
• /prerelease - Triggers a prerelease publish with default arguments
• /poe build - Regenerate git-committed build artifacts, such as the pydantic models which are generated from the manifest JSON schema in YAML.
• /poe <command> - Runs any poe command in the CDK environment

📚 Show Repo Guidance

Helpful Resources

• CDK API Reference

📝 Edit this welcome message.

[github-actions]

PyTest Results (Fast)

4 982 tests  ±0   4 970 ✅ ±0   7m 24s ⏱️ - 2m 9s
    1 suites ±0      12 💤 ±0 
    1 files   ±0       0 ❌ ±0 

Results for commit c9782d8. ± Comparison against base commit d214b0c.

[github-actions]

PyTest Results (Full)

4 985 tests  ±0   4 973 ✅ ±0   16m 8s ⏱️ +4s
    1 suites ±0      12 💤 ±0 
    1 files   ±0       0 ❌ ±0 

Results for commit c9782d8. ± Comparison against base commit d214b0c.