fix(concurrent-source): surface stalled reads instead of hanging silently

[devin-ai-integration]

Summary

When a worker thread wedges inside Partition.read(), the concurrent read path today produces nothing at all: no log, no trace, no state. ConcurrentSource._consume_from_queue() blocks on an untimed queue.get(), PartitionReader.process_partition() only enqueues on return-or-raise, and the platform's 24h heartbeat_timeout ends up being the only detector. This PR closes the detection gap; it does not attempt to fix whatever wedges the worker.

The consumer now polls instead of blocking forever, and tracks a monotonic "last time anything came off the queue" marker:

while True:
    try:
        item = queue.get(timeout=poll_interval)   # was: queue.get()
    except Empty:
        now = time.monotonic()
        if now - last_progress >= self._timeout_seconds:      # warn, repeat once per interval
            logger.warning("No queue progress for %s seconds. %s", ..., in_flight_description)
        if no_progress_timeout is not None and now - last_progress >= no_progress_timeout:
            threadpool.shutdown()
            raise AirbyteTracedException(..., failure_type=FailureType.system_error)
        continue
    last_progress = time.monotonic()
    ...  # unchanged handling / is_done() break semantics

Two deliberate design choices:

• Warnings are on by default; killing the sync is opt-in. A slow-but-alive stream must not be terminated, so the hard fail is off unless no_progress_timeout_seconds is passed or AIRBYTE_NO_PROGRESS_TIMEOUT_SECONDS is set (non-integer / non-positive values are ignored with a warning). With no configuration, the only behaviour change is a periodic WARNING naming the streams that are generating partitions and the streams with running partitions — enough for a support engineer to see which stream is wedged, well before the 24h heartbeat.
• timeout_seconds is repurposed rather than newly enforced. DEFAULT_TIMEOUT_SECONDS = 900 was plumbed through create()/__init__ but never passed to queue.get() — its docstring promised behaviour that did not exist. It is now the warning interval, and the docstring says so. Enforcing it as a hard 15-minute kill would have been a breaking behaviour change for legitimately slow connectors.

The warning/hard-fail checks are evaluated independently, so a no_progress_timeout_seconds smaller than timeout_seconds still fires on time.

Supporting changes: ConcurrentReadProcessor.get_in_flight_streams_description() (deterministic, sorted) for the diagnostic detail, and a public ThreadPoolManager.shutdown() so the hard-fail path doesn't hang on exit.

Declarative-First Evaluation

Not applicable — this is a CDK core change in the concurrent read path, not a connector change; no declarative/manifest feature can express a watchdog over the shared queue.

Test Coverage

New unit_tests/sources/concurrent_source/test_no_progress_watchdog.py (8 tests in the package pass) using a deterministic fake clock: a stalled queue warns and does not raise when the hard timeout is unset; it raises AirbyteTracedException with FailureType.system_error when the timeout comes from the constructor and when it comes from the env var; normal reads and slow-but-progressing reads complete unchanged and do not trip the watchdog.

Also verified: unit_tests/sources/streams/concurrent (242 passed), ruff check, ruff format, mypy airbyte_cdk/sources/concurrent_source.

Verification not done

I could not reproduce the original stall — it needs a GA4 property with a slice exceeding the 100,000-row runReport page limit, and no workspace/connection/job ID was available on the issue. The claim that this path is silent is verified by code read and by the production stack sample in #1015 (main thread parked in queue.get() → not_empty.wait()); the claim that memory exhaustion is what wedges the worker is the reporter's hypothesis and is unverified here.

Related to https://github.com/airbytehq/oncall/issues/13165:

• https://github.com/airbytehq/oncall/issues/13165

The connector-side trigger is tracked separately in airbytehq/oncall#13164 / airbytehq/airbyte#82766.

Requested by airbyte-support-bot via the /ai-fix automation on the oncall issue.

Link to Devin session: https://app.devin.ai/sessions/a787e277eee34a19827948340f9ad782

[devin-ai-integration]

🤖 Devin AI Engineer

I'll be helping with this pull request! Here's what you should know:

✅ I will automatically:

• Address comments on this PR. Add '(aside)' to your comment to have me ignore it.
• Look at CI failures and help fix them

Note: I can only respond to comments from users who have write access to this repository.

⚙️ Control Options:

• Disable automatic comment, CI, and merge conflict monitoring

[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@devin/1785239104-cdk-no-progress-watchdog#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 devin/1785239104-cdk-no-progress-watchdog

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 138 tests  +5   4 126 ✅ +5   6m 53s ⏱️ -42s
    1 suites ±0      12 💤 ±0 
    1 files   ±0       0 ❌ ±0 

Results for commit 3f601e4. ± Comparison against base commit 7e96546.

[github-actions]

PyTest Results (Full)

4 141 tests  +5   4 129 ✅ +5   11m 54s ⏱️ -29s
    1 suites ±0      12 💤 ±0 
    1 files   ±0       0 ❌ ±0 

Results for commit 3f601e4. ± Comparison against base commit 7e96546.