fix: fail fast on non-JSON-serializable types in serialization fallback (AI-Triage PR)

[devin-ai-integration]

Summary

Hardens airbyte_message_to_string() to raise a clear AirbyteTracedException when record data contains types that neither orjson nor stdlib json can serialize, instead of letting an unhandled TypeError silently kill worker threads and deadlock the concurrent source pipeline.

Problem: The existing fallback chain is orjson.dumps() → json.dumps(). But json.dumps() can also raise TypeError for types like Python complex numbers. When this unhandled exception occurs in a worker thread of the concurrent source pipeline, the main thread deadlocks waiting on queue.get() because the worker silently dies.

Fix: Wrap the json.dumps() fallback in its own try/except and raise an AirbyteTracedException with failure_type=system_error and a clear message identifying the serialization failure. This stops the sync immediately with an actionable error rather than deadlocking or silently corrupting data.

Resolves https://github.com/airbytehq/oncall/issues/11654:

• https://github.com/airbytehq/oncall/issues/11654

Related: airbytehq/airbyte#74883

Review & Testing Checklist for Human

• Exception propagation in concurrent pipeline: Verify that AirbyteTracedException raised from airbyte_message_to_string() propagates correctly through the concurrent read pipeline and does not itself cause a deadlock. The function is called via map() in run() — confirm the exception surfaces cleanly to the platform.
• system_error is the correct failure type: This will cause the platform to retry the sync. Confirm this is preferable over config_error (which would not retry) for non-serializable data types in records.
• Verify with the companion connector-level fix: The specific complex type issue in source-google-search-console is handled at the connector level in airbytehq/airbyte#75426. This CDK change is defense-in-depth — test that the connector fix prevents the exception from firing, and that without the connector fix, this CDK change produces a clear error instead of a deadlock.

Suggested test plan: Run source-google-search-console with the search_analytics_by_query stream against a real account. With the companion connector fix, the sync should complete normally. Without it, the sync should fail with a clear AirbyteTracedException message mentioning JSON serialization, rather than deadlocking.

Notes

• The root cause of complex values entering the data flow was separately fixed in CDK PR airbytehq/airbyte-python-cdk#579 (Jinja interpolation rejecting complex types, merged in v6.54.5). This PR hardens the serialization layer as defense-in-depth against any future non-serializable type reaching this code path.
• Prior art: The original orjson → json fallback was added in airbytehq/airbyte-python-cdk#210 (v6.16.1, by Artem Inzhyyants (@artem1205) / Maxime Carbonneau-Leclerc (@maxi297)).
• Companion connector-level fix: airbytehq/airbyte#75426 adds SanitizeNumericFields transformation to extract .real from complex values before serialization.

Updates since last revision

• Replaced default=str silent fallback with AirbyteTracedException per reviewer feedback from suisuixia42 — fail fast with a clear error instead of silently corrupting numeric fields into strings like "(0.0423+0j)".
• Updated test to assert AirbyteTracedException is raised with the expected message.

Link to Devin session: https://app.devin.ai/sessions/d8317c1f4ce64f70b5e807425b72aca2

Important

Auto-merge enabled.

This PR is set to merge automatically when all requirements are met.

Note

Auto-merge may have been disabled. Please check the PR status to confirm.

[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 and CI 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/1773523110-fix-serialization-fallback-complex-types#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/1773523110-fix-serialization-fallback-complex-types

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.

[github-actions]

PyTest Results (Fast)

3 990 tests  +1   3 979 ✅ +1   7m 26s ⏱️ +19s
    1 suites ±0      11 💤 ±0 
    1 files   ±0       0 ❌ ±0 

Results for commit ab099da. ± Comparison against base commit 0b94cbe.

♻️ This comment has been updated with latest results.

[github-actions]

PyTest Results (Full)

3 993 tests  +1   3 981 ✅ +1   10m 43s ⏱️ -30s
    1 suites ±0      12 💤 ±0 
    1 files   ±0       0 ❌ ±0 

Results for commit ab099da. ± Comparison against base commit 0b94cbe.

♻️ This comment has been updated with latest results.

[Copilot]

Pull request overview

This PR hardens AirbyteEntrypoint.airbyte_message_to_string() against record payloads containing Python types that cannot be serialized by either orjson or stdlib json, to avoid worker-thread crashes that can lead to deadlocks in concurrent pipelines.

Changes:

• Add a second serialization fallback: if json.dumps() fails, retry with json.dumps(..., default=str).
• Add a unit test covering a complex value that triggers the new fallback.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.

File	Description
airbyte_cdk/entrypoint.py	Adds a nested JSON serialization fallback using default=str when both orjson and stdlib json fail.
unit_tests/test_entrypoint.py	Adds coverage for the new fallback path using a complex value in record data.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Patrick Nilan (pnilan)]

Suggestion: Pair this with a connector-level fix in source-google-search-console

This CDK-level fix prevents the deadlock, which is good — but the default=str fallback will silently turn numeric values like ctr and position into string representations (e.g., "(0+0j)") in the destination. That's data corruption that's hard to detect.

I'd recommend pairing this with a defensive transformation in the connector itself (source-google-search-console/components.py) that sanitizes numeric fields before they ever reach serialization:

@dataclass
class SanitizeNumericFields(RecordTransformation):
    """
    Ensures numeric metric fields are JSON-serializable by guarding against non-standard
    Python numeric types (e.g., complex) that can cause serialization failures and deadlocks
    in the concurrent read pipeline.

    See: https://github.com/airbytehq/airbyte/issues/74883
    """

    NUMERIC_FIELDS = ("clicks", "impressions", "ctr", "position")

    def transform(
        self,
        record: Dict[str, Any],
        config: Optional[Config] = None,
        stream_state: Optional[StreamState] = None,
        stream_slice: Optional[StreamSlice] = None,
    ) -> None:
        for field in self.NUMERIC_FIELDS:
            value = record.get(field)
            if value is not None and isinstance(value, complex):
                record[field] = value.real

Then wire it into all search analytics streams in manifest.yaml as a transformation:

- type: CustomTransformation
  class_name: source_declarative_manifest.components.SanitizeNumericFields

Why both fixes?

• The CDK fix (this PR) prevents the deadlock for all connectors — good as defense-in-depth
• The connector fix preserves correct numeric values (value.real returns the actual float) instead of silently writing "(0+0j)" strings to the destination
• The connector already has a precedent for this pattern — search_analytics_by_query casts query to string for the same reason (manifest.yaml lines 717-728)

[devin-ai-integration]

Great suggestion — you're right that the default=str fallback preserves the sync but degrades data quality (e.g. "(0+0j)" instead of 0.0). A connector-level SanitizeNumericFields transformation using value.real would preserve correct numeric values.

I looked at the connector and can confirm:

• components.py already imports RecordTransformation and has the pattern ready to use
• manifest.yaml lines 717-728 show the existing precedent (casting query to string for the same serialization reason)
• The ctr and position fields in the schema (lines 163-165) are typed as number with multipleOf: 1e-25, so string coercion would violate the schema contract

Want me to create a companion Draft PR in airbytehq/airbyte for the connector-level fix? I can add the SanitizeNumericFields transformation and wire it into the search analytics streams in the manifest.

[Patrick Nilan (pnilan)]

Updated suggestion: SanitizeNumericFields with warning logging

Building on my previous comment — the connector-level transformation should also log a warning when a complex value is encountered, so the data issue is visible rather than silently discarded:

@dataclass
class SanitizeNumericFields(RecordTransformation):
    """
    Ensures numeric metric fields are JSON-serializable by guarding against non-standard
    Python numeric types (e.g., complex) that can cause serialization failures and deadlocks
    in the concurrent read pipeline.

    See: https://github.com/airbytehq/airbyte/issues/74883
    """

    NUMERIC_FIELDS = ("clicks", "impressions", "ctr", "position")

    def transform(
        self,
        record: Dict[str, Any],
        config: Optional[Config] = None,
        stream_state: Optional[StreamState] = None,
        stream_slice: Optional[StreamSlice] = None,
    ) -> None:
        for field in self.NUMERIC_FIELDS:
            value = record.get(field)
            if value is not None and isinstance(value, complex):
                logger.warning(
                    f"Complex value encountered for field '{field}': {value}. "
                    f"Using real component only."
                )
                record[field] = value.real

Why this is preferable to the default=str approach in this PR:

• Preserves numeric types: .real returns a float (e.g., complex(0.0423, 0).real → 0.0423), so downstream numeric processing in the destination continues to work. default=str produces "(0.0423+0j)" — a string that breaks any numeric column.
• Visible: The warning makes it clear when this edge case fires, rather than silently coercing data.
• Scoped: Only affects the fields that could plausibly have this issue, rather than broadly stringifying any non-serializable type across all connectors.

[devin-ai-integration]

Thanks for the refined suggestion with the warning logging — that's a good addition. I'll create a companion Draft PR in airbytehq/airbyte now with the SanitizeNumericFields transformation. Will post the link here once it's up.

[devin-ai-integration]

Created the companion connector-level Draft PR per Patrick Nilan (@pnilan)'s suggestion: airbytehq/airbyte#75426

This adds a SanitizeNumericFields transformation that extracts .real from complex values to preserve correct numeric data, wired into all 13 search analytics streams.

[Patrick Nilan (pnilan)]

Approvingh

[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/1773523110-fix-serialization-fallback-complex-types#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/1773523110-fix-serialization-fallback-complex-types

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.