dagrun_timeout doesn't kill task instances on timeout

[dmgburg]

Apache Airflow version:

1.10.12

What happened:

I created dag with dagrun_timeout=2 minutes.

After 2 minutes dagrun is marked as failed and the next one is started, but task keeps going.

What you expected to happen:

Task is killed with dag run as it is done when you mark dagrun failed manually.

How to reproduce it:

dag = DAG(dag_id='platform.airflow-test',
          description='',
          schedule_interval="0 0 * * *",
          start_date=datetime(2020, 7, 1),
          max_active_runs=1,
          catchup=True,
          dagrun_timeout=timedelta(minutes=2))

run_this = BashOperator(
    task_id='run_after_loop',
    bash_command=' for((i=1;i<=600;i+=1)); do echo "Welcome $i times"; sleep 1;  done',
    dag=dag,
)

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template!

[yogyang]

hit this problem too in airflow 2.0

And it blocked the next Dag_run as

[2021-01-07 04:35:59,145] {scheduler_job.py:1671} INFO - DAG braze_custom_attrs_v7 already has 1 active runs, not queuing any tasks for run 2020-10-21 08:00:00+00:00

I'm not sure if the block of the next Dag_run is caused by the same issue:
#13407

@turbaszek could you help check this?

[RNHTTR]

Would it make sense to get the DAG's task instances and set any tasks to SKIPPED state (or a new state e.g. DAG_TIMED_OUT) if they haven't finished when the DAG state is marked as FAILED?

https://github.com/apache/airflow/blob/master/airflow/jobs/scheduler_job.py#L1694-L1718