**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,
)
```
airflow/jobs/scheduler_job.py
continue
            active_runs_of_dag = active_runs_of_dags.get(dag.dag_id, 0)
            if dag.max_active_runs and active_runs_of_dag >= dag.max_active_runs:
                self.log.info(
                    "DAG %s is at (or above) max_active_runs (%d of %d), not creating any more runs",
                    dag.dag_id,
                    active_runs_of_dag,
                    dag.max_active_runs,
                )
                dag_model.next_dagrun_create_after = None
            else:
                dag_model.next_dagrun, dag_model.next_dagrun_create_after = dag.next_dagrun_info(
                    dag_model.next_dagrun
                )
    def _schedule_dag_run(
        self,
        dag_run: DagRun,
        currently_active_runs: Set[datetime.datetime],
        session: Session,
    ) -> int:
        """
        Make scheduling decisions about an individual dag run
        ``currently_active_runs`` is passed in so that a batch query can be
        used to ask this for all dag runs in the batch, to avoid an n+1 query.
        :param dag_run: The DagRun to schedule
        :param currently_active_runs: Number of currently active runs of this DAG
        :return: Number of tasks scheduled
        """
        dag = dag_run.dag = self.dagbag.get_dag(dag_run.dag_id, session=session)
        if not dag: