Scheduler double counts Failed status of dag_run in case when dagrun_timeout was specified.
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 7h
- Merged PRs (30d)
- 484
Description
### 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:
### 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:
```python
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': ['airflow@example.com'],
'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
```python
from airflow.plugins_manager import AirflowPlugin
from dag_run_listener import listener
class MyTestListenerPlugin(AirflowPlugin):
name = "my_test_listen_plugin"
listeners = [listener]
```
listener.py
```python
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)
Contributor guide
Research direction
Start at airflow/jobs/scheduler_job_runner.py around line 1433 and reproduce the report with the provided DAG, dagrun_timeout, and listener callback. Trace why the failed status is emitted twice, then verify the listener observes one failure for a timed-out DagRun and that the reproduction no longer shows duplicate entries.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- backend, data-engineering
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 45/100