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