[ZaneHyattAB]

Overview

👉 TL;DR: Connector output written through the CDK's print buffer could sit inside the process for hours before the platform saw it, because the buffer wrote to the real stdout but never asked the interpreter to push those bytes into the pipe. This made healthy, progressing sources look silent to the platform heartbeat and to anyone reading the logs. Specifically, PrintBuffer.flush() now calls sys.__stdout__.flush() after sys.__stdout__.write().

Related: airbytehq/oncall#12977 (internal, source-stripe). Requested by Zane Hyatt (ZaneHyattAB).

Changes

• PrintBuffer.flush() flushes the underlying sys.__stdout__ TextIOWrapper after writing the batched content.
• New unit_tests/utils/test_print_buffer.py:
  • unit test asserting flush() writes to and flushes sys.__stdout__;
  • real-pipe regression test: a child process using the real CDK logger + PRINT_BUFFER (with PYTHONUNBUFFERED unset) emits a RECORD and a logger.info line, then sleeps; the parent asserts both lines arrive on the pipe while the child is still alive. Without the fix this test times out with zero bytes received.

Review Spotlight

Reviewers with limited time, please review first:

• print_buffer.py#L54-L64: the one-line fix
• test_print_buffer.py#L27-L86: the real-pipe regression test

Why this is a correctness bug, not just a debuggability one

When stdout is a pipe (always, on the platform), sys.__stdout__ is a block-buffered TextIOWrapper with an ~8 KiB pending buffer. PrintBuffer.flush() wrote into that buffer and returned, so bytes only reached the pipe when cumulative output crossed 8 KiB or the process exited. Measured with the real CDK logger and a real pipe: 230 B writes arrive in 8,280 B bursts; a RECORD + two logger.info lines emitted at t+1 s reached the reader only at process exit (t+9 s); with PYTHONUNBUFFERED=1 they arrived at t+1.5 s.

The platform's source heartbeat counts messages it receives. For a low-volume or keepalive-only stream (e.g. one 213 B interim STATE per 600 s state-throttle window) filling 8 KiB takes ~36 messages ≈ 6 h, so with the default heartbeat-max-seconds-between-messages of 3 h a source that is progressing can be failed as silent. In the airbytehq/oncall#12977 investigation, one production job's observed 6h03m gap between two adjacent logger.info calls in _on_stream_is_done is exactly this buffer filling (36 × 600 s), not a stall — and three independent analyses of that job reached wrong conclusions from the log because of it.

Design notes: why flush in PrintBuffer.flush() rather than write_through / PYTHONUNBUFFERED

Show/Hide Content

Options considered:

1. sys.__stdout__.flush() inside PrintBuffer.flush() (chosen). PrintBuffer is already the single choke point for stdout in launch(); its existing 0.1 s interval keeps batching for the print() path, and logging.StreamHandler.emit already calls PrintBuffer.flush() after each log line. Adding the fd flush there makes "left PrintBuffer" mean "reached the pipe", with syscall count still bounded by the interval (≤ 10/s for records) plus one per log line. Smallest, most contained change; nothing outside the entrypoint path is affected.
2. sys.stdout.reconfigure(write_through=True) / line buffering at the entrypoint. Equivalent syscall profile for the PrintBuffer path (each PrintBuffer.flush() already writes one combined string), but it is process-global: it also affects any code that writes to sys.__stdout__ directly, and it must be applied in every entrypoint (launch, source_declarative_manifest, connector builder). Rejected as less contained for no benefit.
3. PYTHONUNBUFFERED=1 in the connector base image. Correct and complementary, but outside this repository (base image / airbyte monorepo) and would not fix connectors pinned to older images. Suggested as a follow-up.

Residual behaviour after this change: messages written via print() within 0.1 s after the last flush stay in PrintBuffer until the next write() or logger call (the existing design). A message written after ≥ 0.1 s of idleness is flushed immediately, so a sparse keepalive stream now reaches the platform on emission.

Throughput and syscall measurements

Benchmark: child process with stdout = real pipe, init_logger("airbyte") + with PRINT_BUFFER:, print() of 1,000,000 ~270 B RECORD lines plus one logger.info per 100k records; parent drains the pipe. Python 3.12, same machine, 3 runs each (child-side records/sec = records / time inside the with block).

mode	records/sec (child), 3 runs	write() syscalls (strace -f -c, 1M records)
before (no fd flush)	309,990 / 287,919 / 322,289	44
after (this PR)	321,574 / 316,277 / 307,660	45

Difference is within run-to-run noise; syscall count is unchanged (+1 per logger line, because those are now pushed through instead of coalescing with the next record batch). Per-line flushing was therefore not needed and the 0.1 s batching is preserved.

Risks

• Each logger.* call now costs one write() syscall (previously it could coalesce into the 8 KiB buffer). With LOG_LEVEL=debug (request/response logging) this is one extra syscall per HTTP request — negligible against the request itself.
• Behaviour on a full pipe is unchanged: write() already blocked there when the 8 KiB buffer spilled; the flush just moves the block to the same call.

Follow-ups

• Set PYTHONUNBUFFERED=1 (or -u) in the Python connector base image so stderr and any direct sys.__stdout__ writers get the same guarantee (outside this repo).
• Separate PR: HTTP connect/read timeouts in HttpClient (the other finding from airbytehq/oncall#12977).

Test plan

• poetry run pytest unit_tests/utils/test_print_buffer.py unit_tests/test_logger.py unit_tests/test_entrypoint.py -q → 61 passed
• Verified the real-pipe test fails on main (times out after 5 s with b'' received) and passes with this change.
• poetry run ruff check / ruff format --check on changed files; poetry run mypy --config-file mypy.ini airbyte_cdk → no issues.

Suggested reviewers

Maxime Carbonneau-Leclerc (@maxi297), Artem Inzhyyants (@artem1205) (recent authors of print_buffer.py / logger.py). Not assigned on purpose; Zane will assign.

Link to Devin session: https://app.devin.ai/sessions/0e315592fd95478b8b4db2e122b927a8
Open in Devin Desktop: https://app.devin.ai/desktop/session/0e315592fd95478b8b4db2e122b927a8?variant=devin

[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/1789662473-print-buffer-flush-stdout#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/1789662473-print-buffer-flush-stdout

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.

[Copilot]

🟡 Changes recommended

The new subprocess regression test embeds an indented python -c script that will raise IndentationError due to leading whitespace, causing the test to fail.

Get a fresh assessment by requesting another Copilot review.

Pull request overview

This PR fixes delayed connector/log output by ensuring PrintBuffer.flush() flushes the underlying sys.__stdout__ stream after writing buffered content, so bytes reach the stdout pipe promptly (important for platform heartbeats and log visibility).

Changes:

• Update PrintBuffer.flush() to call sys.__stdout__.flush() after write().
• Add unit/regression tests validating that flush() both writes and flushes, including a real-pipe subprocess test.

File summaries

File	Description
airbyte_cdk/utils/print_buffer.py	Flushes underlying real stdout after writing buffered output to avoid long delays when stdout is a pipe.
unit_tests/utils/test_print_buffer.py	Adds unit/regression coverage for the new flush-through behavior, including a real pipe subprocess scenario.

Review details

• Files reviewed: 2/2 changed files
• Comments generated: 1
• Review effort level: Lite

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[github-actions]

PyTest Results (Fast)

4 449 tests  +2   4 437 ✅ +2   9m 28s ⏱️ + 1m 49s
    1 suites ±0      12 💤 ±0 
    1 files   ±0       0 ❌ ±0 

Results for commit bae4956. ± Comparison against base commit f77450f.

[github-actions]

PyTest Results (Full)

4 452 tests  +2   4 440 ✅ +2   13m 58s ⏱️ -11s
    1 suites ±0      12 💤 ±0 
    1 files   ±0       0 ❌ ±0 

Results for commit bae4956. ± Comparison against base commit f77450f.

[Daryna Ishchenko (darynaishchenko)]

The fix makes sense: without it, batches sit in the ~8 KiB sys.__stdout__ buffer on a pipe instead of reaching the platform every flush_interval. One side effect for logs is noted inline.