Fix AwaitMessageTrigger crash on tombstone messages when apply_function is unset

providers/apache/kafka/src/airflow/providers/apache/kafka/triggers/await_message.py

@@ -132,11 +132,14 @@ async def run(self):
             elif message.error():
                 raise AirflowException(f"Error: {message.error()}")
             else:
-                event = (
-                    await async_message_process(message)
-                    if async_message_process
-                    else message.value().decode("utf-8")
-                )
+                if async_message_process:
+                    event = await async_message_process(message)
+                else:
+                    value = message.value()
+                    # A tombstone (null value, e.g. from a log-compacted topic) carries no
+                    # payload to emit, so treat it like a non-matching message and keep polling
+                    # instead of raising AttributeError on ``None.decode()``.
+                    event = value.decode("utf-8") if value is not None else None
                 if event:
                     if self.commit_offset:
                         await async_commit(message=message, asynchronous=False)

providers/apache/kafka/tests/unit/apache/kafka/triggers/test_await_message.py

@@ -50,6 +50,22 @@ def __init__(*args, **kwargs):
     def error(*args, **kwargs):
         return False
 
+    def value(*args, **kwargs):
+        return b"test_message"
+
+
+class MockedTombstoneMessage:
+    """A message with a null payload, e.g. a deletion marker from a log-compacted topic."""
+
+    def __init__(*args, **kwargs):
+        pass
+
+    def error(*args, **kwargs):
+        return False
+
+    def value(*args, **kwargs):
+        return None
+
 
 class MockedConsumer:
     def __init__(*args, **kwargs) -> None:
@@ -162,6 +178,62 @@ async def test_trigger_run_bad(self, mocker):
         assert task.done() is False
         asyncio.get_event_loop().stop()
 
+    @pytest.mark.asyncio
+    async def test_trigger_run_tombstone_message_keeps_polling(self, mocker):
+        """
+        A tombstone (null-value) message must not crash the trigger when no apply_function is set.
+
+        Regression test: the trigger previously raised
+        ``AttributeError: 'NoneType' object has no attribute 'decode'`` on ``message.value()``
+        returning ``None``. A tombstone carries no payload to emit, so the trigger should
+        commit the offset (when ``commit_offset`` is enabled) and keep polling.
+        """
+        message = MockedTombstoneMessage()
+        consumer = MockedConsumer()
+        mocker.patch.object(consumer, "poll", return_value=message)
+        commit_mock = mocker.patch.object(consumer, "commit")
+        mocker.patch.object(KafkaConsumerHook, "get_consumer", return_value=consumer)
+
+        trigger = AwaitMessageTrigger(
+            kafka_config_id="kafka_d",
+            apply_function=None,
+            topics=["noop"],
+            poll_timeout=0.0001,
+            poll_interval=5,
+        )
+
+        task = asyncio.create_task(trigger.run().__anext__())
+        await asyncio.sleep(1.0)
+        try:
+            # The task must neither raise nor yield an event: the tombstone is skipped
+            # and the trigger sleeps through poll_interval waiting for the next message.
+            assert task.done() is False
+            # The tombstone offset is still committed, like any other non-matching message.
+            assert commit_mock.mock_calls == [call(message=message, asynchronous=False)]
+        finally:
+            task.cancel()
+
+    @pytest.mark.asyncio
+    async def test_trigger_run_without_apply_function_yields_message_value(self, mocker):
+        """A regular message with no apply_function yields the decoded message value."""
+        consumer = MockedConsumer()
+        mocker.patch.object(consumer, "poll", return_value=MockedMessage())
+        mocker.patch.object(consumer, "commit")
+        mocker.patch.object(KafkaConsumerHook, "get_consumer", return_value=consumer)
+
+        trigger = AwaitMessageTrigger(
+            kafka_config_id="kafka_d",
+            apply_function=None,
+            topics=["noop"],
+            poll_timeout=0.0001,
+            poll_interval=5,
+        )
+
+        task = asyncio.create_task(trigger.run().__anext__())
+        await asyncio.sleep(1.0)
+        assert task.done() is True
+        assert task.result().payload == "test_message"
+
     @pytest.mark.asyncio
     async def test_cleanup_closes_consumer(self, mocker):
         consumer = MockedConsumer()