Fix AwaitMessageTrigger missing _task_instance attribute

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 ()

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",