**Apache Airflow version**: 2.1.3
**OS**: Custom Docker Image built from python:3.8-slim
**Apache Airflow Provider versions**:
apache-airflow-providers-cncf-kubernetes==2.0.1
apache-airflow-providers-ftp==2.0.0
apache-airflow-providers-http==2.0.0
apache-airflow-providers-imap==2.0.0

**Deployment**: Self managed on EKS, manifests templated and rendered using krane, dags mounted via PVC & EFS

**What happened**: Task remains in queued state 
<img width="1439" alt="Screen Shot 2021-08-18 at 12 17 28 PM" src="https://user-images.githubusercontent.com/47788186/129936170-e16e1362-24ca-4ce9-b2f7-978f2642d388.png">

**What you expected to happen**: Task starts running

**How to reproduce it**: I believe it is because a node is removed. I've attached both the scheduler/k8s executor logs and the kubernetes logs.  I deployed a custom executor with an extra line in the KubernetesJobWatcher.process_status that logged the Event type for the pod.


```
date,name,message
2021-09-08T01:06:35.083Z,KubernetesJobWatcher,Event: <pod_id> had an event of type ADDED
2021-09-08T01:06:35.084Z,KubernetesJobWatcher,Event: <pod_id> Status: Pending
2021-09-08T01:06:35.085Z,KubernetesJobWatcher,Event: <pod_id> had an event of type MODIFIED
2021-09-08T01:06:35.085Z,KubernetesJobWatcher,Event: <pod_id> Status: Pending
2021-09-08T01:06:43.390Z,KubernetesJobWatcher,Event: <pod_id> had an event of type MODIFIED
2021-09-08T01:06:43.390Z,KubernetesJobWatcher,Event: <pod_id> Status: Pending
2021-09-08T01:06:43.390Z,KubernetesJobWatcher,Event: <pod_id> had an event of type MODIFIED
2021-09-08T01:06:43.391Z,KubernetesJobWatcher,Event: <pod_id> Status: Pending
2021-09-08T01:06:45.392Z,KubernetesJobWatcher,Event: <pod_id> had an event of type MODIFIED
2021-09-08T01:06:45.393Z,KubernetesJobWatcher,Event: <pod_id> Status: is Running
2021-09-08T01:07:33.185Z,KubernetesJobWatcher,Event: <pod_id> had an event of type MODIFIED
2021-09-08T01:07:33.186Z,KubernetesJobWatcher,Event: <pod_id> Status: is Running
2021-09-08T01:09:50.478Z,KubernetesJobWatcher,Event: <pod_id> had an event of type MODIFIED
2021-09-08T01:09:50.479Z,KubernetesJobWatcher,Event: <pod_id> Status: is Running
2021-09-08T01:09:50.479Z,KubernetesJobWatcher,Event: <pod_id> had an event of type DELETED
2021-09-08T01:09:50.480Z,KubernetesJobWatcher,Event: <pod_id> Status: is Running

```

Here are the corresponding EKS Logs

```
@timestamp,@message
2021-09-08 01:06:34      1 factory.go:503] pod etl/<pod_id> is already present in the backoff queue
2021-09-08 01:06:43      1 scheduler.go:742] pod etl/<pod_id> is bound successfully on node ""ip-10-0-133-132.ec2.internal"", 19 nodes evaluated, 1 nodes were found feasible."
2021-09-08 01:07:32      1 controller_utils.go:122] Update ready status of pods on node [ip-10-0-133-132.ec2.internal]
2021-09-08 01:07:32      1 controller_utils.go:139] Updating ready status of pod <pod_id> to false
2021-09-08 01:07:38      1 node_lifecycle_controller.go:889] Node ip-10-0-133-132.ec2.internal is unresponsive as of 2021-09-08 01:07:38.017788167
2021-09-08 01:08:51      1 node_tree.go:100] Removed node "ip-10-0-133-132.ec2.internal" in group "us-east-1:\x00:us-east-1b" from NodeTree
2021-09-08 01:08:53      1 node_lifecycle_controller.go:789] Controller observed a Node deletion: ip-10-0-133-132.ec2.internal
2021-09-08 01:09:50      1 gc_controller.go:185] Found orphaned Pod etl/<pod_id> assigned to the Node ip-10-0-133-132.ec2.internal. Deleting.
2021-09-08 01:09:50      1 gc_controller.go:189] Forced deletion of orphaned Pod etl/<pod_id> succeeded

```
**Are you willing to submit a PR?** I think there is a state & event combination missing from the process_state function in the KubernetesJobWatcher. I will submit a PR to fix.

airflow/executors/kubernetes_executor.py
if self.kube_watcher.is_alive():
            self.log.debug("KubeJobWatcher alive, continuing")
        else:
            self.log.error(
                'Error while health checking kube watcher process. Process died for unknown reasons'
            )
            self.kube_watcher = self._make_kube_watcher()
    def run_next(self, next_job: KubernetesJobType) -> None:
        """
        The run_next command will check the task_queue for any un-run jobs.
        It will then create a unique job-id, launch that job in the cluster,
        and store relevant info in the current_jobs map so we can track the job's
        status
        """
        self.log.info('Kubernetes job is %s', str(next_job))
        key, command, kube_executor_config, pod_template_file = next_job
        dag_id, task_id, run_id, try_number = key
        if command[0:3] != ["airflow", "tasks", "run"]:
            raise ValueError('The command must start with ["airflow", "tasks", "run"].')
        base_worker_pod = get_base_pod_from_template(pod_template_file, self.kube_config)
        if not base_worker_pod:
            raise AirflowException(
                f"could not find a valid worker template yaml at {self.kube_config.pod_template_file}"
            )
        pod = PodGenerator.construct_pod(
            namespace=self.namespace,
            scheduler_job_id=self.scheduler_job_id,
            pod_id=create_pod_id(dag_id, task_id),
            dag_id=dag_id,
            task_id=task_id,