fix: set stderr fd 2 to non-blocking mode to prevent deadlock from pipe backpressure

Co-Authored-By: unknown <>

Author: devin-ai-integration[bot]

airbyte_cdk/entrypoint.py

@@ -3,6 +3,7 @@
 #
 
 import argparse
+import fcntl
 import importlib
 import io
 import ipaddress
@@ -442,18 +443,40 @@ def __getattr__(self, name: str) -> Any:
         return getattr(self._original, name)
 
 
+def _ensure_stderr_nonblock() -> None:
+    """Set stderr fd 2 to non-blocking mode (once).
+
+    When the Airbyte platform stops reading from the source container's
+    stderr pipe, the pipe buffer fills and any ``os.write(2, ...)`` call
+    blocks the calling thread.  If that thread is the main thread, the
+    CDK's record queue fills and all workers deadlock.
+
+    Setting ``O_NONBLOCK`` makes ``os.write(2, ...)`` raise
+    ``BlockingIOError`` (EAGAIN) instead of blocking, which
+    ``_stderr_diag`` already catches.
+    """
+    try:
+        flags = fcntl.fcntl(2, fcntl.F_GETFL)
+        fcntl.fcntl(2, fcntl.F_SETFL, flags | os.O_NONBLOCK)
+    except Exception:
+        # Best-effort; some environments may not support fcntl on fd 2.
+        pass
+
+
 def _stderr_diag(msg: str) -> None:
     """Write a diagnostic message directly to stderr fd.
 
     Uses ``os.write()`` on the raw file descriptor so the write bypasses
     *all* Python buffering (``sys.stderr``, ``PrintBuffer``, logging
-    handlers).  This works even when the stdout/stderr pipes are blocked
-    because ``os.write(2, …)`` is a direct syscall on fd 2.
+    handlers).  fd 2 is set to non-blocking mode so this never stalls
+    the calling thread.
     """
     try:
         os.write(2, f"DIAG: {msg}\n".encode())
     except Exception:
-        pass  # Best-effort; must not prevent caller from continuing.
+        # Best-effort; catches BlockingIOError (EAGAIN) when pipe is
+        # full, plus any other I/O error.
+        pass
 
 
 def _buffered_write_to_stdout(messages: Iterable[str]) -> None:
@@ -490,6 +513,8 @@ def _buffered_write_to_stdout(messages: Iterable[str]) -> None:
             print(message)
         return
 
+    _ensure_stderr_nonblock()
+
     _SENTINEL = None  # signals the writer to stop
     buffer: Queue[Optional[str]] = Queue()
     writer_error: List[Exception] = []

airbyte_cdk/sources/concurrent_source/concurrent_source.py

@@ -3,6 +3,7 @@
 #
 
 import concurrent
+import fcntl
 import logging
 import os
 import sys
@@ -119,6 +120,7 @@ def read(
         # (consumes an item from the queue).  The watchdog reads this to
         # detect when the main thread is stuck.
         self._last_progress_time = time.monotonic()
+        self._ensure_stderr_nonblock()
         self._watchdog_should_run = True
         watchdog = threading.Thread(
             target=self._watchdog_loop,
@@ -162,18 +164,49 @@ def _submit_initial_partition_generators(
             if status_message:
                 yield status_message
 
+    _stderr_nonblock_set = False
+
+    @classmethod
+    def _ensure_stderr_nonblock(cls) -> None:
+        """Set stderr fd 2 to non-blocking mode (once).
+
+        In Airbyte Cloud the platform reads the source container's stdout and
+        stderr pipes.  If the platform pauses reading (e.g. destination
+        backpressure), both pipe buffers fill up.  A blocking ``os.write(2,
+        ...)`` would then stall whichever thread called it — including the
+        main thread, which causes the CDK queue to fill and deadlock all
+        workers.
+
+        Setting ``O_NONBLOCK`` on fd 2 makes ``os.write(2, ...)`` return
+        immediately with ``BlockingIOError`` (EAGAIN) instead of blocking.
+        The ``_diag`` method already catches all exceptions, so the message
+        is simply dropped when the pipe is full.
+        """
+        if cls._stderr_nonblock_set:
+            return
+        try:
+            flags = fcntl.fcntl(2, fcntl.F_GETFL)
+            fcntl.fcntl(2, fcntl.F_SETFL, flags | os.O_NONBLOCK)
+            cls._stderr_nonblock_set = True
+        except Exception:
+            # Best-effort; some environments may not support fcntl on fd 2.
+            pass
+
     @staticmethod
     def _diag(msg: str) -> None:
         """Write diagnostic message directly to stderr fd 2.
 
         Bypasses all Python buffering (sys.stderr, logging, PrintBuffer)
         so the message is visible even when stdout/stderr pipes are blocked.
+        The fd is set to non-blocking mode by ``_ensure_stderr_nonblock``
+        so this call never stalls the calling thread.
         """
         try:
             os.write(2, f"DIAG: {msg}\n".encode())
         except Exception:
             # Intentionally ignored: diagnostics are best-effort and must
-            # never interfere with program execution.
+            # never interfere with program execution.  In non-blocking mode
+            # this catches BlockingIOError (EAGAIN) when the pipe is full.
             pass
 
     def _consume_from_queue(