**Apache Airflow version**: 2.0.0rc2
**Environment**: Breeze with example dags, Python 3.8 postgres.
- **OS** (e.g. from /etc/os-release): Linux
- **Kernel** (e.g. `uname -a`): Breeze CI image
- **Install tools**: Breeze: 
- **Executor**: LocalExecutor
```
./breeze start-airflow --backend postgres --load-example-dags --load-default-connections --install-airflow-version 2.0.0rc2 --skip-mounting-local-sources --python 3.8
```

**What happened**:

When testing airflow logging I occasionally stumble upon "KeyError' from `dag_procesing.py`. I am not sure exactly when it happens. It's not always reproducible but it looks like it is when I restart scheduler and trigger 'example_bash_operator.py" it happens rather randomly (1/10 times more or less). It does not happen always when I triggere task manually. DAG gets correctly executed after triggering, but the log is there and warniing printed in the logs right after the DAG finishes execution.

The error I see in scheduler's logs:

```
[2020-12-13 19:35:33,752] {dagbag.py:440} INFO - Filling up the DagBag from /usr/local/lib/python3.8/site-packages/airflow/example_dags/example_bash_operator.py                                                                                                                                          
Running <TaskInstance: example_bash_operator.run_after_loop 2020-12-13T19:35:30.648020+00:00 [queued]> on host 6611da4b1a27
[2020-12-13 19:35:34,517] {dagrun.py:444} INFO - Marking run <DagRun example_bash_operator @ 2020-12-13 19:35:30.648020+00:00: manual__2020-12-13T19:35:30.648020+00:00, externally triggered: True> successful                                                                                           
[2020-12-13 19:35:34,523] {scheduler_job.py:1193} INFO - Executor reports execution of example_bash_operator.run_after_loop execution_date=2020-12-13 19:35:30.648020+00:00 exited with status success for try_number 1                                                                                   
Process ForkProcess-34:
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/multiprocessing/process.py", line 315, in _bootstrap
    self.run()
  File "/usr/local/lib/python3.8/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/local/lib/python3.8/site-packages/airflow/utils/dag_processing.py", line 365, in _run_processor_manager
    processor_manager.start()
  File "/usr/local/lib/python3.8/site-packages/airflow/utils/dag_processing.py", line 596, in start
    return self._run_parsing_loop()
  File "/usr/local/lib/python3.8/site-packages/airflow/utils/dag_processing.py", line 659, in _run_parsing_loop
    self._processors.pop(processor.file_path)
KeyError: '/usr/local/lib/python3.8/site-packages/airflow/example_dags/example_bash_operator.py'
[2020-12-13 19:35:35,589] {dag_processing.py:396} WARNING - DagFileProcessorManager (PID=1029759) exited with exit code 1 - re-launching
```

**What you expected to happen**:

No error in logs.

**How to reproduce it**:

```
./breeze start-airflow --backend postgres --load-example-dags --load-default-connections --install-airflow-version 2.0.0rc2 --skip-mounting-local-sources --python 3.8
```

Login to the webserver, enable 'example_bash_operator", wait for it to execute. Trigger the example DAG several times (always wait for the end of execution. It happens randomly (for me around 1/10 tasks)


**Anything else we need to know**:

tests/utils/test_dag_processing.py
from airflow.jobs.scheduler_job import SchedulerJob
        dag_id = 'exit_test_dag'
        dag_directory = os.path.normpath(os.path.join(TEST_DAG_FOLDER, os.pardir, "dags_with_system_exit"))
        
        clear_db_dags()
        clear_db_serialized_dags()
        child_pipe, parent_pipe = multiprocessing.Pipe()
        manager = DagFileProcessorManager(
            dag_directory=dag_directory,
            dag_ids=[],
            max_runs=1,
            processor_factory=SchedulerJob._create_dag_file_processor,
            processor_timeout=timedelta(seconds=5),
            signal_conn=child_pipe,
            pickle_dags=False,
            async_mode=True,
        )
        manager._run_parsing_loop()
        while parent_pipe.poll(timeout=None):
            result = parent_pipe.recv()
            if isinstance(result, DagParsingStat) and result.done:
                break
        
        assert len(result.file_paths) == 3
        with create_session() as session:
            assert session.query(DagModel).get(dag_id) is not None
class TestDagFileProcessorAgent(unittest.TestCase):