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

Code before:
        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

    @staticmethod
    def _get_most_recent_pod_index(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):
            pod_start_times: list[datetime.datetime] = [  # type: ignore[no-redef]
                pod_start_time
                if (
                    pod_start_time := pod.to_dict()
                    .get("metadata", {})
                    .get("creation_timestamp", datetime.datetime.now(tz=datetime.timezone.utc))
                )
                else 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
        self.exception = None

    def __enter__(self):
        return self

Review comment (suggestion):
The fallback logic has a flaw. The walrus operator on line 1397 assigns the result of `.get("creation_timestamp", datetime.datetime.now(tz=datetime.timezone.utc))`, which will always return a non-None value due to the default argument in get(). This means the condition on line 1395 (`if pod_start_time`) will always be True, making the `else datetime.datetime.now(tz=datetime.timezone.utc)` branch on line 1401 unreachable.

The code should check if `creation_timestamp` exists before using it, or remove the redundant else clause since the default is already handled by the get() call on line 1399.
```suggestion
            pod_start_times: list[datetime.datetime] = [
                pod.to_dict()
                .get("metadata", {})
                .get("creation_timestamp", datetime.datetime.now(tz=datetime.timezone.utc))
```

Code after:
            )

        labels = self._get_ti_pod_labels(context)
        self.log.info("Building pod %s with labels: %s", pod.metadata.name, labels)

        # Merge Pod Identifying labels with labels passed to operator
        pod.metadata.labels.update(labels)
        # Add Airflow Version to the label
        # And a label to identify that pod is launched by KubernetesPodOperator
        pod.metadata.labels.update(
            {
                "airflow_version": airflow_version.replace("+", "-"),
                "airflow_kpo_in_cluster": str(self.hook.is_in_cluster),
            }
        )
        pod_mutation_hook(pod)
        return pod

    def dry_run(self) -> None:
        """
        Print out the pod definition that would be created by this operator.

        Does not include labels specific to the task instance (since there isn't
        one in a dry_run) and excludes all empty elements.
        """
        pod = self.build_pod_request_obj()
        print(yaml.dump(prune_dict(pod.to_dict(), mode="strict")))

    def process_duplicate_label_pods(self, pod_list: list[k8s.V1Pod]) -> k8s.V1Pod:
        """
        Patch or delete the existing pod with duplicate labels.

        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."""