fix(oauth): emit CONNECTOR_CONFIG once per single-use refresh token rotation

[devin-ai-integration]

Summary

Resolves https://github.com/airbytehq/airbyte-internal-issues/issues/17172:

• https://github.com/airbytehq/airbyte-internal-issues/issues/17172

Since #877 (first shipped in v7.6.4), SingleUseRefreshTokenOauth2Authenticator._emit_control_message() both prints the CONNECTOR_CONFIG control message to stdout and emits it to the message repository. Every declarative connector with a refresh_token_updater gets a real ConcurrentMessageRepository from ModelToComponentFactory, and the concurrent read loop prints everything in that repository to stdout too — so each token refresh puts two identical CONNECTOR_CONFIG messages on stdout (observed in Cloud logs for a source-linear sync as two Eagerly persisting source connector config from CONTROL message. lines per refresh).

This PR makes a refresh emit exactly one platform-facing CONNECTOR_CONFIG message:

 def _emit_control_message(self) -> None:
     emit_configuration_as_airbyte_control_message(self._connector_config)   # always: stdout
-    if not isinstance(self._message_repository, NoopMessageRepository):
+    if self._emit_control_message_to_message_repository and not isinstance(
+        self._message_repository, NoopMessageRepository
+    ):
         self._message_repository.emit_message(create_connector_config_control_message(...))

• New kwarg SingleUseRefreshTokenOauth2Authenticator(..., emit_control_message_to_message_repository: bool = False) (appended last, so existing positional callers are unaffected).
• ModelToComponentFactory.create_oauth_authenticator passes emit_control_message_to_message_repository=self._emit_connector_builder_messages.

Why keep the stdout print rather than restore the pre-#877 either/or

The direct stdout print is the delivery the platform relies on, and it is the only one that works everywhere:

• ConcurrentDeclarativeSource has no message_repository property, so AirbyteEntrypoint._emit_queued_messages never drains it; outside read, messages emitted into the ConcurrentMessageRepository land on the concurrent queue and are never printed. That is why check/discover lost the rotated token before fix(oauth): exclude client credentials from body when Authorization header is present and _emit_control_message #877 (the Gong case that motivated it). Restoring the either/or would reintroduce that regression.
• During read, the repository path is only printed once the main thread reaches the message in the queue (behind already-queued records); if the sync aborts first, the rotated single-use token is lost. The worker-thread stdout print is immediate.

The message-repository emission's in-process consumers are the Connector Builder test read (connector_builder/test_reader/reader.py) and the manifest server test_read, both of which consume AirbyteEntrypoint.read in-process and surface latest_config_update; both run the factory with emit_connector_builder_messages=True. It is therefore kept as opt-in and enabled only in that mode. If you construct SingleUseRefreshTokenOauth2Authenticator directly and read the CONNECTOR_CONFIG message off your own MessageRepository, pass emit_control_message_to_message_repository=True.

Hardening the stdout path

Since the direct print is now the sole default delivery, emit_configuration_as_airbyte_control_message writes the payload and newline in a single write() call (print(f"{line}\n", end="", flush=True), the form already used in http_client.py and entrypoint.py). The bare print(x) issued two write() calls and PrintBuffer only locks per call, so a record written from the main thread could split the JSON line emitted from a worker thread. The deprecation note on that function is replaced with an explanation of why token rotation relies on it.

Caveat

The duplicate emission is a genuine CDK defect (it doubles updateSource writes per refresh), but it is not proven to be what breaks source-linear token persistence: another connector with the same duplicate emission persists its rotated token fine (https://github.com/airbytehq/oncall/issues/13432). The platform side (each source CONTROL message also being published twice) is tracked separately in airbyte-platform-internal and is not touched here.

Test coverage

unit_tests/sources/streams/http/requests_native_auth/test_requests_native_auth.py:

• default flag + NoopMessageRepository → exactly one CONNECTOR_CONFIG on stdout
• default flag + InMemoryMessageRepository → exactly one on stdout, none in the repository (fails on main: one CONTROL message remains queued)
• flag True + InMemoryMessageRepository → one on stdout and one in the repository
• flag True + NoopMessageRepository → one on stdout, no error

unit_tests/sources/declarative/test_concurrent_declarative_source.py, with a manifest declaring refresh_token_updater and HttpMocker:

• read yields no CONTROL message and stdout has exactly one CONNECTOR_CONFIG carrying the rotated refresh token (fails on main: a CONTROL message is yielded, i.e. printed a second time by the entrypoint)
• read with emit_connector_builder_messages=True yields exactly one CONTROL message and stdout has exactly one
• check succeeds and stdout has exactly one CONNECTOR_CONFIG

Ran poetry run ruff format ., poetry run ruff check ., poetry run mypy --config-file mypy.ini airbyte_cdk, and poetry run pytest unit_tests/ -x -q (4369 passed, 2 skipped).

Declarative-First Evaluation

Not applicable: the fix is in the CDK itself; source-linear needs no manifest change.

Link to Devin session: https://app.devin.ai/sessions/c578c8af1b6441b78d57e3edc3a2f5b9
Open in Devin Desktop: https://app.devin.ai/desktop/session/c578c8af1b6441b78d57e3edc3a2f5b9?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/1788470527-single-connector-config-emission#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/1788470527-single-connector-config-emission

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.

[Airbyte Support Bot (airbyte-support-bot)]

/ai-review

[Copilot]

🟢 Approval recommended

The behavioral change is narrowly scoped and is backed by targeted unit tests; remaining feedback is limited to minor doc/test maintainability adjustments.

Pull request overview

This PR fixes duplicate CONNECTOR_CONFIG control-message emission during single-use refresh token rotation by making stdout emission unconditional while gating message-repository emission behind an explicit opt-in (used for in-process consumers like Connector Builder).

Changes:

• Add emit_control_message_to_message_repository: bool = False to SingleUseRefreshTokenOauth2Authenticator and only emit to the repository when enabled.
• Wire the new flag from ModelToComponentFactory.create_oauth_authenticator using the existing Connector Builder mode toggle.
• Expand unit test coverage to assert exactly one CONNECTOR_CONFIG on stdout (and optionally one in the repository when enabled), including concurrent declarative source scenarios.

File summaries

File	Description
airbyte_cdk/sources/streams/http/requests_native_auth/oauth.py	Adds the opt-in flag and gates repository emission to prevent duplicate stdout control messages.
airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py	Passes the new flag when creating OAuth authenticators in Connector Builder mode.
unit_tests/sources/streams/http/requests_native_auth/test_requests_native_auth.py	Updates/adds tests covering stdout vs repository emission behavior under different repository/flag combinations.
unit_tests/sources/declarative/test_concurrent_declarative_source.py	Adds concurrent declarative read/check tests ensuring exactly one stdout config update (and correct behavior in builder mode).

Review details

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

---

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

[devin-ai-integration]

👍 On it. Agreed — AirbyteEntrypoint._emit_queued_messages does drain source.message_repository on check/discover for sources that expose one. Qualifying the docstring: repository delivery depends on the source exposing/draining its repository and on when the drain happens, whereas stdout is immediate for every source and command.

[devin-ai-integration]

☑️ Resolved in 1e32a90. Docstring now says stdout is immediate and independent of whether/when the source's repository is drained (noting ConcurrentDeclarativeSource only drains during read), instead of claiming the repository is never drained on check/discover.

---

Devin session

[devin-ai-integration]

👍 On it. Will parse each line once in _connector_config_lines.

[devin-ai-integration]

☑️ Resolved in 1e32a90. _connector_config_lines parses each line once and filters the parsed messages.

---

Devin session

[github-actions]

PyTest Results (Fast)

4 368 tests  +5   4 356 ✅ +5   9m 27s ⏱️ -20s
    1 suites ±0      12 💤 ±0 
    1 files   ±0       0 ❌ ±0 

Results for commit 73c0ae3. ± Comparison against base commit 64bdfc9.

This pull request removes 1 and adds 6 tests. Note that renamed tests count towards both.

unit_tests.sources.streams.http.requests_native_auth.test_requests_native_auth.TestSingleUseRefreshTokenOauth2Authenticator ‑ test_given_message_repository_when_get_access_token_then_emit_message

unit_tests.sources.declarative.test_concurrent_declarative_source ‑ test_given_refresh_token_updater_and_connector_builder_mode_when_read_then_control_message_yielded_once
unit_tests.sources.declarative.test_concurrent_declarative_source ‑ test_given_refresh_token_updater_when_check_then_connector_config_message_on_stdout
unit_tests.sources.declarative.test_concurrent_declarative_source ‑ test_given_refresh_token_updater_when_read_then_exactly_one_connector_config_message_on_stdout
unit_tests.sources.streams.http.requests_native_auth.test_requests_native_auth.TestSingleUseRefreshTokenOauth2Authenticator ‑ test_given_emit_control_message_to_message_repository_when_get_access_token_then_emit_to_stdout_and_repository
unit_tests.sources.streams.http.requests_native_auth.test_requests_native_auth.TestSingleUseRefreshTokenOauth2Authenticator ‑ test_given_message_repository_when_get_access_token_then_emit_exactly_one_control_message_to_stdout
unit_tests.sources.streams.http.requests_native_auth.test_requests_native_auth.TestSingleUseRefreshTokenOauth2Authenticator ‑ test_given_noop_message_repository_and_emit_flag_when_get_access_token_then_only_stdout

♻️ This comment has been updated with latest results.

[github-actions]

PyTest Results (Full)

4 371 tests   4 359 ✅  11m 31s ⏱️
    1 suites     12 💤
    1 files        0 ❌

Results for commit 73c0ae3.

♻️ This comment has been updated with latest results.

[Patrick Nilan (pnilan)]

🤖 This comment was generated by an AI Agent.

The surviving stdout path is not interleave-safe, and it just lost its backup

The reasoning for keeping the stdout print is sound — ConcurrentDeclarativeSource exposes no message_repository property and nothing drains its repository outside read, so restoring the pre-#877 either/or really would reintroduce the check/discover token loss. I verified that independently (hasattr(ConcurrentDeclarativeSource, "message_repository") is False across the whole MRO).

The concern is the form of the print that this now depends on exclusively.

config_observation.py emits with a bare print(x) — default end="\n", which is two write() calls. Measured through the real PrintBuffer:

emit_configuration_as_airbyte_control_message -> 2 write() calls; last='\n'

PrintBuffer.write takes its RLock inside a single write(), so the lock does not span the pair. And the emission runs on a worker thread during a concurrent read — instrumented on a real ConcurrentDeclarativeSource:

emit_configuration_as_airbyte_control_message call sites: [('workerpool_0', False)]

(False = not the main thread.) Meanwhile the main thread writes records through that same buffer, and logger.py points the root log handler at it too. A main-thread write landing between the payload and its newline fuses two JSON objects onto one line, losing both. Stress test against the real PrintBuffer:

GIL switch interval	competing writers	CONTROL lost / 20,000
0.005 (default)	1	0
0.005 (default)	4	1
0.005 (default)	8	0
0.0001	4	77
1e-6	4	2,156

Honest read: order 1 in 10⁴–10⁵ per rotation under record-write contention, with high run-to-run variance. Rare — but the consequence is what this PR changes. Today a mangled print costs the record fused onto that line while the CONNECTOR_CONFIG still arrives via the repository copy on the main thread's single-write path. After this PR the same event loses the record and the rotated single-use token, which means a dead token on the next sync and a manual re-auth.

The CDK has fixed this exact bug class before — 7446e96d "Fix broken print in ccdk (#34578)", whose comment still sits at entrypoint.py:397-399 — and http_client.py:529 already uses the safe form. config_observation.py is the one that never got it.

Recommendation

One line, and worth doing regardless of this PR since it also stops the record loss that exists on main today:

# airbyte_cdk/config_observation.py
- print(orjson.dumps(AirbyteMessageSerializer.dump(airbyte_message)).decode())
+ print(f"{orjson.dumps(AirbyteMessageSerializer.dump(airbyte_message)).decode()}\n", end="", flush=True)

It just becomes load-bearing once the redundant copy is gone.

Worth noting too that emit_configuration_as_airbyte_control_message is marked WARNING: deprecated ... in favor of the MessageRepository mechanism in its own docstring, and this PR makes that deprecated, unhardened function the sole default delivery. Either un-deprecating it for this use or leaving a note there explaining why the OAuth rotation path keeps it would save the next reader the round trip.

Note that no test in this PR can catch this: PrintBuffer.__enter__ skips stdout substitution under pytest, and the new tests call source.read(...) with capsys rather than going through launch().

[devin-ai-integration]

👍 On it. Re the interleave-safety comment: agreed the bare print(x) is two write() calls and PrintBuffer only locks per call, so this PR makes that path the sole delivery. Switching config_observation.py to the single-write form (print(f"{...}\n", end="", flush=True), matching http_client.py) and replacing the "deprecated" docstring note with an explanation of why token rotation keeps using it. Not adding a launch()-level test since PrintBuffer skips stdout substitution under pytest, as you note.

---

Devin session

[devin-ai-integration]

☑️ Resolved in 73c0ae3. emit_configuration_as_airbyte_control_message now writes payload+newline in one write() with flush=True, and its docstring explains why token rotation keeps this direct stdout path instead of the deprecation warning.

---

Devin session

[Patrick Nilan (pnilan)]

/prerelease

Prerelease Job Info

This job triggers the publish workflow with default arguments to create a prerelease.

Prerelease job started... Check job output.

✅ Prerelease workflow triggered successfully.

View the publish workflow run: https://github.com/airbytehq/airbyte-python-cdk/actions/runs/33815635514