**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,
)
```
tests/jobs/test_scheduler_job.py
empty_dir = mkdtemp()
        current_process = psutil.Process()
        old_children = current_process.children(recursive=True)
        scheduler = SchedulerJob(subdir=empty_dir, num_runs=1, executor=MockExecutor(do_update=False))
        scheduler.run()
        shutil.rmtree(empty_dir)
        
        current_children = set(current_process.children(recursive=True)) - set(old_children)
        assert not current_children
    @mock.patch('airflow.jobs.scheduler_job.TaskCallbackRequest')
    @mock.patch('airflow.jobs.scheduler_job.Stats.incr')
    def test_process_executor_events(self, mock_stats_incr, mock_task_callback):
        dag_id = "test_process_executor_events"
        dag_id2 = "test_process_executor_events_2"
        task_id_1 = 'dummy_task'
        dag = DAG(dag_id=dag_id, start_date=DEFAULT_DATE, full_filepath="/test_path1/")
        dag2 = DAG(dag_id=dag_id2, start_date=DEFAULT_DATE, full_filepath="/test_path1/")
        task1 = DummyOperator(dag=dag, task_id=task_id_1)
        DummyOperator(dag=dag2, task_id=task_id_1)
        dag.fileloc = "/test_path1/"
        dag2.fileloc = "/test_path1/"
        executor = MockExecutor(do_update=False)
        task_callback = mock.MagicMock()
        mock_task_callback.return_value = task_callback
        scheduler = SchedulerJob(executor=executor)
        scheduler.processor_agent = mock.MagicMock()
        session = settings.Session()
        dag.sync_to_db(session=session)