Repository: apache/airflow
File: providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/spark_kubernetes.py (Python)

Code before:
                key=lambda p: (
                    p.metadata.deletion_timestamp is None,  # not a terminating pod in pending
                    p.status.phase == PodPhase.SUCCEEDED,  # if the job succeeded while the worker was down
                    p.metadata.creation_timestamp or datetime.min.replace(tzinfo=timezone.utc),
                    p.metadata.name or "",
                ),
            )
            self.log.warning(
                "Found %d Spark driver pods matching labels %s; "
                "selecting pod %s for reattachment based on status.",
                len(pod_list),
                label_selector,
                pod.metadata.name,
            )

        if len(pod_list) == 1:
            pod = pod_list[0]
            self.log.info(
                "Found matching driver pod %s with labels %s", pod.metadata.name, pod.metadata.labels
            )
            self.log.info("`try_number` of task_instance: %s", context["ti"].try_number)
            self.log.info("`try_number` of pod: %s", pod.metadata.labels.get("try_number", "unknown"))
        return pod

    def _get_field_selector(self) -> str:
        return f"status.phase!={PodPhase.FAILED},status.phase!={PodPhase.UNKNOWN}"

    def process_pod_deletion(self, pod, *, reraise=True):
        if pod is not None:
            if self.delete_on_termination:
                pod_name = pod.metadata.name.replace("-driver", "")
                self.log.info("Deleting spark job: %s", pod_name)
                self.launcher.delete_spark_job(pod_name)
            else:
                self.log.info("skipping deleting spark job: %s", pod.metadata.name)

    @cached_property
    def hook(self) -> KubernetesHook:
        hook = KubernetesHook(
            conn_id=self.kubernetes_conn_id,
            in_cluster=self.in_cluster or self.template_body.get("kubernetes", {}).get("in_cluster", False),
            config_file=self.config_file
            or self.template_body.get("kubernetes", {}).get("kube_config_file", None),
            cluster_context=self.cluster_context
            or self.template_body.get("kubernetes", {}).get("cluster_context", None),
        )
        return hook

    @cached_property
    def client(self) -> CoreV1Api:
        return self.hook.core_v1_client

Review comment (bug):
Perhaps you can add a comment here as well as the intent is slightly unintuitive:
`# Exclude terminal failure states to allow Pending, Running, and Succeeded pods.`

Code after:
                    p.status.phase == PodPhase.SUCCEEDED,  # if the job succeeded while the worker was down
                    p.status.phase == PodPhase.PENDING,
                    p.metadata.creation_timestamp or datetime.min.replace(tzinfo=timezone.utc),
                    p.metadata.name or "",
                ),
            )
            self.log.warning(
                "Found %d Spark driver pods matching labels %s; "
                "selecting pod %s for reattachment based on status.",
                len(pod_list),
                label_selector,
                pod.metadata.name,
            )

        if len(pod_list) == 1:
            pod = pod_list[0]
            self.log.info(
                "Found matching driver pod %s with labels %s", pod.metadata.name, pod.metadata.labels
            )
            self.log.info("`try_number` of task_instance: %s", context["ti"].try_number)
            self.log.info("`try_number` of pod: %s", pod.metadata.labels.get("try_number", "unknown"))
        return pod

    def _get_field_selector(self) -> str:
        # exclude terminal failure states, to get only running, pending and succeeded states.
        return f"status.phase!={PodPhase.FAILED},status.phase!={PodPhase.UNKNOWN}"

    def process_pod_deletion(self, pod, *, reraise=True):
        if pod is not None:
            if self.delete_on_termination:
                pod_name = pod.metadata.name.replace("-driver", "")
                self.log.info("Deleting spark job: %s", pod_name)
                self.launcher.delete_spark_job(pod_name)
            else:
                self.log.info("skipping deleting spark job: %s", pod.metadata.name)

    @cached_property
    def hook(self) -> KubernetesHook:
        hook = KubernetesHook(
            conn_id=self.kubernetes_conn_id,
            in_cluster=self.in_cluster or self.template_body.get("kubernetes", {}).get("in_cluster", False),
            config_file=self.config_file
            or self.template_body.get("kubernetes", {}).get("kube_config_file", None),
            cluster_context=self.cluster_context
            or self.template_body.get("kubernetes", {}).get("cluster_context", None),
        )
        return hook

    @cached_property
    def client(self) -> CoreV1Api:
        return self.hook.core_v1_client