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

Code before:
        Find an existing Spark driver pod for this task instance.

        The pod is identified using Airflow task context labels. If multiple
        driver pods match the same labels (which can occur if cleanup did not
        run after an abrupt failure), a single pod is selected deterministically
        for reattachment, preferring a Running driver pod when present.
        """
        label_selector = (
            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,
            # preferring Succeeded, then Running (while not in terminating) then 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 on the pod.
            # 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,  # 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

Review comment (bug):
I would tighten this comment like this:

```
# 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.
```

Right now, it sounds a bit unclear and rambly. 

Code after:
        Find an existing Spark driver pod for this task instance.

        The pod is identified using Airflow task context labels. If multiple
        driver pods match the same labels (which can occur if cleanup did not
        run after an abrupt failure), a single pod is selected deterministically
        for reattachment, preferring a Running driver pod when present.
        """
        label_selector = (
            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