Fix AwaitMessageTrigger missing super().__init__() call

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

@@ -180,6 +180,21 @@ async def test_cleanup_does_not_raise_without_consumer(self):
 
         await trigger.cleanup()
 
+    def test_trigger_has_task_instance_attribute(self):
+        """AwaitMessageTrigger must initialise _task_instance via super().__init__().
+
+        Without the super().__init__() call, accessing trigger.task_instance raises
+        AttributeError when used as an AssetWatcher trigger (where the triggerer never
+        sets the attribute via the task_instance setter).
+        """
+        trigger = AwaitMessageTrigger(
+            kafka_config_id="kafka_d",
+            topics=["noop"],
+        )
+
+        assert hasattr(trigger, "_task_instance")
+        assert trigger.task_instance is None
+
 
 @mark_common_msg_queue_test
 class TestMessageQueueTrigger: