Repository: airtai/faststream (language: python)

Issue:
Bug: Broker subscribe handler falls in Error when try to handle message without value (but key exists!!)
python: 3.12.7
faststream: 0.5.30 and 0.5.32


Source code samle:
```py
from faststream.kafka import KafkaBroker
from faststream.kafka import KafkaMessage

broker = KafkaBroker(
    bootstrap_servers=KAFKA_SERVER,
    security=get_security(),
    log_level=logging.ERROR
)

@broker.subscriber(
    TOPIC,
    group_id='some-group'
    auto_offset_reset='earliest',
    session_timeout_ms=KAFKA_SESSION_TIMEOUT_MS,
    heartbeat_interval_ms=KAFKA_HEARTBEAT_INTERVAL_MS,
    max_poll_interval_ms=KAFKA_MAX_POLL_INTERVAL_MS
)
async def handle_messages(msg: KafkaMessage) -> None:
  ...

async def run_broker():
    await broker.start()
```

We starts it and broker begin to consume messages but it falls in error when message has key, but don't have value.
This error occures on the Faststream files level, that is why I am not able to handle it in my handler function.


ERROR LOG:
```
File "/home/user/.local/lib/python3.12/site-packages/faststream/broker/subscriber/usecase.py", line 402, in process_message
     raise parsing_error
   File "/home/user/.local/lib/python3.12/site-packages/faststream/broker/subscriber/usecase.py", line 343, in process_message
     message = await h.is_suitable(msg, cache)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
   File "/home/user/.local/lib/python3.12/site-packages/faststream/broker/subscriber/call_item.py", line 143, in is_suitable
     message._decoded_body = cache[decoder] = cache.get(decoder) or await decoder(
                                                                    ^^^^^^^^^^^^^^
   File "/home/user/.local/lib/python3.12/site-packages/faststream/kafka/parser.py", line 53, in decode_message
     return decode_message(msg)
            ^^^^^^^^^^^^^^^^^^^
   File "/home/user/.local/lib/python3.12/site-packages/faststream/broker/message.py", line 124, in decode_message
     m = json_loads(body)
         ^^^^^^^^^^^^^^^^
   File "/usr/local/lib/python3.12/json/__init__.py", line 339, in loads
     raise TypeError(f'the JSON object must be str, bytes or bytearray, '
 TypeError: the JSON object must be str, bytes or bytearray, not NoneType
 2024-12-05 16:09:47,553 ERROR    -                                                                                                                       |                       |            - Processed
```

I have tried to handle it with custom value serializer but it doesn't help 

Bug: Broker subscribe handler falls in Error when try to handle message without value (but key exists!!)
python: 3.12.7
faststream: 0.5.30 and 0.5.32


Source code samle:
```py
from faststream.kafka import KafkaBroker
from faststream.kafka import KafkaMessage

broker = KafkaBroker(
    bootstrap_servers=KAFKA_SERVER,
    security=get_security(),
    log_level=logging.ERROR
)

@broker.subscriber(
    TOPIC,
    group_id='some-group'
    auto_offset_reset='earliest',
    session_timeout_ms=KAFKA_SESSION_TIMEOUT_MS,
    heartbeat_interval_ms=KAFKA_HEARTBEAT_INTERVAL_MS,
    max_poll_interval_ms=KAFKA_MAX_POLL_INTERVAL_MS
)
async def handle_messages(msg: KafkaMessage) -> None:
  ...

async def run_broker():
    await broker.start()
```

We starts it and broker begin to consume messages but it falls in error when message has key, but don't have value.
This error occures on the Faststream files level, that is why I am not able to handle it in my handler function.


ERROR LOG:
```
File "/home/user/.local/lib/python3.12/site-packages/faststream/broker/subscriber/usecase.py", line 402, in process_message
     raise parsing_error
   File "/home/user/.local/lib/python3.12/site-packages/faststream/broker/subscriber/usecase.py", line 343, in process_message
     message = await h.is_suitable(msg, cache)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
   File "/home/user/.local/lib/python3.12/site-packages/faststream/broker/subscriber/call_item.py", line 143, in is_suitable
     message._decoded_body = cache[decoder] = cache.get(decoder) or await decoder(
                                                                    ^^^^^^^^^^^^^^
   File "/home/user/.local/lib/python3.12/site-packages/faststream/kafka/parser.py", line 53, in decode_message
     return decode_message(msg)
            ^^^^^^^^^^^^^^^^^^^
   File "/home/user/.local/lib/python3.12/site-packages/faststream/broker/message.py", line 124, in decode_message
     m = json_loads(body)
         ^^^^^^^^^^^^^^^^
   File "/usr/local/lib/python3.12/json/__init__.py", line 339, in loads
     raise TypeError(f'the JSON object must be str, bytes or bytearray, '
 TypeError: the JSON object must be str, bytes or bytearray, not NoneType
 2024-12-05 16:09:47,553 ERROR    -                                                                                                                       |                       |            - Processed
```

I have tried to handle it with custom value serializer but it doesn't help 



Discussion / hints:
Can you show the published message itself? I can't reproduce the problem without it
Well, I found MRE

```python
from faststream import FastStream
from faststream.kafka import KafkaBroker

broker = KafkaBroker()
app = FastStream(broker)

@broker.subscriber("test")
async def handler(): ...

@app.after_startup
async def t():
    await broker._producer._producer.send("test", key=b"test")
```
> Can you show the published message itself? I can't reproduce the problem without it

{
	"Value": null,
	"Offset": 4209,
	"Key": "some-key",
	"Partition": 0,
	"Headers": {},
	"Timestamp": "2024-12-05T13:09:45.028Z"
}

message example from kafka-ui. As fact this message wasn't published by faststream publisher instance.
Can you show the published message itself? I can't reproduce the problem without it
Well, I found MRE

```python
from faststream import FastStream
from faststream.kafka import KafkaBroker

broker = KafkaBroker()
app = FastStream(broker)

@broker.subscriber("test")
async def handler(): ...

@app.after_startup
async def t():
    await broker._producer._producer.send("test", key=b"test")
```
> Can you show the published message itself? I can't reproduce the problem without it

{
	"Value": null,
	"Offset": 4209,
	"Key": "some-key",
	"Partition": 0,
	"Headers": {},
	"Timestamp": "2024-12-05T13:09:45.028Z"
}

message example from kafka-ui. As fact this message wasn't published by faststream publisher instance.

Fix patch:
diff --git a/faststream/__about__.py b/faststream/__about__.py
index bca873f17e..2c1ebe109b 100644
--- a/faststream/__about__.py
+++ b/faststream/__about__.py
@@ -1,5 +1,5 @@
 """Simple and fast framework to create message brokers based microservices."""
 
-__version__ = "0.5.32"
+__version__ = "0.5.33"
 
 SERVICE_NAME = f"faststream-{__version__}"
diff --git a/faststream/confluent/parser.py b/faststream/confluent/parser.py
index 24f3de19ce..fff96a8f12 100644
--- a/faststream/confluent/parser.py
+++ b/faststream/confluent/parser.py
@@ -22,7 +22,7 @@ async def parse_message(
         """Parses a Kafka message."""
         headers = _parse_msg_headers(message.headers() or ())
 
-        body = message.value()
+        body = message.value() or b""
         offset = message.offset()
         _, timestamp = message.timestamp()
 
@@ -52,7 +52,7 @@ async def parse_message_batch(
         last = message[-1]
 
         for m in message:
-            body.append(m.value())
+            body.append(m.value() or b"")
             batch_headers.append(_parse_msg_headers(m.headers() or ()))
 
         headers = next(iter(batch_headers), {})
diff --git a/faststream/kafka/parser.py b/faststream/kafka/parser.py
index 92ad8c163f..bea29db170 100644
--- a/faststream/kafka/parser.py
+++ b/faststream/kafka/parser.py
@@ -34,7 +34,7 @@ async def parse_message(
         handler: Optional[LogicSubscriber[Any]] = context.get_local("handler_")
 
         return self.msg_class(
-            body=message.value,
+            body=message.value or b"",
             headers=headers,
             reply_to=headers.get("reply_to", ""),
             content_type=headers.get("content-type"),
@@ -72,7 +72,7 @@ async def parse_message(
         last = message[-1]
 
         for m in message:
-            body.append(m.value)
+            body.append(m.value or b"")
             batch_headers.append({i: j.decode() for i, j in m.headers})
 
         headers = next(iter(batch_headers), {})


Test patch:
diff --git a/tests/brokers/kafka/test_consume.py b/tests/brokers/kafka/test_consume.py
index 9593f61a48..ed7fc52f7c 100644
--- a/tests/brokers/kafka/test_consume.py
+++ b/tests/brokers/kafka/test_consume.py
@@ -351,3 +351,57 @@ async def handler(msg):
         assert event.is_set()
         assert event2.is_set()
         assert mock.call_count == 2, mock.call_count
+
+    @pytest.mark.asyncio
+    async def test_consume_without_value(
+        self,
+        mock: MagicMock,
+        queue: str,
+        event: asyncio.Event,
+    ) -> None:
+        consume_broker = self.get_broker()
+
+        @consume_broker.subscriber(queue)
+        async def handler(msg):
+            event.set()
+            mock(msg)
+
+        async with self.patch_broker(consume_broker) as br:
+            await br.start()
+
+            await asyncio.wait(
+                (
+                    asyncio.create_task(br._producer._producer.send(queue, key=b"")),
+                    asyncio.create_task(event.wait()),
+                ),
+                timeout=3,
+            )
+
+            mock.assert_called_once_with(b"")
+
+    @pytest.mark.asyncio
+    async def test_consume_batch_without_value(
+        self,
+        mock: MagicMock,
+        queue: str,
+        event: asyncio.Event,
+    ) -> None:
+        consume_broker = self.get_broker()
+
+        @consume_broker.subscriber(queue, batch=True)
+        async def handler(msg):
+            event.set()
+            mock(msg)
+
+        async with self.patch_broker(consume_broker) as br:
+            await br.start()
+
+            await asyncio.wait(
+                (
+                    asyncio.create_task(br._producer._producer.send(queue, key=b"")),
+                    asyncio.create_task(event.wait()),
+                ),
+                timeout=3,
+            )
+
+            mock.assert_called_once_with([b""])