Skip to content

Fix: Handle non-UTF-8 headers in Kafka/Confluent message parsing #2459

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wants to merge 5 commits into
base: main
Choose a base branch
from
+26 −19
Draft

Fix: Handle non-UTF-8 headers in Kafka/Confluent message parsing #2459

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
5dab07b
fix: add replacing non-utf-8 bytes on decoding headers
Aug 15, 2025
1d30220
draft: add support for binary Kafka headers (confluent only)
Aug 16, 2025
0014459
refactor: Bytes decoding has been changed to 'replace' undecodable ch…
Aug 20, 2025
fe0f8bd
draft: remove headers decoding for confluent
Aug 22, 2025
d0ee2c6
feat: refactored the decoding of key headers
Aug 25, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 23 additions & 16 deletions faststream/confluent/parser.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,3 @@
from collections.abc import Sequence
from typing import TYPE_CHECKING, Any

from faststream.message import StreamMessage, decode_message
Expand Down Expand Up @@ -28,7 +27,15 @@ async def parse_message(
message: "Message",
) -> KafkaMessage:
"""Parses a Kafka message."""
headers = _parse_msg_headers(message.headers() or ())
headers = dict(message.headers() or ())
decoded_headers = {
header: value.decode(errors="replace")
for header, value in (
headers.get("reply_to"),
headers.get("content-type"),
headers.get("correlation_id"),
Comment on lines +34 to +36
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

what does headers.get return? Can you show example or write type?

Copy link
Contributor

@ozeranskii ozeranskii Oct 13, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A small note: comprehension iterates over values (bytes) rather than pairs (key, value), so the expression for header, value in (...) will raise a ValueError.
It is worth iterating over headers.items() or decoding values using a separate helper.

decoded_headers = {
    k: v.decode(errors="replace")
    for k, v in headers.items()
    if k in ("reply_to", "content-type", "correlation_id") and v
}

Or, even better and clearer, since iteration here seems excessive to me and makes the code difficult to understand. See https://github.com/ag2ai/faststream/pull/2459/files/d0ee2c6911627934e78d958536e428ebb2cb1ec6#r2427545938

reply_to = self._decode_header(headers, "reply_to")
content_type = self._decode_header(headers, "content-type")
correlation_id = self._decode_header(headers, "correlation_id")

)
}

body = message.value() or b""
offset = message.offset()
Expand All @@ -37,10 +44,10 @@ async def parse_message(
return KafkaMessage(
body=body,
headers=headers,
reply_to=headers.get("reply_to", ""),
content_type=headers.get("content-type"),
reply_to=decoded_headers.get("reply_to") or None,
content_type=decoded_headers.get("content-type") or None,
message_id=f"{offset}-{timestamp}",
correlation_id=headers.get("correlation_id"),
correlation_id=decoded_headers.get("correlation_id") or None,
raw_message=message,
consumer=self._consumer,
is_manual=self.is_manual,
Expand All @@ -52,14 +59,14 @@ async def parse_message_batch(
) -> KafkaMessage:
"""Parses a batch of messages from a Kafka consumer."""
body: list[Any] = []
batch_headers: list[dict[str, str]] = []
batch_headers: list[dict[str, bytes]] = []

first = message[0]
last = message[-1]

for m in message:
body.append(m.value() or b"")
batch_headers.append(_parse_msg_headers(m.headers() or ()))
batch_headers.append(dict(m.headers() or ()))

headers = next(iter(batch_headers), {})

Expand All @@ -69,10 +76,16 @@ async def parse_message_batch(
body=body,
headers=headers,
batch_headers=batch_headers,
reply_to=headers.get("reply_to", ""),
content_type=headers.get("content-type"),
reply_to=headers.get("reply_to").decode()
if "content-type" in headers
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

content-type -> reply_to

else None,
content_type=headers.get("content-type").decode()
if "content-type" in headers
else None,
message_id=f"{first.offset()}-{last.offset()}-{first_timestamp}",
correlation_id=headers.get("correlation_id"),
correlation_id=headers.get("correlation_id").decode()
if "correlation_id" in headers
else None,
Comment on lines +79 to +88
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What do u this about this way for better readable:

reply_to=headers.get("reply_to", b"").decode() or None
content_type=headers.get("content-type", b"").decode() or None
correlation_id=headers.get("correlation_id", b"").decode() or None

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I suggest doing it this way

@staticmethod
def _decode_header(headers: dict[str, bytes | None], key: str) -> str | None:
    val = headers.get(key)
    return val.decode(errors="replace") if val else None

and then, where we need to get the value, we do this

reply_to=self._decode_header(headers, "reply_to")
content_type=self._decode_header(headers, "content-type")
correlation_id=self._decode_header(headers, "correlation_id")

This is already applicable in parse_message and parse_message_batch from what I've noticed.

raw_message=message,
consumer=self._consumer,
is_manual=self.is_manual,
Expand All @@ -91,9 +104,3 @@ async def decode_message_batch(
) -> "DecodedMessage":
"""Decode a batch of messages."""
return [decode_message(await self.parse_message(m)) for m in msg.raw_message]


def _parse_msg_headers(
headers: Sequence[tuple[str, bytes | str]],
) -> dict[str, str]:
return {i: j if isinstance(j, str) else j.decode() for i, j in headers}
2 changes: 1 addition & 1 deletion faststream/kafka/parser.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@ async def parse_message(
message: Union["ConsumerRecord", "KafkaRawMessage"],
) -> "StreamMessage[ConsumerRecord]":
"""Parses a Kafka message."""
headers = {i: j.decode() for i, j in message.headers}
headers = {i: j.decode(errors="replace") for i, j in message.headers}

return self.msg_class(
body=message.value or b"",
Expand Down
4 changes: 2 additions & 2 deletions faststream/message/message.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,9 +31,9 @@ def __init__(
raw_message: "MsgType",
body: bytes | Any,
*,
headers: dict[str, Any] | None = None,
headers: dict[str, bytes] | None = None,
reply_to: str = "",
batch_headers: list[dict[str, Any]] | None = None,
batch_headers: list[dict[str, bytes]] | None = None,
path: dict[str, Any] | None = None,
content_type: str | None = None,
correlation_id: str | None = None,
Expand Down