[I] Deferred trigger tasks seem to get stuck in Airflow 3.0.2 [airflow]

[via GitHub]

haritamar opened a new issue, #52247:
URL: https://github.com/apache/airflow/issues/52247

   ### Apache Airflow version
   
   3.0.2
   
   ### If "Other Airflow 2 version" selected, which one?
   
   _No response_
   
   ### What happened?
   
   Hi, I'm currently in the process of migrating to Airflow 3.0.2.
   As a part of our Airflow project, we are utilizing `TriggerDagRunOperator` 
with `deferrable=True` to trigger downstream DAGs, and we also use 
`wait_for_completion=True` since we measure SLAs using the duration of the main 
DAG.
   
   We noticed that after some time the triggering tasks seem to get stuck at 
"deferred" even though the underlying DAG is completed:
   <img width="558" alt="Image" 
src="https://github.com/user-attachments/assets/7c94c2f9-6218-4162-be72-a59e60ed28d6";
 />
   
   I believe what happens behind the scenes is that the async loop in the 
triggerer gets stuck, there are logs that seem to indicate that:
   <img width="1723" alt="Image" 
src="https://github.com/user-attachments/assets/99aa6595-e46b-4a17-a860-8f762257fa1f";
 />
   
   ### What you think should happen instead?
   
   All deferred tasks should be completed after the underlying DAGs completed 
(with an acceptable small delay)
   
   ### How to reproduce
   
   The example below causes the issue to reproduce for me.
   The issue seems to be **statistical** - I need to trigger the DAG about 15 
times in a row and it will eventually get stuck.
   
   ```python
   from datetime import datetime
   from airflow.decorators import dag, task
   from airflow.providers.standard.operators.trigger_dagrun import 
TriggerDagRunOperator
   
   @dag(
       dag_id="test_defer_issue_a",
       start_date=datetime(2025, 6, 25),
       schedule=None,
   )
   def test_defer_issue_a():
       TriggerDagRunOperator(
           task_id="trigger_b_task",
           trigger_dag_id="test_defer_issue_b",
           deferrable=True,
           wait_for_completion=True
       )
   
   test_defer_issue_a()
   
   
   @dag(
       dag_id="test_defer_issue_b",
       start_date=datetime(2025, 6, 25),
       schedule=None,
   )
   def test_defer_issue_b():
       @task
       def test_task():
           print("test_task")
   
       test_task()
   
   test_defer_issue_b()
   ```
   
   ### Operating System
   
   MacOS / Linux (both locally and in our dev envs)
   
   ### Versions of Apache Airflow Providers
   
   _No response_
   
   ### Deployment
   
   Official Apache Airflow Helm Chart
   
   ### Deployment details
   
   We use the official helm chart, version 1.16.0.
   
   ### Anything else?
   
   The issue might be related to 
[this](https://github.com/apache/airflow/discussions/51875) one, but I wasn't 
sure if it's the same thing since the behavior is a bit different in my case.
   Feel free to merge if you feel otherwise.
   
   ### Are you willing to submit PR?
   
   - [ ] Yes I am willing to submit a PR!
   
   ### Code of Conduct
   
   - [x] I agree to follow this project's [Code of 
Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscr...@airflow.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org