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

Code before:
            + ",spark-role=driver"
        )
        # since we did not specify a resource version, we make sure to get the latest data
        # we make sure we get only running or pending pods.
        field_selector = self._get_field_selector()
        pod_list = self.client.list_namespaced_pod(
            self.namespace, label_selector=label_selector, field_selector=field_selector
        ).items

        pod = None
        if len(pod_list) > 1:
            # When multiple pods match the same labels, select one deterministically,
            # preferring a Running or Pending pod, as if another pod was created, it will be in either the
            # terminating status or a terminal phase, if it is in terminating, it will have a
            # deletion_timestamp.
            # pending pods need to also be selected, as what if a driver pod just failed and a new pod is
            # created, we do not want the task to fail.
            pod = max(
                pod_list,
                key=lambda p: (p.metadata.deletion_timestamp is None, 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.RUNNING},status.phase!={PodPhase.PENDING}"

    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

Review comment (suggestion):
Exclusion of terminating pods is a great catch, which was previously overlooked. But why remove the other checks? I would suggest adding this new condition as a priority and keeping the remaining ones. 

Code after:
            self._build_find_pod_label_selector(context, exclude_checked=exclude_checked)
            + ",spark-role=driver"
        )
        # since we did not specify a resource version, we make sure to get the latest data
        # we make sure we get only running or pending pods.
        field_selector = self._get_field_selector()
        pod_list = self.client.list_namespaced_pod(
            self.namespace, label_selector=label_selector, field_selector=field_selector
        ).items

        pod = None
        if len(pod_list) > 1:
            # When multiple pods match the same labels, select one deterministically.
            # Prefer Succeeded, then Running (excluding terminating), then Pending.
            # Terminating pods can be identified via deletion_timestamp.
            # Pending pods are included to handle recent driver restarts without failing the task.
            pod = max(
                pod_list,
                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.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", "")