**Apache Airflow version**: 2.0.0a2

**Kubernetes version (if you are using kubernetes)** (use `kubectl version`): v1.15.12

**Environment**:
- **OS** (e.g. from /etc/os-release): CentOS 7


**What happened**:

When using KubernetesExecutor, a single SIGTERM doesn't result in the scheduler actually exiting cleanly. However, it will exit after receiving a second SIGTERM.

The same behavior happens if you have an exception in the scheduler. An easy way to replicate it is to raise in `pod_mutation_hook`.

In practice, this means the scheduler doesn't get restarted by K8s after encountering a exception. Or, when terminating the scheduler pod, it sticks around until k8s ultimately sends it a SIGKILL after its grace period is up.

**What you expected to happen**:

The scheduler to exit after a single SIGTERM.

**How to reproduce it**:
The easiest way to reproduce it is to simply send a SIGKILL to your KubernetesExecutor scheduler pod:

```
$ kubectl exec {scheduler_pod_name} kill 1
```

Note, the pod stays running, while the logs report:

```
[2020-10-30 17:34:09,581] {scheduler_job.py:794} INFO - Exiting gracefully upon receiving signal 15
[2020-10-30 17:34:10,586] {process_utils.py:97} INFO - Sending Signals.SIGTERM to GPID 59
[2020-10-30 17:34:10,879] {process_utils.py:63} INFO - Process psutil.Process(pid=59, status='terminated', exitcode=0, started=
'17:33:23') (59) terminated with exit code 0
[2020-10-30 17:34:10,882] {process_utils.py:97} INFO - Sending Signals.SIGTERM to GPID 59
[2020-10-30 17:34:10,882] {scheduler_job.py:1329} INFO - Exited execute loop
```

Issue another `kill`:
```
$ kubectl exec {scheduler_pod_name} kill 1
```

And the pod will be restarted by k8s.

If you can't reproduce it, let me know and I'll try and get something running on kind.

**Anything else we need to know**:

The same behavior happens if you have an exception in the scheduler. An easy way to replicate it is to raise in a `pod_mutation_hook`.


airflow/jobs/scheduler_job.py
)
        for task_instance in tis_to_set_to_scheduled:
            self.executor.queued_tasks.pop(task_instance.key)
        task_instance_str = "\n\t".join(repr(x) for x in tis_to_set_to_scheduled)
        self.log.info("Set the following tasks to scheduled state:\n\t%s", task_instance_str)
    @provide_session
    def _process_executor_events(self, session: Session = None) -> int:
        """Respond to executor events."""
        if not self.processor_agent:
            raise ValueError("Processor agent is not started.")
        ti_primary_key_to_try_number_map: Dict[Tuple[str, str, datetime.datetime], int] = {}
        event_buffer = self.executor.get_event_buffer()
        tis_with_right_state: List[TaskInstanceKey] = []
        
        for ti_key, value in event_buffer.items():
            state: str
            state, _ = value
            
            ti_primary_key_to_try_number_map[ti_key.primary] = ti_key.try_number
            self.log.info(
                "Executor reports execution of %s.%s execution_date=%s "
                "exited with status %s for try_number %s",
                ti_key.dag_id,
                ti_key.task_id,
                ti_key.execution_date,
                state,
                ti_key.try_number,
            )
            if state in (State.FAILED, State.SUCCESS, State.QUEUED):
                tis_with_right_state.append(ti_key)