fix: support binary Kafka headers without decoding errors

docs/docs/en/confluent/message.md

@@ -20,6 +20,8 @@ This object serves as a unified **FastStream** wrapper around the native broker
 
 * `#!python body: bytes`
 * `#!python headers: dict[str, str]`
+* `#!python raw_headers: dict[str, bytes | None]`
+* `#!python batch_raw_headers: list[dict[str, bytes | None]]`
 
 and its `raw_message` (the `confluent_kafka.Message`) exposes the rest through methods, including:
 
@@ -59,6 +61,28 @@ async def base_handler(
     print(msg.raw_message.headers())
 ```
 
+## Binary Headers
+
+Publishing methods, publishers, and `KafkaResponse` accept header values of type
+`str`, `bytes`, or `None`. Strings are encoded as UTF-8; bytes and null values are
+sent unchanged. For example, pass `headers={"event_id": b"\x00\xff"}` to
+`broker.publish()` or `broker.publish_batch()`.
+
+Use `message.raw_headers["event_id"]` to read the original bytes. Kafka does not
+record whether a header was originally a string or bytes, so `raw_headers` also
+contains bytes for headers published as strings.
+
+The existing `message.headers` remains a decoded string view: invalid UTF-8 bytes
+are replaced with `�`, and null values are omitted. Empty bytes become an empty
+string. `raw_headers` retains invalid bytes, empty bytes, and null values without
+loss. `raw_headers` keeps the last value when a header name occurs more than
+once; use `message.raw_message.headers()` if you need duplicate entries.
+
+For a batch, `raw_headers` contains the first message's headers and
+`batch_raw_headers` contains one dictionary per message, in order. For a single
+message, `batch_raw_headers` is empty. The dictionaries are snapshots: changing
+them does not modify the native message.
+
 ## Message Fields Access
 
 In most cases, you don't need all message fields; you need to know just a part of them.

docs/docs/en/kafka/message.md

@@ -20,6 +20,8 @@ This object serves as a unified **FastStream** wrapper around the native broker
 
 * `#!python body: bytes`
 * `#!python headers: dict[str, str]`
+* `#!python raw_headers: dict[str, bytes | None]`
+* `#!python batch_raw_headers: list[dict[str, bytes | None]]`
 
 and its `raw_message` (the `aiokafka.ConsumerRecord`) contains the rest of the required information, including:
 
@@ -48,6 +50,28 @@ async def base_handler(
     print(msg.headers)
 ```
 
+## Binary Headers
+
+Publishing methods, publishers, and `KafkaResponse` accept header values of type
+`str`, `bytes`, or `None`. Strings are encoded as UTF-8; bytes and null values are
+sent unchanged. For example, pass `headers={"event_id": b"\x00\xff"}` to
+`broker.publish()` or `broker.publish_batch()`.
+
+Use `message.raw_headers["event_id"]` to read the original bytes. Kafka does not
+record whether a header was originally a string or bytes, so `raw_headers` also
+contains bytes for headers published as strings.
+
+The existing `message.headers` remains a decoded string view: invalid UTF-8 bytes
+are replaced with `�`, and null values are omitted. Empty bytes become an empty
+string. `raw_headers` retains invalid bytes, empty bytes, and null values without
+loss. `raw_headers` keeps the last value when a header name occurs more than
+once; use the native message's header sequence if you need duplicate entries.
+
+For a batch, `raw_headers` contains the first message's headers and
+`batch_raw_headers` contains one dictionary per message, in order. For a single
+message, `batch_raw_headers` is empty. The dictionaries are snapshots: changing
+them does not modify the native message.
+
 ## Message Fields Access
 
 In most cases, you don't need all message fields; you need to know just a part of them.

faststream/_internal/kafka/__init__.py

@@ -6,6 +6,7 @@
 """
 
 from .calls import KafkaCallAssertions, field_reader
+from .headers import encode_headers
 from .keys import (
     extract_per_message_keys_and_bodies,
     key_for_index,
@@ -14,6 +15,7 @@
 
 __all__ = (
     "KafkaCallAssertions",
+    "encode_headers",
     "extract_per_message_keys_and_bodies",
     "field_reader",
     "key_for_index",

faststream/_internal/kafka/headers.py

@@ -0,0 +1,13 @@
+"""Header encoding shared by the Kafka drivers and their test brokers."""
+
+from collections.abc import Mapping
+
+
+def encode_headers(
+    headers: Mapping[str, str | bytes | None],
+) -> list[tuple[str, bytes | None]]:
+    """Encode text values while preserving binary and null headers."""
+    return [
+        (name, value.encode() if isinstance(value, str) else value)
+        for name, value in headers.items()
+    ]

faststream/confluent/broker/broker.py

@@ -1,6 +1,6 @@
 import asyncio
 import logging
-from collections.abc import Callable, Iterable, Sequence
+from collections.abc import Callable, Iterable, Mapping, Sequence
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -341,7 +341,7 @@ async def publish(
         key: bytes | str | None = None,
         partition: int | None = None,
         timestamp_ms: int | None = None,
-        headers: dict[str, str] | None = None,
+        headers: Mapping[str, str | bytes | None] | None = None,
         correlation_id: str | None = None,
         reply_to: str = "",
         no_confirm: Literal[False] = False,
@@ -356,7 +356,7 @@ async def publish(
         key: bytes | str | None = None,
         partition: int | None = None,
         timestamp_ms: int | None = None,
-        headers: dict[str, str] | None = None,
+        headers: Mapping[str, str | bytes | None] | None = None,
         correlation_id: str | None = None,
         reply_to: str = "",
         no_confirm: Literal[True] = ...,
@@ -371,7 +371,7 @@ async def publish(
         key: bytes | str | None = None,
         partition: int | None = None,
         timestamp_ms: int | None = None,
-        headers: dict[str, str] | None = None,
+        headers: Mapping[str, str | bytes | None] | None = None,
         correlation_id: str | None = None,
         reply_to: str = "",
         no_confirm: bool = False,
@@ -386,7 +386,7 @@ async def publish(
         key: bytes | str | None = None,
         partition: int | None = None,
         timestamp_ms: int | None = None,
-        headers: dict[str, str] | None = None,
+        headers: Mapping[str, str | bytes | None] | None = None,
         correlation_id: str | None = None,
         reply_to: str = "",
         no_confirm: bool = False,
@@ -438,7 +438,7 @@ async def request(  # type: ignore[override]
         key: bytes | str | None = None,
         partition: int | None = None,
         timestamp_ms: int | None = None,
-        headers: dict[str, str] | None = None,
+        headers: Mapping[str, str | bytes | None] | None = None,
         correlation_id: str | None = None,
         timeout: float = 0.5,
     ) -> "KafkaMessage":
@@ -467,7 +467,7 @@ async def publish_batch(  # type: ignore[override]
         topic: str,
         partition: int | None = None,
         timestamp_ms: int | None = None,
-        headers: dict[str, str] | None = None,
+        headers: Mapping[str, str | bytes | None] | None = None,
         reply_to: str = "",
         correlation_id: str | None = None,
         no_confirm: bool = False,

faststream/confluent/broker/registrator.py

@@ -1,4 +1,4 @@
-from collections.abc import Callable, Sequence
+from collections.abc import Callable, Mapping, Sequence
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -467,7 +467,7 @@ def publisher(
         *,
         key: bytes | str | None = None,
         partition: int | None = None,
-        headers: dict[str, str] | None = None,
+        headers: Mapping[str, str | bytes | None] | None = None,
         reply_to: str = "",
         batch: Literal[False] = False,
         # basic args
@@ -487,7 +487,7 @@ def publisher(
         *,
         key: bytes | str | None = None,
         partition: int | None = None,
-        headers: dict[str, str] | None = None,
+        headers: Mapping[str, str | bytes | None] | None = None,
         reply_to: str = "",
         batch: Literal[True] = ...,
         # basic args
@@ -507,7 +507,7 @@ def publisher(
         *,
         key: bytes | str | None = None,
         partition: int | None = None,
-        headers: dict[str, str] | None = None,
+        headers: Mapping[str, str | bytes | None] | None = None,
         reply_to: str = "",
         batch: bool = False,
         # basic args
@@ -530,7 +530,7 @@ def publisher(
         *,
         key: bytes | str | None = None,
         partition: int | None = None,
-        headers: dict[str, str] | None = None,
+        headers: Mapping[str, str | bytes | None] | None = None,
         reply_to: str = "",
         batch: bool = False,
         # basic args

faststream/confluent/broker/router.py

@@ -1,4 +1,4 @@
-from collections.abc import Awaitable, Callable, Iterable, Sequence
+from collections.abc import Awaitable, Callable, Iterable, Mapping, Sequence
 from typing import TYPE_CHECKING, Any, Literal, Optional, Union
 
 from faststream._internal.broker.router import (
@@ -37,7 +37,7 @@ def __init__(
         *,
         key: bytes | str | None = None,
         partition: int | None = None,
-        headers: dict[str, str] | None = None,
+        headers: Mapping[str, str | bytes | None] | None = None,
         reply_to: str = "",
         batch: bool = False,
         # AsyncAPI args

faststream/confluent/fastapi/fastapi.py

@@ -1,5 +1,5 @@
 import logging
-from collections.abc import Callable, Iterable, Sequence
+from collections.abc import Callable, Iterable, Mapping, Sequence
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -919,7 +919,7 @@ def publisher(
         *,
         key: bytes | str | None = None,
         partition: int | None = None,
-        headers: dict[str, str] | None = None,
+        headers: Mapping[str, str | bytes | None] | None = None,
         reply_to: str = "",
         batch: Literal[False] = False,
         # basic args
@@ -937,7 +937,7 @@ def publisher(
         *,
         key: bytes | str | None = None,
         partition: int | None = None,
-        headers: dict[str, str] | None = None,
+        headers: Mapping[str, str | bytes | None] | None = None,
         reply_to: str = "",
         batch: Literal[True] = ...,
         # basic args
@@ -955,7 +955,7 @@ def publisher(
         *,
         key: bytes | str | None = None,
         partition: int | None = None,
-        headers: dict[str, str] | None = None,
+        headers: Mapping[str, str | bytes | None] | None = None,
         reply_to: str = "",
         batch: bool = False,
         # basic args
@@ -973,7 +973,7 @@ def publisher(
         *,
         key: bytes | str | None = None,
         partition: int | None = None,
-        headers: dict[str, str] | None = None,
+        headers: Mapping[str, str | bytes | None] | None = None,
         reply_to: str = "",
         batch: bool = False,
         # basic args

faststream/confluent/helpers/client.py

@@ -25,7 +25,7 @@
     class _SendKwargs(TypedDict):
         value: bytes | str | None
         key: bytes | str | None
-        headers: list[tuple[str, str | bytes]] | None
+        headers: Sequence[tuple[str, str | bytes | None]] | None
         partition: NotRequired[int]
         timestamp: NotRequired[int]
         on_delivery: NotRequired[Callable[..., None]]
@@ -101,7 +101,7 @@ async def send(
         key: bytes | str | None = None,
         partition: int | None = None,
         timestamp_ms: int | None = None,
-        headers: list[tuple[str, str | bytes]] | None = None,
+        headers: Sequence[tuple[str, str | bytes | None]] | None = None,
         no_confirm: bool = False,
     ) -> "asyncio.Future[Message | None] | Message | None":
         """Sends a single message to a Kafka topic."""
@@ -143,9 +143,9 @@ def ack_callback(err: Any, msg: Message | None) -> None:
             if isinstance(kwargs["key"], (bytes, type(None)))
             else kwargs["key"].encode()
         )
-        produce_headers: (
-            dict[str, str | bytes | None] | list[tuple[str, str | bytes | None]] | None
-        ) = cast("Any", kwargs["headers"]) if kwargs.get("headers") is not None else None
+        produce_headers = (
+            list(kwargs["headers"]) if kwargs["headers"] is not None else None
+        )
         produce_kwargs: dict[str, Any] = {
             "value": produce_value,
             "key": produce_key,
@@ -466,7 +466,7 @@ def append(
         value: bytes | str | None = None,
         key: bytes | str | None = None,
         timestamp: int | None = None,
-        headers: list[tuple[str, bytes]] | None = None,
+        headers: Sequence[tuple[str, bytes | None]] | None = None,
     ) -> None:
         """Appends a message to the batch with optional timestamp, key, value, and headers."""
         if key is None and value is None:

faststream/confluent/message.py

@@ -59,6 +59,23 @@ def __init__(
         if not is_manual:
             self.committed = AckStatus.ACKED
 
+    @property
+    def raw_headers(self) -> dict[str, bytes | None]:
+        """Original header bytes, or the first message's headers for a batch."""
+        raw_message = self.raw_message
+        if isinstance(raw_message, tuple):
+            if not raw_message:
+                return {}
+            raw_message = raw_message[0]
+        return _get_raw_headers(raw_message)
+
+    @property
+    def batch_raw_headers(self) -> list[dict[str, bytes | None]]:
+        """Original header bytes for every batch message; empty for a single message."""
+        if isinstance(self.raw_message, tuple):
+            return [_get_raw_headers(message) for message in self.raw_message]
+        return []
+
     async def ack(self) -> None:
         """Acknowledge the Kafka message."""
         if self.is_manual and not self.committed:
@@ -86,3 +103,11 @@ async def nack(self) -> None:
                     offset=offset,
                 )
         await super().nack()
+
+
+def _get_raw_headers(message: Message) -> dict[str, bytes | None]:
+    headers = dict(message.headers() or ())
+    return {
+        key: value.encode() if isinstance(value, str) else value
+        for key, value in headers.items()
+    }

faststream/confluent/parser.py

@@ -109,8 +109,13 @@ async def decode_batch(
 
 
 def _parse_msg_headers(headers: "_HeadersInput") -> dict[str, str]:
+    """Decode header values into a text view without changing the native message."""
     if isinstance(headers, dict):
         seq: Sequence[tuple[str, bytes | str | None]] = list(headers.items())
     else:
         seq = headers
-    return {i: (j if isinstance(j, str) else j.decode()) for i, j in seq if j is not None}
+    return {
+        key: (value if isinstance(value, str) else value.decode(errors="replace"))
+        for key, value in seq
+        if value is not None
+    }

faststream/confluent/publisher/config.py

@@ -1,3 +1,4 @@
+from collections.abc import Mapping
 from dataclasses import dataclass, field
 
 from faststream._internal.configs import (
@@ -19,5 +20,5 @@ class KafkaPublisherConfig(PublisherUsecaseConfig):
     key: bytes | str | None
     topic: str
     partition: int | None
-    headers: dict[str, str] | None
+    headers: Mapping[str, str | bytes | None] | None
     reply_to: str

faststream/confluent/publisher/factory.py

@@ -1,3 +1,4 @@
+from collections.abc import Mapping
 from functools import wraps
 from typing import TYPE_CHECKING, Any, Union
 
@@ -20,7 +21,7 @@ def create_publisher(
     key: bytes | str | None,
     topic: Union[str, "Topic"],
     partition: int | None,
-    headers: dict[str, str] | None,
+    headers: Mapping[str, str | bytes | None] | None,
     reply_to: str,
     # Publisher args
     config: "KafkaBrokerConfig",