Fix AwaitMessageTrigger missing _task_instance attribute

[jason810496]

• related: Add e2e test suite for Airflow event-driven DAGs with Apache Kafka #64833
• Re-enable start_from_trigger feature with rendering of template fields #55068 (introduce the trigger._task_instance)

Why

While working on adding E2E test for event driving Dag with Kafka in #64833, I encounter the following error:

  airflow-triggerer-1  | 2026-05-02T15:25:00.831436Z [error    ] Trigger ID 1 exited with error 'AwaitMessageTrigger' object has no attribute '_task_instance' [airflow.jobs.triggerer_job_runner] error_detail=[{'exc_type': 'AttributeError', 'exc_value': "'AwaitMessageTrigger' object has no attribute '_task_instance'", 'exc_notes': [], 'syntax_error': None, 'is_cause': False, 'frames': [{'filename': '/home/airflow/.local/lib/python3.10/site-packages/airflow/jobs/triggerer_job_runner.py', 'lineno': 1238, 'name': 'cleanup_finished_triggers'}, {'filename': '/home/airflow/.local/lib/python3.10/site-packages/greenback/_impl.py', 'lineno': 119, 'name': 'greenback_shim'}, {'filename': '/home/airflow/.local/lib/python3.10/site-packages/greenback/_impl.py', 'lineno': 208, 'name': '_greenback_shim'}, {'filename': '/home/airflow/.local/lib/python3.10/site-packages/greenback/_impl.py', 'lineno': 84, 'name': 'trampoline'}, {'filename': '/home/airflow/.local/lib/python3.10/site-packages/outcome/_impl.py', 'lineno': 185, 'name': 'send'}, {'filename': '/home/airflow/.local/lib/python3.10/site-packages/airflow/jobs/triggerer_job_runner.py', 'lineno': 1387, 'name': 'run_trigger'}, {'filename': '/home/airflow/.local/lib/python3.10/site-packages/airflow/triggers/base.py', 'lineno': 104, 'name': 'task_instance'}], 'is_group': False, 'exceptions': []}] loc=triggerer_job_runner.py:929
  airflow-triggerer-1  | 2026-05-02T15:25:00.832067Z [error    ] Trigger exited without sending an event. Dependent tasks will be failed. [airflow.jobs.triggerer_job_runner] loc=triggerer_job_runner.py:929 name='ID 1'
  airflow-triggerer-1  | 2026-05-02T15:25:02.858280Z [error    ] Trigger ID 1 exited with error 'AwaitMessageTrigger' object has no attribute '_task_instance' [airflow.jobs.triggerer_job_runner] error_detail=[{'exc_type': 'AttributeError', 'exc_value': "'AwaitMessageTrigger' object has no attribute '_task_instance'", 'exc_notes': [], 'syntax_error': None, 'is_cause': False, 'frames': [{'filename': '/home/airflow/.local/lib/python3.10/site-packages/airflow/jobs/triggerer_job_runner.py', 'lineno': 1238, 'name': 'cleanup_finished_triggers'}, {'filename': '/home/airflow/.local/lib/python3.10/site-packages/greenback/_impl.py', 'lineno': 119, 'name': 'greenback_shim'}, {'filename': '/home/airflow/.local/lib/python3.10/site-packages/greenback/_impl.py', 'lineno': 208, 'name': '_greenback_shim'}, {'filename': '/home/airflow/.local/lib/python3.10/site-packages/greenback/_impl.py', 'lineno': 84, 'name': 'trampoline'}, {'filename': '/home/airflow/.local/lib/python3.10/site-packages/outcome/_impl.py', 'lineno': 185, 'name': 'send'}, {'filename': '/home/airflow/.local/lib/python3.10/site-packages/airflow/jobs/triggerer_job_runner.py', 'lineno': 1387, 'name': 'run_trigger'}, {'filename': '/home/airflow/.local/lib/python3.10/site-packages/airflow/triggers/base.py', 'lineno': 104, 'name': 'task_instance'}], 'is_group': False, 'exceptions': []}] loc=triggerer_job_runner.py:929

The root cause is: AwaitMessageTrigger.__init__ never calls super().__init__(), so _task_instance is never set. The triggerer crashes on cleanup with AttributeError: 'AwaitMessageTrigger' object has no attribute '_task_instance', and dependent tasks fail.

What

• Call super().__init__() in AwaitMessageTrigger.__init__.
• Add a unit test asserting _task_instance / task_instance are initialized after construction.

---

Was generative AI tooling used to co-author this PR?

• Yes, with help of Claude Code following the guidelines

[Copilot]

Pull request overview

Fixes a crash in the Kafka provider’s AwaitMessageTrigger where BaseTrigger initialization was skipped, leaving _task_instance unset and causing AttributeError during triggerer cleanup.

Changes:

• Call super().__init__() in AwaitMessageTrigger.__init__() so base trigger state (including _task_instance) is always initialized.
• Add a unit test that verifies _task_instance / task_instance exist and start as None immediately after constructing the trigger.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated no comments.

File	Description
providers/apache/kafka/src/airflow/providers/apache/kafka/triggers/await_message.py	Ensures base trigger initialization runs by calling super().__init__() in the trigger constructor.
providers/apache/kafka/tests/unit/apache/kafka/triggers/test_await_message.py	Adds a regression test to confirm base trigger state is initialized on construction.