**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("Processing file %s for tasks to queue", file_path)
        try:
            dagbag = DagBag(file_path, include_examples=False, include_smart_sensor=False)
        except Exception:  
            self.log.exception("Failed at reloading the DAG file %s", file_path)
            Stats.incr('dag_file_refresh_error', 1, 1)
            return 0, 0
        if len(dagbag.dags) > 0:
            self.log.info("DAG(s) %s retrieved from %s", dagbag.dags.keys(), file_path)
        else:
            self.log.warning("No viable dags retrieved from %s", file_path)
            self.update_import_errors(session, dagbag)
            return 0, len(dagbag.import_errors)
        self.execute_callbacks(dagbag, callback_requests)
        
        dagbag.sync_to_db()
        if pickle_dags:
            paused_dag_ids = DagModel.get_paused_dag_ids(dag_ids=dagbag.dag_ids)
            unpaused_dags: List[DAG] = [
                dag for dag_id, dag in dagbag.dags.items() if dag_id not in paused_dag_ids
            ]
            for dag in unpaused_dags:
                dag.pickle(session)
        
        try:
            self.update_import_errors(session, dagbag)
        except Exception:  
            self.log.exception("Error logging import errors!")
        return len(dagbag.dags), len(dagbag.import_errors)
class SchedulerJob(BaseJob):