Running tasks marked as skipped on DagRun timeout
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 7h
- Merged PRs (30d)
- 484
Description
### Apache Airflow version
2.5.2
### What happened
Users are experiencing the following:
* A DAG begins to run
* Task(s) go into running state, as expected
* The DagRun times out, marking any currently running task as SKIPPED
* Because tasks are not mark as failed the `on_failure_callback` never gets revoked
Here are some example logs:
```
[2023-03-22, 16:30:02 PDT] {local_task_job.py:266} WARNING - DagRun timed out after 4:00:02.394287.
[2023-03-22, 16:30:07 PDT] {local_task_job.py:266} WARNING - DagRun timed out after 4:00:07.447373.
[2023-03-22, 16:30:07 PDT] {local_task_job.py:272} WARNING - State of this instance has been externally set to skipped. Terminating instance.
[2023-03-22, 16:30:07 PDT] {process_utils.py:129} INFO - Sending Signals.SIGTERM to group 8515. PIDs of all processes in the group: [8515]
```
### What you think should happen instead
Once a DagRun times out, tasks that are currently in RUNNING should be marked as FAILED and downstream tasks should be marked as UPSTREAM_FAILED
### How to reproduce
The following DAG will cause this intermittently
```python
import time
import logging
from airflow.decorators import dag, task
from airflow.utils.dates import datetime, timedelta
@task
def task_1():
import random
pulses = random.randint(5, 10)
for i in range(pulses):
logging.info(f"pulsing: pulse...{i}")
time.sleep(4)
@task
def task_2():
import random
pulses = random.randint(10, 20)
for i in range(pulses):
logging.info(f"pulsing: pulse...{i}")
time.sleep(5)
@task
def downstream_finished_task():
logging.info("task finished")
time.sleep(20)
@dag(dag_id="dagrun_interval_test",
schedule_interval="*/5 * * * *",
start_date=datetime(2023, 3, 23),
dagrun_timeout=timedelta(seconds=30),
catchup=False)
def my_dag():
return [task_1(), task_2()] >> downstream_finished_task()
dag = my_dag()
```
* Running tasks marked skipped
* Downstream left with `no status`
See screenshot

### Operating System
MacOS
### Versions of Apache Airflow Providers
N/A
### Deployment
Astronomer
### Deployment details
Airflow Version 2.5.2
### Anything else
Every time a DagRun times out
### 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 with the DagRun timeout handling shown in local_task_job.py and reproduce the behavior using the DAG in the issue. Trace how running tasks and downstream tasks are updated when the timeout occurs. Done means running tasks become FAILED, downstream tasks become UPSTREAM_FAILED, and the failure callback is invoked.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- data-engineering
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100