**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
):
                self.log.info(
                    "DAG %s already has %d active runs, not queuing any tasks for run %s",
                    dag.dag_id,
                    len(currently_active_runs),
                    dag_run.execution_date,
                )
                return 0
        self._verify_integrity_if_dag_changed(dag_run=dag_run, session=session)
        
        schedulable_tis, callback_to_run = dag_run.update_state(session=session, execute_callbacks=False)
        self._send_dag_callbacks_to_processor(dag_run, callback_to_run)
        
        
        
        
        return dag_run.schedule_tis(schedulable_tis, session)
    @provide_session
    def _verify_integrity_if_dag_changed(self, dag_run: DagRun, session=None):
        """Only run DagRun.verify integrity if Serialized DAG has changed since it is slow"""
        latest_version = SerializedDagModel.get_latest_version_hash(dag_run.dag_id, session=session)
        if dag_run.dag_hash == latest_version:
            self.log.debug("DAG %s not changed structure, skipping dagrun.verify_integrity", dag_run.dag_id)
            return
        dag_run.dag_hash = latest_version
        
        dag_run.dag = self.dagbag.get_dag(dag_id=dag_run.dag_id, session=session)
        
        dag_run.verify_integrity(session=session)
    def _send_dag_callbacks_to_processor(