feat: add diagnostic logging for deadlock debugging

- Add heartbeat logging to _consume_from_queue (logs every 60s when no items received)
- Add STARTED/PROGRESS/COMPLETED/FAILED logging to PartitionReader.process_partition
- Add STARTED/COMPLETED/FAILED logging to PartitionEnqueuer.generate_partitions
- Use queue.get(timeout=60) instead of blocking queue.get() to enable heartbeat

Co-Authored-By: unknown <>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/concurrent_source/concurrent_source.py

@@ -4,7 +4,9 @@
 
 import concurrent
 import logging
-from queue import Queue
+import threading
+import time
+from queue import Empty, Queue
 from typing import Iterable, Iterator, List, Optional
 
 from airbyte_cdk.models import AirbyteMessage
@@ -143,7 +145,42 @@ def _consume_from_queue(
         queue: Queue[QueueItem],
         concurrent_stream_processor: ConcurrentReadProcessor,
     ) -> Iterable[AirbyteMessage]:
-        while airbyte_message_or_record_or_exception := queue.get():
+        last_item_time = time.monotonic()
+        heartbeat_interval = 60.0  # Log heartbeat every 60 seconds
+        items_since_last_heartbeat = 0
+
+        while True:
+            try:
+                airbyte_message_or_record_or_exception = queue.get(timeout=heartbeat_interval)
+            except Empty:
+                elapsed = time.monotonic() - last_item_time
+                self._logger.info(
+                    "Queue heartbeat: no items received for %.0fs. "
+                    "queue_size=%d, threadpool_done=%s, active_threads=%d",
+                    elapsed,
+                    queue.qsize(),
+                    self._threadpool.is_done(),
+                    threading.active_count(),
+                )
+                continue
+
+            if not airbyte_message_or_record_or_exception:
+                break
+
+            now = time.monotonic()
+            items_since_last_heartbeat += 1
+            if now - last_item_time >= heartbeat_interval:
+                self._logger.info(
+                    "Queue heartbeat: processed %d items in last %.0fs. "
+                    "queue_size=%d, item_type=%s",
+                    items_since_last_heartbeat,
+                    now - last_item_time,
+                    queue.qsize(),
+                    type(airbyte_message_or_record_or_exception).__name__,
+                )
+                items_since_last_heartbeat = 0
+            last_item_time = now
+
             yield from self._handle_item(
                 airbyte_message_or_record_or_exception,
                 concurrent_stream_processor,

airbyte_cdk/sources/streams/concurrent/partition_enqueuer.py

@@ -1,6 +1,7 @@
 #
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
+import logging
 import time
 from queue import Queue
 
@@ -33,17 +34,30 @@ def __init__(
         self._sleep_time_in_seconds = sleep_time_in_seconds
 
     def generate_partitions(self, stream: AbstractStream) -> None:
-        """
-        Generate partitions from a partition generator and put them in a queue.
-        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.
+        """Generate partitions from a partition generator and put them in a queue.
+
+        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
+        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
+        went wrong and will wait until the timeout is reached.
 
         This method is meant to be called in a separate thread.
         """
+        logger = logging.getLogger(f"airbyte.partition_enqueuer.{stream.name}")
+        logger.info("Partition generation STARTED for stream=%s", stream.name)
+        partition_count = 0
+        start_time = time.monotonic()
         try:
             for partition in stream.generate_partitions():
+                partition_count += 1
+                logger.info(
+                    "Partition generation: enqueuing partition #%d for stream=%s, slice=%s",
+                    partition_count,
+                    stream.name,
+                    partition.to_slice(),
+                )
                 # Adding partitions to the queue generates futures. To avoid having too many futures, we throttle here. We understand that
                 # we might add more futures than the limit by throttling in the threads while it is the main thread that actual adds the
                 # future but we expect the delta between the max futures length and the actual to be small enough that it would not be an
@@ -58,7 +72,22 @@ def generate_partitions(self, stream: AbstractStream) -> None:
                 while self._thread_pool_manager.prune_to_validate_has_reached_futures_limit():
                     time.sleep(self._sleep_time_in_seconds)
                 self._queue.put(partition)
+            elapsed = time.monotonic() - start_time
+            logger.info(
+                "Partition generation COMPLETED for stream=%s: %d partitions in %.1fs",
+                stream.name,
+                partition_count,
+                elapsed,
+            )
             self._queue.put(PartitionGenerationCompletedSentinel(stream))
         except Exception as e:
+            elapsed = time.monotonic() - start_time
+            logger.info(
+                "Partition generation FAILED for stream=%s after %.1fs with %d partitions: %s",
+                stream.name,
+                elapsed,
+                partition_count,
+                str(e)[:200],
+            )
             self._queue.put(StreamThreadException(e, stream.name))
             self._queue.put(PartitionGenerationCompletedSentinel(stream))

airbyte_cdk/sources/streams/concurrent/partition_reader.py

@@ -1,6 +1,7 @@
 # Copyright (c) 2025 Airbyte, Inc., all rights reserved.
 
 import logging
+import time
 from queue import Queue
 from typing import Optional
 
@@ -72,15 +73,53 @@ def process_partition(self, partition: Partition, cursor: Cursor) -> None:
         :param partition: The partition to read data from
         :return: None
         """
+        partition_start = time.monotonic()
+        stream_name = partition.stream_name()
+        slice_info = partition.to_slice()
+        logger = logging.getLogger(f"airbyte.partition_reader.{stream_name}")
+        logger.info(
+            "Partition read STARTED for stream=%s, slice=%s",
+            stream_name,
+            slice_info,
+        )
         try:
             if self._partition_logger:
                 self._partition_logger.log(partition)
 
+            record_count = 0
+            last_progress_time = partition_start
             for record in partition.read():
                 self._queue.put(record)
                 cursor.observe(record)
+                record_count += 1
+                now = time.monotonic()
+                if now - last_progress_time >= 30.0:
+                    logger.info(
+                        "Partition read PROGRESS for stream=%s: %d records read so far (%.0fs elapsed), slice=%s",
+                        stream_name,
+                        record_count,
+                        now - partition_start,
+                        slice_info,
+                    )
+                    last_progress_time = now
             cursor.close_partition(partition)
+            elapsed = time.monotonic() - partition_start
+            logger.info(
+                "Partition read COMPLETED for stream=%s: %d records in %.1fs, slice=%s",
+                stream_name,
+                record_count,
+                elapsed,
+                slice_info,
+            )
             self._queue.put(PartitionCompleteSentinel(partition, self._IS_SUCCESSFUL))
         except Exception as e:
-            self._queue.put(StreamThreadException(e, partition.stream_name()))
+            elapsed = time.monotonic() - partition_start
+            logger.info(
+                "Partition read FAILED for stream=%s after %.1fs: %s, slice=%s",
+                stream_name,
+                elapsed,
+                str(e)[:200],
+                slice_info,
+            )
+            self._queue.put(StreamThreadException(e, stream_name))
             self._queue.put(PartitionCompleteSentinel(partition, not self._IS_SUCCESSFUL))