apache / apache/airflow

Scheduler double counts Failed status of dag_run in case when dagrun_timeout was specified.

Open
#55,299 3 comments 0 reactions 0 assignees View on GitHub
_eol_affected_version:2.10 area:core area:scheduler kind:bug needs-triage
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:
Image

### 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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.