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 196a1d8aa4899..3dbc5e467d419 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 @@ -82,6 +82,7 @@ def __init__( poll_interval: float = 5, commit_offset: bool = True, ) -> None: + super().__init__() self.topics = topics self.apply_function = apply_function self.apply_function_args = apply_function_args or () 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 397992ec9ea40..374fdb0f2f96d 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 @@ -30,6 +30,7 @@ collect_queue_param_deprecation_warning, mark_common_msg_queue_test, ) +from tests_common.test_utils.version_compat import AIRFLOW_V_3_3_PLUS USED_FIXTURES = [collect_queue_param_deprecation_warning] @@ -77,6 +78,21 @@ def setup_connections(self, create_connection_without_db): ) ) + def test_trigger_initializes_base_state(self): + trigger = AwaitMessageTrigger( + kafka_config_id="kafka_d", + apply_function="test.noop", + topics=["noop"], + ) + + if AIRFLOW_V_3_3_PLUS: + # The trigger._task_instance attribute was introduced in https://github.com/apache/airflow/pull/55068 + assert trigger._task_instance is None + assert trigger.task_instance is None + else: + assert not hasattr(trigger, "_task_instance") + assert trigger.task_instance is None + def test_trigger_serialization(self): trigger = AwaitMessageTrigger( kafka_config_id="kafka_d",