How to ignore message.headers (corrupt data chokes consumer)

[ChadDa3mon]

Greetings,

Some of the upstream publishers (things I don't have control over) are putting corrupt data into message.headers and it's choking the consumer. How can I get things to ignore this? I can't do anything in my code like a try/except block as it's happening before things get to my function.

I don't need anything in the headers, so I'm hoping to find a way to just ignore this issue so I can parse out the body.

Example error:

2023-11-10 11:07:37,680 ERROR    -                |            - UnicodeDecodeError: 'utf-8' codec can't decode byte 0xe7 in position 8: invalid continuation byte
Traceback (most recent call last):
  File "/Users/me/.pyenv/versions/my-kafka-consumer/lib/python3.9/site-packages/faststream/broker/handler.py", line 277, in consume
    message = await parser(msg)
  File "/Users/me/.pyenv/versions/my-kafka-consumer/lib/python3.9/site-packages/faststream/utils/functions.py", line 90, in to_async_wrapper
    return await call_or_await(func, *args, **kwargs)
  File "/Users/me/.pyenv/versions/my-kafka-consumer/lib/python3.9/site-packages/fast_depends/utils.py", line 38, in run_async
    return await cast(Callable[P, Awaitable[T]], func)(*args, **kwargs)
  File "/Users/me/.pyenv/versions/my-kafka-consumer/lib/python3.9/site-packages/faststream/kafka/parser.py", line 29, in parse_message
    headers = {i: j.decode() for i, j in message.headers}
  File "/Users/me/.pyenv/versions/my-kafka-consumer/lib/python3.9/site-packages/faststream/kafka/parser.py", line 29, in <dictcomp>
    headers = {i: j.decode() for i, j in message.headers}
UnicodeDecodeError: 'utf-8' codec can't decode byte 0xe7 in position 8: invalid continuation byte

Small example of my function that handles this...it never gets to the logger.debug statement, so I'm not sure how I can effectively handle this.

@router.subscriber(settings.kafka.topic_consume)
async def handle_msg(payload: KafkaMessage):
    """
    Primary entry point, consumes a message from kafka, processes it, and will send a notification 
    if a match is found
    """
    logger.debug(f"New Alert: {payload}")

[Lancetnik]

Hi, @ChadDa3mon! In this case you should use custom wrapper
You can just wrap original parser calling by try ... except ... and raise RejectMessage exception on error. Also, you can replace broken headers by empty one at exception and retry to serialize it by original parser again.

[Lancetnik]

Also, if you need no headers at all, you can just replace all headers by empty one before passing message to the original parser