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

Code before:
            startup_timeout=self.startup_timeout_seconds,
            startup_check_interval=self.startup_check_interval_seconds,
            schedule_timeout=self.schedule_timeout_seconds,
            base_container_name=self.base_container_name,
            on_finish_action=self.on_finish_action.value,
            last_log_time=last_log_time,
            logging_interval=self.logging_interval,
            trigger_kwargs=self.trigger_kwargs,
        )
        container_state = trigger.define_container_state(self.pod)
        if container_state == ContainerState.TERMINATED or container_state == ContainerState.FAILED:
            self.log.info("Skipping deferral as pod is already in a terminal state")
            self.trigger_reentry(
                context=context,
                event={
                    "status": "success" if container_state == ContainerState.TERMINATED else "failed",
                    "namespace": self.pod.metadata.namespace,
                    "name": self.pod.metadata.name,
                    "message": "Container succeeded"
                    if container_state == ContainerState.TERMINATED
                    else "Container failed",
                    "last_log_time": last_log_time,
                    **(self.trigger_kwargs or {}),
                },
            )
            return

        self.defer(trigger=trigger, method_name="trigger_reentry")

    def trigger_reentry(self, context: Context, event: dict[str, Any]) -> Any:
        """
        Point of re-entry from trigger.

        If ``logging_interval`` is None, then at this point, the pod should be done, and we'll just fetch
        the logs and exit.

        If ``logging_interval`` is not None, it could be that the pod is still running, and we'll just
        grab the latest logs and defer back to the trigger again.
        """
        self.pod = None
        xcom_sidecar_output = None
        try:
            pod_name = event["name"]
            pod_namespace = event["namespace"]

            self.pod = self.hook.get_pod(pod_name, pod_namespace)

            if not self.pod:
                raise PodNotFoundException("Could not find pod after resuming from deferral")

            follow = self.logging_interval is None

Review comment (suggestion):
I would write this as an if else statement instead of return, because you actually don't return anything in this case, but that's a matter of taste, don't now what the consensus is in this case for Airflow code base.

Code after:
        """Redefine triggers which are being used in child classes."""
        self.convert_config_file_to_dict()

        connection_extras = None
        if self.kubernetes_conn_id:
            try:
                conn = BaseHook.get_connection(self.kubernetes_conn_id)
            except AirflowNotFoundException:
                self.log.warning(
                    "Could not resolve connection extras for deferral: connection `%s` not found. "
                    "Triggerer will try to resolve it from its own environment.",
                    self.kubernetes_conn_id,
                )
            else:
                connection_extras = conn.extra_dejson
                self.log.info("Successfully resolved connection extras for deferral.")

        trigger_start_time = datetime.datetime.now(tz=datetime.timezone.utc)

        trigger = KubernetesPodTrigger(
            pod_name=self.pod.metadata.name,  # type: ignore[union-attr]
            pod_namespace=self.pod.metadata.namespace,  # type: ignore[union-attr]
            trigger_start_time=trigger_start_time,
            kubernetes_conn_id=self.kubernetes_conn_id,
            connection_extras=connection_extras,
            cluster_context=self.cluster_context,
            config_dict=self._config_dict,
            in_cluster=self.in_cluster,
            poll_interval=self.poll_interval,
            get_logs=self.get_logs,
            startup_timeout=self.startup_timeout_seconds,
            startup_check_interval=self.startup_check_interval_seconds,
            schedule_timeout=self.schedule_timeout_seconds,
            base_container_name=self.base_container_name,
            on_finish_action=self.on_finish_action.value,
            last_log_time=last_log_time,
            logging_interval=self.logging_interval,
            trigger_kwargs=self.trigger_kwargs,
        )
        container_state = trigger.define_container_state(self.pod) if self.pod else None
        if context and (
            container_state == ContainerState.TERMINATED or container_state == ContainerState.FAILED
        ):
            self.log.info("Skipping deferral as pod is already in a terminal state")
            self.trigger_reentry(
                context=context,
                event={
                    "status": "failed" if container_state == ContainerState.FAILED else "success",
                    "namespace": trigger.pod_namespace,
                    "name": trigger.pod_name,
                    "message": "Container failed"