**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
)
                self.kube_scheduler.delete_pod(pod.metadata.name, pod.metadata.namespace)
    def _change_state(self, key: TaskInstanceKey, state: Optional[str], pod_id: str, namespace: str) -> None:
        if state != State.RUNNING:
            if self.kube_config.delete_worker_pods:
                if not self.kube_scheduler:
                    raise AirflowException(NOT_STARTED_MESSAGE)
                if state != State.FAILED or self.kube_config.delete_worker_pods_on_failure:
                    self.kube_scheduler.delete_pod(pod_id, namespace)
                    self.log.info('Deleted pod: %s in namespace %s', str(key), str(namespace))
            try:
                self.running.remove(key)
            except KeyError:
                self.log.debug('Could not find key: %s', str(key))
        self.event_buffer[key] = state, None
    def try_adopt_task_instances(self, tis: List[TaskInstance]) -> List[TaskInstance]:
        tis_to_flush = [ti for ti in tis if not ti.queued_by_job_id]
        scheduler_job_ids = {ti.queued_by_job_id for ti in tis}
        pod_ids = {ti.key: ti for ti in tis if ti.queued_by_job_id}
        kube_client: client.CoreV1Api = self.kube_client
        for scheduler_job_id in scheduler_job_ids:
            scheduler_job_id = pod_generator.make_safe_label_value(str(scheduler_job_id))
            kwargs = {'label_selector': f'airflow-worker={scheduler_job_id}'}
            pod_list = kube_client.list_namespaced_pod(namespace=self.kube_config.kube_namespace, **kwargs)
            for pod in pod_list.items:
                self.adopt_launched_task(kube_client, pod, pod_ids)
        self._adopt_completed_pods(kube_client)
        tis_to_flush.extend(pod_ids.values())
        return tis_to_flush
    def adopt_launched_task(