Trigger and Watcher it not running for External-Event Driven Dag

[jason810496]

Apache Airflow version

latest v3-2-test branch and cherry-picked both #64598 patch and #64615 patch

Reproduce Step

Either a. or b. can work

a.

minimum Dag by @uranusjr

from __future__ import annotations

from airflow.providers.apache.kafka.triggers.await_message import AwaitMessageTrigger
from airflow.sdk import DAG, Asset, AssetWatcher, task

trigger = AwaitMessageTrigger(topics=())
asset = Asset("x", watchers=[AssetWatcher("am", trigger)])

with DAG(dag_id="t", schedule=asset) as dag:
    pass

b.
follow #64613 to reproduce with e2e flow

What happened

The Triggerer showing no trigger and watcher is running:

Additiaonlly, it show 0 tigger in the DB:

airflow=# SELECT * FROM trigger;
 id | classpath | kwargs | created_date | triggerer_id | queue 
----+-----------+--------+--------------+--------------+-------
(0 rows)

What you think should happen instead?

No response

Operating System

Breeze

Versions of Apache Airflow Providers

No response

Deployment

None

Deployment details

No response

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[jason810496]

Entrypoint of the add_asset_trigger_references in airflow-core/src/airflow/dag_processing/collection.py

My plan: verifying the serialization result from the v3-1-test and the v3-2-test (including #64598 patch and #64615 patch)

---

I found that running airflow dags reserialize explicitly in another shell make the trigger show up in DB -> then make the triggers start running.

Perhaps we can check the dag-parsing different starting from the airflow dags reserialize CLI and Dag-processor paring entrypoint.

---

Noted that: We can only show trigger RIGHT AFTER running the airflow dags reserialize, when the new dag-parsing loop comes, the triggers will be deleted from DB again.

Behavior by timeline:

• airflow dags list -> double checked the Dag is in unpause state via Airflow UI
• PGPASSWORD='airflow' psql -h postgres -U postgres -d airflow -c "SELECT * FROM trigger;" -> showing 0 rows
• On Triggerer Side: show 0 triggers and 0 triggers
• airflow dags reserialize
• PGPASSWORD='airflow' psql -h postgres -U postgres -d airflow -c "SELECT * FROM trigger;", showing trigger now

 id |                                 classpath                                 |                                                                                                                                                                                                                                                                    kwargs                                                                                                                                                                                                                                                                    |         created_date         | triggerer_id | queue 
----+---------------------------------------------------------------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+------------------------------+--------------+-------
 14 | airflow.providers.apache.kafka.triggers.await_message.AwaitMessageTrigger | gAAAAABpzj6kMEnuSkGpKccCThL-5hSukJnoFpSof6Rl3c45TinNW_tifet-BcZTF_PvkKJlJJ5nMJ_CCEgRpzrXDwmEkfuTqE0RwbxXy74LFf--HWLsmZx0JnfhPxiONNUoF6Sd7HjNnX6mpKov0O0Wqoo4yPQFLWw2anzbLBzpNb8SUdlOan4FGEIzYplOabaTIBHMeilp6Qv9ywMeZBJQkzYa4siHaDbulNhc6IZo_tGaK3oqENfsFyDCxbvxRldz1nw2Qx5ZbpOpWiJFtlByK-uq2LRtU-XamWpKkmpP6Jg7YVG3i6IH6aNQbuCTxWIgzqVrTmBQSY9iWO49cV-ITCOuRvGkdQrWes-rWsmf9ULmBr38dnA-dedBtKS7525WyZRfc6FtArup_d495WVZ52ke2cKWk8bTsJ5lzuy918DYs0wUGzOFsKefTasUdgweZmIqA-cmlkuRUq8hKCjX31fU4agmTDMVr37rgBG2x0-ELF0s8cKnPIYQER_hwcZsw7Prt_aa | 2026-04-02 10:02:12.42857+00 |            3 | 
(1 row)

• On Triggerer Side: showing 0 triggers , 1 watcher, but encounter the following error:

2026-04-02T10:01:20.327520Z [error    ] Trigger ID 9 exited with error apply_function() got an unexpected keyword argument 'Encoding.VAR' [airflow.jobs.triggerer_job_runner] error_detail=[{'exc_type': 'TypeError', 'exc_value': "apply_function() got an unexpected keyword argument 'Encoding.VAR'", 'exc_notes': [], 'syntax_error': None, 'is_cause': False, 'frames': [{'filename': '/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py', 'lineno': 1069, 'name': 'cleanup_finished_triggers'}, {'filename': '/usr/python/lib/python3.10/site-packages/greenback/_impl.py', 'lineno': 119, 'name': 'greenback_shim'}, {'filename': '/usr/python/lib/python3.10/site-packages/greenback/_impl.py', 'lineno': 208, 'name': '_greenback_shim'}, {'filename': '/usr/python/lib/python3.10/site-packages/greenback/_impl.py', 'lineno': 84, 'name': 'trampoline'}, {'filename': '/usr/python/lib/python3.10/site-packages/outcome/_impl.py', 'lineno': 185, 'name': 'send'}, {'filename': '/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py', 'lineno': 1216, 'name': 'run_trigger'}, {'filename': '/opt/airflow/providers/apache/kafka/src/airflow/providers/apache/kafka/triggers/await_message.py', 'lineno': 131, 'name': 'run'}, {'filename': '/usr/python/lib/python3.10/site-packages/asgiref/sync.py', 'lineno': 506, 'name': '__call__'}, {'filename': '/usr/python/lib/python3.10/site-packages/greenback/_impl.py', 'lineno': 217, 'name': '_greenback_shim'}, {'filename': '/usr/python/lib/python3.10/concurrent/futures/thread.py', 'lineno': 58, 'name': 'run'}, {'filename': '/usr/python/lib/python3.10/site-packages/asgiref/sync.py', 'lineno': 561, 'name': 'thread_handler'}], 'is_group': False, 'exceptions': []}] loc=triggerer_job_runner.py:796
2026-04-02T10:01:20.328088Z [error    ] Trigger exited without sending an event. Dependent tasks will be failed. [airflow.jobs.triggerer_job_runner] loc=triggerer_job_runner.py:796 name='ID 9'

• wait for a few second
• PGPASSWORD='airflow' psql -h postgres -U postgres -d airflow -c "SELECT * FROM trigger;" -> showing 0 rows due to new dag-parsing loop