Crowiant opened a new issue, #55299: URL: https://github.com/apache/airflow/issues/55299
### Apache Airflow version Other Airflow 2 version (please specify below) ### If "Other Airflow 2 version" selected, which one? 2.9.3, 2.10.5 ### What happened? I am using Airflow 2.9.3 with a listener plugin to track dag_run status, similar to the example in the official documentation: https://airflow.apache.org/docs/apache-airflow/2.9.3/howto/listener-plugin.html. I noticed that when a dag_run times out, the Failed status is triggered twice instead of once. The run_id gets a Failed status the first time when the dagrun_timeout is hit and then a second time a little while later. A similar issue appears to occur in version 2.10.5. Here you can see run_id repeats and have different duration: <img width="841" height="143" alt="Image" src="https://github.com/user-attachments/assets/08993e8b-40c4-4af6-8130-73608cee8867"; /> ### What you think should happen instead? Status Failed of DagRun should be emitted only once when dagrun timed-out. ### How to reproduce Here is DAG that I use to reproduce issue: ` from datetime import timedelta from airflow import DAG from airflow.operators.bash_operator import BashOperator from airflow.utils.dates import days_ago default_args = { 'owner': 'airflow', 'depends_on_past': False, 'start_date': days_ago(1), 'schedule_interval': '@daily', 'email': ['[email protected]'], 'email_on_failure': False, 'email_on_retry': False, 'retries': 0, 'retry_delay': timedelta(minutes=5), } dag = DAG( 'mydag-fail-1', default_args=default_args, description='A simple tutorial DAG', schedule_interval=timedelta(minutes=5), dagrun_timeout=timedelta(seconds=20), max_active_runs=1 ) t1 = BashOperator( task_id='echo_sth', bash_command='echo sth', dag=dag, ) t2 = BashOperator( task_id='sleeping_task', depends_on_past=False, bash_command='sleep 30', dag=dag, ) t1 >> t2 ` Code of my plugin: plugun.py ` from airflow.plugins_manager import AirflowPlugin from dag_run_listener import listener class MyTestListenerPlugin(AirflowPlugin): name = "my_test_listen_plugin" listeners = [listener] ` listener.py ` from __future__ import annotations from typing import TYPE_CHECKING from airflow.listeners import hookimpl from airflow.stats import Stats if TYPE_CHECKING: from airflow.models.dagrun import DagRun @hookimpl def on_dag_run_failed(dag_run: DagRun, msg: str): _emit_metrics_on_dag_run_finished(dag_run=dag_run) def _emit_metrics_on_dag_run_finished(dag_run: DagRun): Stats.incr(f"dagrun.count.{dag_run.dag_id}.{dag_run.state}", 1) ` Or you can add log.info here: https://github.com/apache/airflow/blob/v2-9-stable/airflow/jobs/scheduler_job_runner.py#L1433 and you will see that this block executes twice. ### Operating System cloud composer ### Versions of Apache Airflow Providers _No response_ ### Deployment Google Cloud Composer ### Deployment details _No response_ ### Anything else? If dagrun_timeout is set to 20 seconds and sleep to 30 then TaskInstance will be marked as SKIPPED first but then switch to SUCCESS second emit of FAILED status. If I increase sleep value for example 300(actually works with everything more than 90 seconds) then TaskInstance will stay in the SKIPPED status, but FAILED status of DagRun still emits twice. ### 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
