feat: publish TOMBSTONE explicitly, None encodes normally

Consumers already told None apart from TOMBSTONE. Publish did not,
so there was no way to send a real b"null". Now None goes through
the codec like any other value, and TOMBSTONE is the only way to
send a real Kafka tombstone. aiokafka requires a key for a
tombstone, confluent does not enforce it but a key is still
recommended.

Author: aradng

docs/docs/en/confluent/message.md

@@ -29,6 +29,8 @@ This object serves as a unified **FastStream** wrapper around the native broker
 !!! note
     A `None` value() is a Kafka tombstone, the delete marker on a compacted topic. FastStream represents it as `faststream.message.TOMBSTONE`, kept distinct from an empty payload (`b""`).
 
+    To publish a tombstone, pass `faststream.message.TOMBSTONE` as the message body, along with a key. `broker.publish(TOMBSTONE, key=b"...")`
+
 For example, if you would like to access the headers of an incoming message, you would do so like this:
 
 ```python hl_lines="1 6"

docs/docs/en/kafka/message.md

@@ -34,6 +34,8 @@ This object serves as a unified **FastStream** wrapper around the native broker
 !!! note
     A `None` value is a Kafka tombstone, the delete marker on a compacted topic. FastStream represents it as `faststream.message.TOMBSTONE`, kept distinct from an empty payload (`b""`).
 
+    To publish a tombstone, pass `faststream.message.TOMBSTONE` as the message body and a key. `broker.publish(TOMBSTONE, key=b"...")`
+
 For example, if you would like to access the headers of an incoming message, you would do so like this:
 
 ```python hl_lines="1 6"

faststream/confluent/publisher/producer.py

@@ -9,6 +9,7 @@
 from faststream.confluent.parser import AsyncConfluentParser
 from faststream.confluent.response import KafkaPublishCommand
 from faststream.exceptions import FeatureNotSupportedException
+from faststream.message import TOMBSTONE
 
 from .state import EmptyProducerState, ProducerState, RealProducer
 
@@ -139,7 +140,10 @@ async def publish(
         cmd: "KafkaPublishCommand",
     ) -> "asyncio.Future[Message | None] | Message | None":
         """Publish a message to a topic."""
-        if cmd.body is None:
+        if cmd.body is TOMBSTONE:
+            # None goes through the codec like any other value now (it can
+            # encode to a real b"null"). TOMBSTONE is the explicit way to
+            # send a real Kafka tombstone.
             message, content_type = None, None
         else:
             message, content_type = await self.codec.encode(cmd.body, self.serializer)

faststream/confluent/testing.py

@@ -22,6 +22,7 @@
 from faststream.confluent.subscriber.usecase import BatchSubscriber
 from faststream.exceptions import SubscriberNotFound
 from faststream.message import gen_cor_id
+from faststream.message.utils import _Tombstone
 
 if TYPE_CHECKING:
     from fast_depends.library.serializer import SerializerProto
@@ -279,7 +280,7 @@ async def _execute_handler(
 class MockConfluentMessage:
     def __init__(
         self,
-        raw_msg: bytes,
+        raw_msg: bytes | None,
         topic: str,
         key: bytes | str,
         headers: list[tuple[str, bytes]],
@@ -304,7 +305,7 @@ def __init__(
         self._timestamp = (timestamp_type, timestamp_ms)
 
     def len(self) -> int:
-        return len(self._raw_msg)
+        return 0 if self._raw_msg is None else len(self._raw_msg)
 
     def error(self) -> str | None:
         return self._error
@@ -327,12 +328,12 @@ def timestamp(self) -> tuple[int, int]:
     def topic(self) -> str:
         return self._topic
 
-    def value(self) -> bytes:
+    def value(self) -> bytes | None:
         return self._raw_msg
 
 
 async def build_message(
-    message: "SendableMessage",
+    message: "SendableMessage | _Tombstone",
     topic: str,
     *,
     correlation_id: str | None = None,
@@ -345,8 +346,11 @@ async def build_message(
     codec: Optional["CodecProto"] = None,
 ) -> MockConfluentMessage:
     """Build a mock confluent_kafka.Message for a sendable message."""
-    codec_instance = codec or DefaultCodec()
-    msg, content_type = await codec_instance.encode(message, serializer)
+    if isinstance(message, _Tombstone):
+        msg, content_type = None, None
+    else:
+        codec_instance = codec or DefaultCodec()
+        msg, content_type = await codec_instance.encode(message, serializer)
     k = key or b""
     headers = {
         "content-type": content_type or "",

faststream/kafka/publisher/producer.py

@@ -6,11 +6,12 @@
 from faststream._internal.endpoint.utils import ParserComposition
 from faststream._internal.parser import BatchCodecProto, DefaultCodec
 from faststream._internal.producer import ProducerProto
-from faststream.exceptions import FeatureNotSupportedException
+from faststream.exceptions import FeatureNotSupportedException, SetupError
 from faststream.kafka.exceptions import BatchBufferOverflowException
 from faststream.kafka.message import KafkaMessage
 from faststream.kafka.parser import AioKafkaParser
 from faststream.kafka.response import KafkaPublishCommand
+from faststream.message import TOMBSTONE
 
 from .state import EmptyProducerState, ProducerState, RealProducer
 
@@ -110,9 +111,14 @@ async def publish(
         cmd: "KafkaPublishCommand",
     ) -> Union["asyncio.Future[RecordMetadata]", "RecordMetadata"]:
         """Publish a message to a topic."""
-        if cmd.body is None and cmd.key is not None:
-            # keyed None is a tombstone: aiokafka requires at least key or value,
-            # so a keyless None still goes through the codec as b""
+        if cmd.body is TOMBSTONE:
+            # None now goes through the codec like any other value.
+            # TOMBSTONE is the explicit way to send a real Kafka tombstone.
+            # aiokafka requires at least a key or value, so a tombstone
+            # needs a key.
+            if cmd.key is None:
+                msg = "a Kafka tombstone requires a key"
+                raise SetupError(msg)
             message, content_type = None, None
         else:
             message, content_type = await self.codec.encode(cmd.body, self.serializer)

faststream/kafka/testing.py

@@ -16,7 +16,7 @@
     TestBroker,
     change_producer,
 )
-from faststream.exceptions import SubscriberNotFound
+from faststream.exceptions import SetupError, SubscriberNotFound
 from faststream.kafka import TopicPartition
 from faststream.kafka.broker import KafkaBroker
 from faststream.kafka.message import KafkaMessage
@@ -25,6 +25,7 @@
 from faststream.kafka.publisher.usecase import BatchPublisher
 from faststream.kafka.subscriber.usecase import BatchSubscriber
 from faststream.message import gen_cor_id
+from faststream.message.utils import _Tombstone
 
 if TYPE_CHECKING:
     from fast_depends.library.serializer import SerializerProto
@@ -296,7 +297,7 @@ async def _execute_handler(
 
 
 async def build_message(
-    message: "SendableMessage",
+    message: "SendableMessage | _Tombstone",
     topic: str,
     partition: int | None = None,
     timestamp_ms: int | None = None,
@@ -309,7 +310,13 @@ async def build_message(
     codec: Optional["CodecProto"] = None,
 ) -> "ConsumerRecord":
     """Build a Kafka ConsumerRecord for a sendable message."""
-    msg, content_type = await (codec or DefaultCodec()).encode(message, serializer)
+    if isinstance(message, _Tombstone):
+        if key is None:
+            msg_text = "a Kafka tombstone requires a key"
+            raise SetupError(msg_text)
+        msg, content_type = None, None
+    else:
+        msg, content_type = await (codec or DefaultCodec()).encode(message, serializer)
 
     k = key or b""
 
@@ -328,8 +335,8 @@ async def build_message(
         partition=partition or 0,
         key=k,
         serialized_key_size=len(k),
-        serialized_value_size=len(msg),
-        checksum=sum(msg),
+        serialized_value_size=0 if msg is None else len(msg),
+        checksum=0 if msg is None else sum(msg),
         offset=0,
         headers=[(i, j.encode()) for i, j in headers.items()],
         timestamp_type=1,

tests/brokers/confluent/test_publish.py

@@ -6,6 +6,7 @@
 
 from faststream import Context
 from faststream.confluent import KafkaPublishMessage, KafkaResponse
+from faststream.message import TOMBSTONE
 from tests.brokers.base.publish import BrokerPublishTestcase
 
 from .basic import ConfluentTestcaseConfig
@@ -301,7 +302,7 @@ async def handler(msg: Any, raw_msg=Context("message")) -> None:
         assert messages_queue.empty()
 
     @pytest.mark.asyncio()
-    async def test_publish_none_sends_a_real_tombstone(self, queue: str) -> None:
+    async def test_publish_none_encodes_normally(self, queue: str) -> None:
         pub_broker = self.get_broker(apply_types=True)
 
         values: asyncio.Queue[bytes | None] = asyncio.Queue()
@@ -314,7 +315,26 @@ async def handler(msg: Any = Context("message")) -> None:
 
         async with self.patch_broker(pub_broker) as br:
             await br.start()
-            await br.publish(None, queue, key=b"tombstone-key")
+            await br.publish(None, queue)
+            value = await asyncio.wait_for(values.get(), timeout=self.timeout)
+
+        assert value == b""
+
+    @pytest.mark.asyncio()
+    async def test_publish_tombstone_sends_a_real_tombstone(self, queue: str) -> None:
+        pub_broker = self.get_broker(apply_types=True)
+
+        values: asyncio.Queue[bytes | None] = asyncio.Queue()
+
+        args, kwargs = self.get_subscriber_params(queue)
+
+        @pub_broker.subscriber(*args, **kwargs)
+        async def handler(msg: Any = Context("message")) -> None:
+            await values.put(msg.raw_message.value())
+
+        async with self.patch_broker(pub_broker) as br:
+            await br.start()
+            await br.publish(TOMBSTONE, queue, key=b"tombstone-key")
             value = await asyncio.wait_for(values.get(), timeout=self.timeout)
 
         assert value is None

tests/brokers/confluent/test_test_client.py

@@ -3,10 +3,11 @@
 
 import pytest
 
-from faststream import AckPolicy, BaseMiddleware
+from faststream import AckPolicy, BaseMiddleware, Context
 from faststream.confluent.annotations import KafkaMessage
 from faststream.confluent.message import FAKE_CONSUMER
 from faststream.confluent.testing import FakeProducer
+from faststream.message import TOMBSTONE
 from tests.brokers.base.testclient import BrokerTestclientTestcase
 from tests.tools import spy_decorator
 
@@ -288,3 +289,37 @@ async def test_publisher_without_destination(self) -> None:
 
             await another_publisher.publish(None, topic="new-key")
             another_publisher.mock.assert_called_once()
+
+    async def test_publish_none_encodes_normally(self, queue: str) -> None:
+        broker = self.get_broker(apply_types=True)
+
+        values: asyncio.Queue[bytes | None] = asyncio.Queue()
+
+        args, kwargs = self.get_subscriber_params(queue)
+
+        @broker.subscriber(*args, **kwargs)
+        async def handler(msg=Context("message")) -> None:
+            await values.put(msg.raw_message.value())
+
+        async with self.patch_broker(broker) as br:
+            await br.publish(None, queue)
+            value = await asyncio.wait_for(values.get(), timeout=3)
+
+        assert value == b""
+
+    async def test_publish_tombstone_sends_a_real_tombstone(self, queue: str) -> None:
+        broker = self.get_broker(apply_types=True)
+
+        values: asyncio.Queue[bytes | None] = asyncio.Queue()
+
+        args, kwargs = self.get_subscriber_params(queue)
+
+        @broker.subscriber(*args, **kwargs)
+        async def handler(msg=Context("message")) -> None:
+            await values.put(msg.raw_message.value())
+
+        async with self.patch_broker(broker) as br:
+            await br.publish(TOMBSTONE, queue, key=b"tombstone-key")
+            value = await asyncio.wait_for(values.get(), timeout=3)
+
+        assert value is None

tests/brokers/kafka/test_publish.py

@@ -6,8 +6,10 @@
 from aiokafka.structs import RecordMetadata
 
 from faststream import Context
+from faststream.exceptions import SetupError
 from faststream.kafka import KafkaPublishMessage, KafkaResponse
 from faststream.kafka.exceptions import BatchBufferOverflowException
+from faststream.message import TOMBSTONE
 from tests.brokers.base.publish import BrokerPublishTestcase
 
 from .basic import KafkaTestcaseConfig
@@ -353,7 +355,7 @@ async def handler(msg: Any, raw_msg=Context("message")) -> None:
         assert messages_queue.empty()
 
     @pytest.mark.asyncio()
-    async def test_publish_none_sends_a_real_tombstone(self, queue: str) -> None:
+    async def test_publish_none_encodes_normally(self, queue: str) -> None:
         pub_broker = self.get_broker(apply_types=True)
 
         values: asyncio.Queue[bytes | None] = asyncio.Queue()
@@ -364,7 +366,33 @@ async def handler(msg: Any = Context("message")) -> None:
 
         async with self.patch_broker(pub_broker) as br:
             await br.start()
-            await br.publish(None, queue, key=b"tombstone-key")
+            await br.publish(None, queue)
+            value = await asyncio.wait_for(values.get(), timeout=self.timeout)
+
+        assert value == b""
+
+    @pytest.mark.asyncio()
+    async def test_publish_tombstone_sends_a_real_tombstone(self, queue: str) -> None:
+        pub_broker = self.get_broker(apply_types=True)
+
+        values: asyncio.Queue[bytes | None] = asyncio.Queue()
+
+        @pub_broker.subscriber(queue)
+        async def handler(msg: Any = Context("message")) -> None:
+            await values.put(msg.raw_message.value)
+
+        async with self.patch_broker(pub_broker) as br:
+            await br.start()
+            await br.publish(TOMBSTONE, queue, key=b"tombstone-key")
             value = await asyncio.wait_for(values.get(), timeout=self.timeout)
 
         assert value is None
+
+    @pytest.mark.asyncio()
+    async def test_publish_tombstone_without_key_raises(self, queue: str) -> None:
+        pub_broker = self.get_broker(apply_types=True)
+
+        async with self.patch_broker(pub_broker) as br:
+            await br.start()
+            with pytest.raises(SetupError):
+                await br.publish(TOMBSTONE, queue)

tests/brokers/kafka/test_test_client.py

@@ -3,11 +3,13 @@
 
 import pytest
 
-from faststream import AckPolicy, BaseMiddleware
+from faststream import AckPolicy, BaseMiddleware, Context
+from faststream.exceptions import SetupError
 from faststream.kafka import TopicPartition
 from faststream.kafka.annotations import KafkaMessage
 from faststream.kafka.message import FAKE_CONSUMER
 from faststream.kafka.testing import FakeProducer
+from faststream.message import TOMBSTONE
 from tests.brokers.base.testclient import BrokerTestclientTestcase
 from tests.tools import spy_decorator
 
@@ -350,3 +352,40 @@ async def test_publisher_without_destination(self) -> None:
 
             await another_publisher.publish(None, topic="new-key")
             another_publisher.mock.assert_called_once()
+
+    async def test_publish_none_encodes_normally(self, queue: str) -> None:
+        broker = self.get_broker(apply_types=True)
+
+        values: asyncio.Queue[bytes | None] = asyncio.Queue()
+
+        @broker.subscriber(queue)
+        async def handler(msg=Context("message")) -> None:
+            await values.put(msg.raw_message.value)
+
+        async with self.patch_broker(broker) as br:
+            await br.publish(None, queue)
+            value = await asyncio.wait_for(values.get(), timeout=3)
+
+        assert value == b""
+
+    async def test_publish_tombstone_sends_a_real_tombstone(self, queue: str) -> None:
+        broker = self.get_broker(apply_types=True)
+
+        values: asyncio.Queue[bytes | None] = asyncio.Queue()
+
+        @broker.subscriber(queue)
+        async def handler(msg=Context("message")) -> None:
+            await values.put(msg.raw_message.value)
+
+        async with self.patch_broker(broker) as br:
+            await br.publish(TOMBSTONE, queue, key=b"tombstone-key")
+            value = await asyncio.wait_for(values.get(), timeout=3)
+
+        assert value is None
+
+    async def test_publish_tombstone_without_key_raises(self, queue: str) -> None:
+        broker = self.get_broker(apply_types=True)
+
+        async with self.patch_broker(broker) as br:
+            with pytest.raises(SetupError):
+                await br.publish(TOMBSTONE, queue)