fix(low-code): classify text-decode failures in CompositeRawDecoder parsers

[devin-ai-integration]

Summary

When an endpoint returns bytes that aren't text in the parser's configured encoding (most commonly a still-gzipped body reaching a text parser), the declarative decoder parsers let the raw UnicodeDecodeError escape. The user got a Python traceback — UnicodeDecodeError: 'utf-8' codec can't decode byte 0x8b in position 1 — with no AirbyteTracedException, no user-facing message, and no FailureType.

This adds one shared wrapper, modelled on the existing handling in zipfile_decoder.py, and calls it from every place in composite_raw_decoder.py that turns bytes into text:

def _raise_decode_error(exc: UnicodeDecodeError, encoding: str) -> NoReturn:
    internal_message = (
        f"Response body failed to decode using the configured encoding {encoding!r}: {exc}. "
        f"Leading bytes of the chunk that failed to decode (hex): {exc.object[:4].hex(' ')}."
        f"{format_guess}"  # " Format guess: gzip." for 1f 8b, " Format guess: zip." for PK
    )
    logger.error(internal_message)
    raise AirbyteTracedException(
        message="Response body cannot be decoded as text using the decoder's configured encoding.",
        internal_message=internal_message,
        failure_type=FailureType.system_error,
    ) from exc

Call sites: CsvParser.parse (the error surfaces while iterating csv.DictReader over the TextIOWrapper, so the iteration is what's guarded), JsonLineParser.parse (per line), _Utf8Recoder.read (the JsonItemsParser non-UTF-8 path), and JsonParser.parse.

JsonParser also needed a small restructure: it previously decoded inside both _parse_orjson and _parse_json, where a decode failure was swallowed into Optional[Any] = None and collapsed into the generic "Response JSON data failed to be parsed" system_error. It now decodes once up front and passes str to those two private helpers:

-        body_json = self._parse_orjson(raw_data) or self._parse_json(raw_data)
+        try:
+            decoded_data = raw_data.decode(self.encoding)
+        except UnicodeDecodeError as exc:
+            _raise_decode_error(exc, self.encoding)
+        body_json = self._parse_orjson(decoded_data) or self._parse_json(decoded_data)

The orjson-then-json fallback and the existing generic exception for genuinely malformed JSON are unchanged.

Why system_error and not config_error

config_error was the triage suggestion, on the grounds that the decoder configuration doesn't match what the endpoint returns and won't fix itself on retry. I went with system_error instead: the mismatch is between the connector's manifest and the payload, not between the user's config (credentials, parameters) and the payload. For any released connector the user cannot change the decoder, so config_error would attribute the failure to them and imply an action they can't take. This also matches how the neighbouring failure modes in this package are already classified (zipfile_decoder.py, and JsonParser's existing parse failure). Happy to flip it if maintainers read the ownership boundary differently.

Behaviour notes

• No behaviour change for records that decode successfully.
• JsonLineParser's json.JSONDecodeError handling is untouched: a validly-decodable but malformed line still logs a warning and is skipped. Only the previously-uncaught UnicodeDecodeError path changed, and it still fails the sync — just with a classified error. Nothing new is silently dropped; there is a new test pinning the skip behaviour.
• The logged hex prefix is capped at 4 bytes (enough for the gzip/zip magic numbers) so record data from a text payload can't land in the logs.
• message deliberately omits the encoding value, byte offsets, and codec wording so it stays deterministic and usable as a log aggregation key; all of that detail lives in internal_message.

Declarative-First Evaluation

This change is in the CDK's parser layer, not in a connector manifest, so no declarative feature can deliver it. RecordFilter, AddFields/RemoveFields, transformations, DatetimeBasedCursor, DefaultPaginator, SubstreamPartitionRouter, and $ref overrides all operate on records that have already been decoded and parsed — they never run when the byte-to-text step itself raises. HttpRequester error handlers classify HTTP-level responses (status codes, response bodies) and do not see exceptions raised inside a decoder's parser. No custom Python component is added to any connector: the fix is entirely inside the shared CDK decoder that every declarative connector already uses, so connectors get the improved classification with no manifest change at all.

Overlap with an in-flight PR

#1124 rewrites GzipParser.parse in this same file to auto-detect whether the payload is actually gzipped — that's the root-cause fix for the originating incident, this one is the error-message hardening. I deliberately left GzipParser alone, but both PRs insert a module-level helper just below logger = ..., so whichever lands second will likely need a trivial rebase in that one spot.

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

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

Requested via the /ai-fix workflow from AI Triage on that issue.

Test plan

New tests in unit_tests/sources/declarative/decoders/test_composite_decoder.py: a gzip payload fed directly to CsvParser, JsonParser, JsonLineParser, and JsonItemsParser asserts AirbyteTracedException with failure_type == FailureType.system_error, the exact message, absence of codec / UnicodeDecodeError / 0x8b / utf-8 from message, and presence of the raw decode error plus the 1f 8b signature in internal_message. Plus a test pinning JsonLineParser's malformed-line skip. All four decode tests fail on main (raw UnicodeDecodeError for CSV / JSON Lines / JSON Items, generic parse error for JSON).

• poetry run pytest unit_tests/sources/declarative/decoders/ -q — 66 passed
• poetry run ruff format --check . && poetry run ruff check . — passed
• poetry run mypy --config-file mypy.ini airbyte_cdk — passed

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

[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/1787279627-decode-error-wrapping#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/1787279627-decode-error-wrapping

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]

Pull request overview

This pull request hardens the CDK’s declarative decoder parsers by consistently classifying text decoding failures (e.g., gzipped bytes reaching a text parser) into an AirbyteTracedException with a deterministic user-facing message and FailureType.system_error, instead of letting raw UnicodeDecodeError escape or being collapsed into a generic JSON parse failure.

Changes:

• Added a shared _raise_decode_error(...) helper in composite_raw_decoder.py and used it in all byte-to-text decode sites across CSV/JSON/JSONL/JSON-items parsing.
• Refactored JsonParser to decode once up-front, ensuring decode errors are classified distinctly from JSON parse errors.
• Added unit tests covering decode-failure classification for multiple parsers and pinned JsonLineParser’s malformed-line skip behavior.

Reviewed changes

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

File	Description
airbyte_cdk/sources/declarative/decoders/composite_raw_decoder.py	Introduces a shared decode-error wrapper and applies it across all parsers that convert bytes to text; refactors JsonParser decode flow for correct error classification.
unit_tests/sources/declarative/decoders/test_composite_decoder.py	Adds regression tests ensuring decode failures raise AirbyteTracedException with FailureType.system_error and verifies JSONL malformed-line skipping remains unchanged.

---

💡 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 343 tests  +5   4 332 ✅ +5   10m 48s ⏱️ + 2m 16s
    1 suites ±0      11 💤 ±0 
    1 files   ±0       0 ❌ ±0 

Results for commit b034f4d. ± Comparison against base commit 0655f52.

[github-actions]

PyTest Results (Full)

4 346 tests  +5   4 334 ✅ +5   13m 47s ⏱️ +14s
    1 suites ±0      12 💤 ±0 
    1 files   ±0       0 ❌ ±0 

Results for commit b034f4d. ± Comparison against base commit 0655f52.