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

Code before:
        This is to handle an edge case that can happen only if reattach_on_restart
        flag is False, and the previous run attempt has failed because the task
        process has been killed externally by the cluster or another process.

        If the task process is killed externally, it breaks the code execution and
        immediately exists the task. As a result the pod created in the previous attempt
        will not be properly deleted or patched by cleanup() method.

        Return the newly created pod to be used for the next run attempt.
        """
        new_pod = pod_list.pop(self._get_most_recent_pod_index(pod_list))
        old_pod = pod_list[0]
        self.patch_already_checked(old_pod, reraise=False)
        if self.on_finish_action == OnFinishAction.DELETE_POD:
            self.process_pod_deletion(old_pod)
        return new_pod

    def _get_most_recent_pod_index(self, pod_list: list[k8s.V1Pod]) -> int:
        """Loop through a list of V1Pod objects and get the index of the most recent one."""
        pod_start_times: list[datetime.datetime] = [
            pod.to_dict().get("status").get("start_time") for pod in pod_list
        ]
        if not all(pod_start_times):
            self.log.info(
                "Some of the start_time values are None, switch to creation_timestamp to choose most recent pod"
            )
            pod_start_times: list[datetime.datetime] = [  # type: ignore[no-redef]
                pod.to_dict()
                .get("metadata", {})
                .get("creation_timestamp", datetime.datetime.now(tz=datetime.timezone.utc))
                for pod in pod_list
            ]
        most_recent_start_time = max(pod_start_times)
        return pod_start_times.index(most_recent_start_time)


class _optionally_suppress(AbstractContextManager):
    """
    Returns context manager that will swallow and log exceptions.

    By default swallows descendents of Exception, but you can provide other classes through
    the vararg ``exceptions``.

    Suppression behavior can be disabled with reraise=True.

    :meta private:
    """

    def __init__(self, *exceptions, reraise: bool = False) -> None:
        self._exceptions = exceptions or (Exception,)
        self.reraise = reraise

Review comment (suggestion):
```suggestion
            self.log.info(
                "Unable to determine most recent pod using start_time (some pods have not started yet). Falling back to creation_timestamp from pod metadata."
            )
```

Code after:
        This is to handle an edge case that can happen only if reattach_on_restart
        flag is False, and the previous run attempt has failed because the task
        process has been killed externally by the cluster or another process.

        If the task process is killed externally, it breaks the code execution and
        immediately exists the task. As a result the pod created in the previous attempt
        will not be properly deleted or patched by cleanup() method.

        Return the newly created pod to be used for the next run attempt.
        """
        new_pod = pod_list.pop(self._get_most_recent_pod_index(pod_list))
        old_pod = pod_list[0]
        self.patch_already_checked(old_pod, reraise=False)
        if self.on_finish_action == OnFinishAction.DELETE_POD:
            self.process_pod_deletion(old_pod)
        return new_pod

    def _get_most_recent_pod_index(self, pod_list: list[k8s.V1Pod]) -> int:
        """Loop through a list of V1Pod objects and get the index of the most recent one."""
        pod_start_times: list[datetime.datetime] = [
            pod.to_dict().get("status").get("start_time") for pod in pod_list
        ]
        if not all(pod_start_times):
            self.log.info(
                "Unable to determine most recent pod using start_time (some pods have not started yet). Falling back to creation_timestamp from pod metadata."
            )
            pod_start_times: list[datetime.datetime] = [  # type: ignore[no-redef]
                pod.to_dict()
                .get("metadata", {})
                .get("creation_timestamp", datetime.datetime.now(tz=datetime.timezone.utc))
                for pod in pod_list
            ]
        most_recent_start_time = max(pod_start_times)
        return pod_start_times.index(most_recent_start_time)


class _optionally_suppress(AbstractContextManager):
    """
    Returns context manager that will swallow and log exceptions.

    By default swallows descendents of Exception, but you can provide other classes through
    the vararg ``exceptions``.

    Suppression behavior can be disabled with reraise=True.

    :meta private:
    """

    def __init__(self, *exceptions, reraise: bool = False) -> None:
        self._exceptions = exceptions or (Exception,)
        self.reraise = reraise