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

[Bazarovinc]

This PR addresses the UnicodeDecodeError that occurred when processing Kafka messages containing headers with non-UTF-8 byte sequences

Fixes #2458, FIxes #2214

Type of change

• Bug fix (a non-breaking change that resolves an issue)

Checklist

• My code adheres to the style guidelines of this project (just lint shows no errors)
• I have conducted a self-review of my own code
• I have made the necessary changes to the documentation
• My changes do not generate any new warnings
• I have added tests to validate the effectiveness of my fix or the functionality of my new feature
• Both new and existing unit tests pass successfully on my local environment by running just test-coverage
• I have ensured that static analysis tests are passing by running just static-analysis
• I have included code examples to illustrate the modifications

[CLAassistant]

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our Contributor License Agreement before we can accept your contribution.

---

Nikita Veselenko seems not to be a GitHub user. You need a GitHub account to be able to sign the CLA. If you have already a GitHub account, please add the email address used for this commit to your account.
_{You have signed the CLA already but the status is still pending? Let us recheck it.}

[draincoder]

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

[Bazarovinc]

@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.

[Sehat1137]

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

[ozeranskii]

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")

[Sehat1137]

content-type -> reply_to

[Sehat1137]

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

[ozeranskii]

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.