feat: make record queue unlimited (maxsize=0) to prevent deadlock

Co-Authored-By: unknown <>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/concurrent_source/concurrent_source.py

@@ -99,7 +99,7 @@ def __init__(
         # threads generating partitions that than are max number of workers. If it weren't the case, we could have threads only generating
         # partitions which would fill the queue. This number is arbitrarily set to 10_000 but will probably need to be changed given more
         # information and might even need to be configurable depending on the source
-        self._queue = queue or Queue(maxsize=10_000)
+        self._queue = queue or Queue(maxsize=0)
 
     def read(
         self,

airbyte_cdk/sources/concurrent_source/thread_pool_manager.py

@@ -12,7 +12,7 @@ class ThreadPoolManager:
     Wrapper to abstract away the threadpool and the logic to wait for pending tasks to be completed.
     """
 
-    DEFAULT_MAX_QUEUE_SIZE = 50_000
+    DEFAULT_MAX_QUEUE_SIZE = 0  # 0 means unlimited
 
     def __init__(
         self,

airbyte_cdk/sources/declarative/concurrent_declarative_source.py

@@ -157,7 +157,7 @@ def __init__(
         # threads generating partitions that than are max number of workers. If it weren't the case, we could have threads only generating
         # partitions which would fill the queue. This number is arbitrarily set to 10_000 but will probably need to be changed given more
         # information and might even need to be configurable depending on the source
-        queue: Queue[QueueItem] = Queue(maxsize=10_000)
+        queue: Queue[QueueItem] = Queue(maxsize=0)
         message_repository = InMemoryMessageRepository(
             Level.DEBUG if emit_connector_builder_messages else Level.INFO
         )