Skip to content

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

Closed

Description

@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:

Image

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

Activity

  1. self-assigned this
    on Apr 2, 2026
  2. added and removed
    needs-triagelabel for new issues that we didn't triage yet
    on Apr 2, 2026
  3. jason810496 commented on Apr 2, 2026

    @jason810496
    MemberAuthor

    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
  4. added
    priority:highHigh priority bug that should be patched quickly but does not require immediate new release
    on Apr 2, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

Labels

affected_version:3.2Use for reporting issues with 3.2area:corekind:bugThis is a clearly a bugpriority:highHigh priority bug that should be patched quickly but does not require immediate new release

Type

No type

Projects

No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions