This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 6669e7eed8d Fix AwaitMessageTrigger crash on tombstone messages when
apply_function is unset (#69665)
6669e7eed8d is described below
commit 6669e7eed8d39c5ab9a549570314202ef7725797
Author: Tim <[email protected]>
AuthorDate: Fri Jul 31 22:50:28 2026 -0700
Fix AwaitMessageTrigger crash on tombstone messages when apply_function is
unset (#69665)
Signed-off-by: Timur Rakhmatullin
<[email protected]>
Co-authored-by: Timur Rakhmatullin
<[email protected]>
---
.../apache/kafka/triggers/await_message.py | 13 ++--
.../apache/kafka/triggers/test_await_message.py | 72 ++++++++++++++++++++++
2 files changed, 80 insertions(+), 5 deletions(-)
diff --git
a/providers/apache/kafka/src/airflow/providers/apache/kafka/triggers/await_message.py
b/providers/apache/kafka/src/airflow/providers/apache/kafka/triggers/await_message.py
index 3dbc5e467d4..645965cf7ad 100644
---
a/providers/apache/kafka/src/airflow/providers/apache/kafka/triggers/await_message.py
+++
b/providers/apache/kafka/src/airflow/providers/apache/kafka/triggers/await_message.py
@@ -132,11 +132,14 @@ class AwaitMessageTrigger(BaseEventTrigger):
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)
diff --git
a/providers/apache/kafka/tests/unit/apache/kafka/triggers/test_await_message.py
b/providers/apache/kafka/tests/unit/apache/kafka/triggers/test_await_message.py
index 374fdb0f2f9..ed950bea9ab 100644
---
a/providers/apache/kafka/tests/unit/apache/kafka/triggers/test_await_message.py
+++
b/providers/apache/kafka/tests/unit/apache/kafka/triggers/test_await_message.py
@@ -50,6 +50,22 @@ class MockedMessage:
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 @@ class TestTrigger:
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()
