Skip to content

fix: support binary Kafka headers without decoding errors - #2459

Open
Bazarovinc wants to merge 16 commits into
ag2ai:mainfrom
Bazarovinc:fix/non-utf8-kafka-message-headers
Open

Bazarovinc wants to merge 16 commits into
ag2ai:mainfrom
Bazarovinc:fix/non-utf8-kafka-message-headers

Conversation

@Bazarovinc

@Bazarovinc Bazarovinc commented Aug 15, 2025 •

Copy link
Copy Markdown
Contributor

Summary

Kafka headers containing arbitrary bytes raise UnicodeDecodeError during single-message and batch consumption. Publishing binary headers also fails because producers call .encode() on every value.

Fixes #2458. Fixes #2214.

publish header: str -> UTF-8 bytes; bytes/None -> unchanged
receive header:
  message.headers            -> compatible text view, invalid UTF-8 replaced
  message.raw_headers        -> original bytes/None without loss
  message.batch_raw_headers  -> originals for every message in a batch

Support both AIOKafka and Confluent, including publishers, responses, batches, FastAPI publishers and in-memory TestBrokers. Keep header decoding compatible with existing text consumers. Normalize Confluent header sequences before calling the native client. Review regressions cover each reserved header independently, especially reply_to without content-type, nullable/empty/malformed values, and first-message metadata in a batch.

Evidence

  • Before: on upstream c363b80a0 (2026-10-13), all four parser regression tests fail with UnicodeDecodeError. Restoring the old producer conversions also fails the binary publishing regressions.
  • After: 44 parser cases pass, including 40 focused reserved-header cases added from review feedback. The current affected-scenario run passes 273 tests; two existing cases remain expected failures for Bug: fake consumer doesn't account for dynamic routing key passed to publisher.publish #2513. This includes Confluent FastAPI/Response, binary TestBroker publishing and raw header access. The producer regressions and four real-broker modes also passed in the preceding validation.
  • Strict mypy passes on 1,913 source files. Ruff, codespell, Pyright, Pyrefly, Bandit, Semgrep, import contracts, slotscheck and the documentation build pass. Local pre-commit file checks pass; already completed analyzers were not run a second time through pre-commit.
  • Codespell excludes the ignored, generated graphify-out/ directory in addition to the generated documentation site.
  • Real Kafka 8.0.0 verification passes in all four modes: AIOKafka single/batch and Confluent single/batch. It checks Unicode strings, valid and invalid UTF-8 bytes, NUL, empty bytes and null headers, including every message in a batch.

Type of change

  • Bug fix
  • New feature: lossless raw header access and binary/null header publishing
  • Documentation update

Checklist

  • Code style and self-review completed
  • Documentation updated
  • Regression tests added and verified against old behavior
  • Selected unit and integration tests pass
  • Static analysis passes
  • Full just test-coverage matrix (left to CI)

Merge Danger

Door: two-way; revert the code to roll back.

Blast Radius: Kafka header handling in the AIOKafka and Confluent integrations. Existing decoded string headers remain available. Null values are omitted from the decoded view and retained in raw headers. Duplicate header entries remain accessible through the native message's header sequence.

@Bazarovinc
Bazarovinc requested a review from Lancetnik as a code owner August 15, 2025 13:53
@CLAassistant

CLAassistant commented Aug 15, 2025 •

Copy link
Copy Markdown

CLA assistant check
All committers have signed the CLA.

@github-actions github-actions Bot added Confluent Issues related to `faststream.confluent` module AioKafka Issues related to `faststream.kafka` module labels Aug 15, 2025
@Bazarovinc
Bazarovinc changed the base branch from dev to main August 15, 2025 14:09
@draincoder

Copy link
Copy Markdown
Collaborator

@Bazarovinc Please look at this #2214 issue as well and add comprehensive tests.

@Lancetnik
Lancetnik changed the base branch from main to dev August 15, 2025 18:26
@Bazarovinc
Bazarovinc marked this pull request as draft August 16, 2025 08:30
@Bazarovinc

Copy link
Copy Markdown
Contributor Author

@draincoder Here’s a preliminary implementation for handling Kafka message headers. Please review and let me know if this approach looks good or if any adjustments are needed.

@Lancetnik
Lancetnik changed the base branch from dev to main September 4, 2025 20:51
Comment thread faststream/confluent/parser.py Outdated
Comment thread faststream/confluent/parser.py Outdated
Comment thread faststream/confluent/parser.py Outdated
@Bazarovinc
Bazarovinc force-pushed the fix/non-utf8-kafka-message-headers branch from 8585bf6 to d0ee2c6 Compare April 21, 2026 19:07
@Bazarovinc
Bazarovinc force-pushed the fix/non-utf8-kafka-message-headers branch from d0ee2c6 to b326e40 Compare April 21, 2026 19:45
@fstermann

Copy link
Copy Markdown

@Bazarovinc @Sehat1137 @draincoder Any updates on this PR? We are currently stumbling into the same original issue and needed to implement a workaround for it. Would be nice to have this included in a near future release, thanks for an update! :)

@github-actions github-actions Bot added the documentation Improvements or additions to documentation label Oct 17, 2026
@Bazarovinc Bazarovinc changed the title Fix: Handle non-UTF-8 headers in Kafka/Confluent message parsing fix: support binary Kafka headers without decoding errors Oct 17, 2026
@Bazarovinc
Bazarovinc force-pushed the fix/non-utf8-kafka-message-headers branch from 708e64f to 79ed88c Compare October 17, 2026 16:24
@Bazarovinc
Bazarovinc requested a review from Sehat1137 October 17, 2026 16:30
@Bazarovinc
Bazarovinc marked this pull request as ready for review October 17, 2026 16:31
@Bazarovinc

Copy link
Copy Markdown
Contributor Author

@draincoder @fstermann Following up on the binary-header/test-coverage request and the status question: the PR has been updated and is ready for another review.

Both AIOKafka and Confluent now accept str | bytes | None header values for publishing, including batches. The shared encoder encodes strings once and passes bytes/null through unchanged. Consumers keep the compatible decoded message.headers view; message.raw_headers and message.batch_raw_headers expose binary/null values without replacement. Native header sequences remain available through raw_message when duplicate header names matter.

Coverage includes single/batch binary publishing and Response forwarding, real-producer encoding, native Confluent sequence normalization, raw-header access, and 44 parser cases. The focused local affected-scenario run passed 273 tests with two existing expected failures for #2513; real Kafka validation passed for both drivers in single and batch mode.

The full CI matrix has now passed on 79ed88ccb, including Python 3.10–3.14 and the real-broker jobs. The CLA check also confirms that all committers have signed. I have replied to each inline review thread with the implementation and regression-test details.

Comment thread tests/brokers/base/header.py Outdated
Comment thread tests/brokers/base/header.py Outdated
Comment thread tests/brokers/base/header.py Outdated
Comment thread tests/brokers/confluent/test_producer_headers.py
Comment thread tests/brokers/confluent/test_producer_headers.py Outdated
Comment thread tests/brokers/confluent/test_parser.py
Comment thread tests/brokers/confluent/test_raw_headers.py Outdated
Comment thread tests/brokers/kafka/test_parser.py
Comment thread tests/brokers/kafka/test_producer_headers.py Outdated
Comment thread tests/brokers/kafka/test_raw_headers.py Outdated
@Bazarovinc

Copy link
Copy Markdown
Contributor Author

@IvanKirpichnikov Addressed all 10 comments from your latest review in cf7042938.

  • The shared binary-header tests now use the mock fixture and mock.assert_called_once_with(...) instead of collecting results in lists.
  • Split single-message and batch scenarios into separate tests for producer encoding, reserved-header parsing, and raw-header access in both AIOKafka and Confluent.
  • Split the native Confluent header-sequence normalization check into separate send and send_batch tests.

All existing header-value cases are retained. The focused local suite passes all 63 cases. Mypy (1,913 files), Ruff, the remaining static analyzers, and file-level pre-commit checks passed; codespell excluded the generated graphify-out directory.

CI is also green for this commit: Run all tests, Run linters, and CodeQL.

Ready for another review.

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

AioKafka Issues related to `faststream.kafka` module Confluent Issues related to `faststream.confluent` module documentation Improvements or additions to documentation

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Bug:FastStream fails to decode non-UTF-8 Kafka message headers Feature: allow defining kafka headers as bytes

7 participants