cruseakshay · Jan 16, 2026 · Jan 16, 2026 · Jan 16, 2026 · Jan 20, 2026 · Jan 22, 2026
@@ -39,3 +39,19 @@ class KubernetesApiError(AirflowException):
 
 class KubernetesApiPermissionError(AirflowException):
     """Raised when an error is encountered while trying access Kubernetes API."""
+
+
+class PodNotFoundException(AirflowException):
+    """Expected pod does not exist in kube-api."""
+
+
+class PodPreemptedException(AirflowException):
+    """
+    Pod was preempted and is no longer available.
+
+    This exception is raised when a pod returns a 404 error and it was never
+    observed in Running state. This typically happens during node bootstrap
+    when daemonsets preempt task pods on newly created nodes.
+
+    See: https://github.com/apache/airflow/issues/59626
+    """
@@ -35,7 +35,12 @@
 from urllib3.exceptions import HTTPError
 
 from airflow.models import Connection
-from airflow.providers.cncf.kubernetes.exceptions import KubernetesApiError, KubernetesApiPermissionError
+from airflow.providers.cncf.kubernetes.exceptions import (
+    KubernetesApiError,
+    KubernetesApiPermissionError,
+    PodNotFoundException,
+    PodPreemptedException,
+)
 from airflow.providers.cncf.kubernetes.kube_client import _disable_verify_ssl, _enable_tcp_keepalive
 from airflow.providers.cncf.kubernetes.kubernetes_helper_functions import (
     API_TIMEOUT,
@@ -55,6 +60,8 @@
     from kubernetes.client import V1JobList
     from kubernetes.client.models import CoreV1Event, CoreV1EventList, V1Job, V1Pod
 
+    from airflow.providers.cncf.kubernetes.utils.pod_manager import PodPhaseTracker
+
 LOADING_KUBE_CONFIG_FILE_RESOURCE = "Loading Kubernetes configuration file kube_config from {}..."
 
 JOB_FINAL_STATUS_CONDITION_TYPES = {
@@ -914,12 +921,15 @@ async def get_conn(self) -> AsyncGenerator[async_client.ApiClient, None]:
                 await kube_client.close()
 
     @generic_api_retry
-    async def get_pod(self, name: str, namespace: str) -> V1Pod:
+    async def get_pod(
+        self, name: str, namespace: str, *, phase_tracker: PodPhaseTracker | None = None
+    ) -> V1Pod:
         """
         Get pod's object.
 
         :param name: Name of the pod.
         :param namespace: Name of the pod's namespace.
+        :param phase_tracker: Optional tracker for state-aware 404 handling.
         """
         async with self.get_conn() as connection:
             try:
@@ -929,11 +939,43 @@ async def get_pod(self, name: str, namespace: str) -> V1Pod:
                     namespace=namespace,
                 )
                 return pod
+            except async_client.ApiException as e:
+                if e.status == 404:
+                    self._handle_pod_not_found_async(name, namespace, phase_tracker)
+                if e.status == 403:
+                    raise KubernetesApiPermissionError("Permission denied (403) from Kubernetes API.") from e
+                raise KubernetesApiError from e
             except HTTPError as e:
                 if hasattr(e, "status") and e.status == 403:
                     raise KubernetesApiPermissionError("Permission denied (403) from Kubernetes API.") from e
                 raise KubernetesApiError from e
 
+    def _handle_pod_not_found_async(
+        self, name: str, namespace: str, phase_tracker: PodPhaseTracker | None
+    ) -> None:
+        """
+        Handle 404 error with state-aware exception raising.
+
+        :raises PodNotFoundException: If pod was previously running.
+        :raises PodPreemptedException: If pod never reached running.
+        """
+        if phase_tracker and not phase_tracker.is_safe_to_retry_on_404():
+            # Pod was running - don't retry to prevent duplicate execution
+            raise PodNotFoundException(
+                f"Pod '{name}' in namespace '{namespace}' not found. "
+                f"The pod was previously observed in Running state (last phase: "
+                f"{phase_tracker.last_observed_phase}), so this is treated as a "
+                f"terminal failure to prevent duplicate execution of non-idempotent workloads."
+            )
+        # Pod never ran - likely preempted during node bootstrap
+        raise PodPreemptedException(
+            f"Pod '{name}' in namespace '{namespace}' not found. "
+            f"The pod was never observed in Running state, which may indicate "
+            f"preemption by higher-priority pods (e.g., daemonsets) during node bootstrap. "
+            f"Check Kubernetes events: kubectl get events --field-selector "
+            f"reason=Preempted -n {namespace}"
+        )
+
     @generic_api_retry
     async def delete_pod(self, name: str, namespace: str):
         """

@@ -32,6 +32,7 @@
 
 from airflow.configuration import conf
 from airflow.providers.cncf.kubernetes.backcompat import get_logical_date_key
+from airflow.providers.cncf.kubernetes.exceptions import PodPreemptedException
 from airflow.providers.common.compat.sdk import AirflowException
 
 if TYPE_CHECKING:
@@ -64,7 +65,10 @@ class KubernetesApiException(AirflowException):
 
 
 def _should_retry_api(exc: BaseException) -> bool:
-    """Retry on selected ApiException status codes, plus plain HTTP/timeout errors."""
+    """Retry on transient API errors, HTTP errors, and PodPreemptedException."""
+    if isinstance(exc, PodPreemptedException):
+        # Pod was preempted before running - safe to retry
+        return True
     if isinstance(exc, (SyncApiException, AsyncApiException)):
         return exc.status in TRANSIENT_STATUS_CODES
     return isinstance(exc, (HTTPError, KubernetesApiException))

@@ -33,6 +33,7 @@
 
 from airflow.configuration import conf
 from airflow.exceptions import AirflowProviderDeprecationWarning
+from airflow.providers.cncf.kubernetes.exceptions import PodNotFoundException
 from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
 from airflow.providers.cncf.kubernetes.kubernetes_helper_functions import (
     POD_NAME_MAX_LENGTH,
@@ -42,7 +43,7 @@
 from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator
 from airflow.providers.cncf.kubernetes.pod_generator import PodGenerator, merge_objects
 from airflow.providers.cncf.kubernetes.triggers.job import KubernetesJobTrigger
-from airflow.providers.cncf.kubernetes.utils.pod_manager import EMPTY_XCOM_RESULT, PodNotFoundException
+from airflow.providers.cncf.kubernetes.utils.pod_manager import EMPTY_XCOM_RESULT
 from airflow.providers.cncf.kubernetes.version_compat import AIRFLOW_V_3_1_PLUS
 from airflow.providers.common.compat.sdk import AirflowException
 from airflow.utils import yaml

@@ -56,6 +56,7 @@
     convert_volume_mount,
 )
 from airflow.providers.cncf.kubernetes.callbacks import ExecutionMode, KubernetesPodOperatorCallback
+from airflow.providers.cncf.kubernetes.exceptions import PodNotFoundException
 from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
 from airflow.providers.cncf.kubernetes.kubernetes_helper_functions import (
     POD_NAME_MAX_LENGTH,
@@ -75,7 +76,6 @@
     OnFinishAction,
     PodLaunchFailedException,
     PodManager,
-    PodNotFoundException,
     PodPhase,
 )
 from airflow.providers.cncf.kubernetes.version_compat import AIRFLOW_V_3_1_PLUS

@@ -26,13 +26,14 @@
 
 import tenacity
 
-from airflow.providers.cncf.kubernetes.exceptions import KubernetesApiPermissionError
+from airflow.providers.cncf.kubernetes.exceptions import KubernetesApiPermissionError, PodPreemptedException
 from airflow.providers.cncf.kubernetes.hooks.kubernetes import AsyncKubernetesHook
 from airflow.providers.cncf.kubernetes.utils.pod_manager import (
     AsyncPodManager,
     OnFinishAction,
     PodLaunchTimeoutException,
     PodPhase,
+    PodPhaseTracker,
 )
 from airflow.triggers.base import BaseTrigger, TriggerEvent
 
@@ -122,6 +123,8 @@ def __init__(
         self.on_finish_action = OnFinishAction(on_finish_action)
         self.trigger_kwargs = trigger_kwargs or {}
         self._since_time = None
+        # Track pod phase to enable safe retry on 404 errors (preemption handling)
+        self._phase_tracker = PodPhaseTracker()
 
     def serialize(self) -> tuple[str, dict[str, Any]]:
         """Serialize KubernetesCreatePodTrigger arguments and classpath."""
@@ -211,6 +214,25 @@ async def run(self) -> AsyncIterator[TriggerEvent]:
                 }
             )
             return
+        except PodPreemptedException as e:
+            # Pod was preempted and could not be recovered after retries
+            message = (
+                f"Pod preemption detected: {e}. "
+                "The pod was likely preempted by higher-priority pods (e.g., daemonsets) "
+                "during node bootstrap and could not be rescheduled after multiple retries. "
+                "Consider increasing task retries or investigating cluster capacity."
+            )
+            self.log.warning(message)
+            yield TriggerEvent(
+                {
+                    "name": self.pod_name,
+                    "namespace": self.pod_namespace,
+                    "status": "failed",
+                    "message": message,
+                    **self.trigger_kwargs,
+                }
+            )
+            return
         except Exception as e:
             self.log.exception(
                 "Unexpected error while waiting for pod %s in namespace %s",
@@ -314,8 +336,15 @@ async def _wait_for_container_completion(self) -> TriggerEvent:
 
     @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
     async def _get_pod(self) -> V1Pod:
-        """Get the pod from Kubernetes with retries."""
-        pod = await self.hook.get_pod(name=self.pod_name, namespace=self.pod_namespace)
+        """Get the pod from Kubernetes with retries and phase tracking."""
+        pod = await self.hook.get_pod(
+            name=self.pod_name,
+            namespace=self.pod_namespace,
+            phase_tracker=self._phase_tracker,
+        )
+        # Update phase tracker with observed pod state
+        if pod:
+            self._phase_tracker.update_from_pod(pod)
         # Due to AsyncKubernetesHook overriding get_pod, we need to cast the return
         # value to kubernetes_asyncio.V1Pod, because it's perceived as different type
         return cast("V1Pod", pod)

@@ -38,6 +38,7 @@
 from urllib3.exceptions import HTTPError, TimeoutError
 
 from airflow.providers.cncf.kubernetes.callbacks import ExecutionMode, KubernetesPodOperatorCallback
+from airflow.providers.cncf.kubernetes.exceptions import PodNotFoundException, PodPreemptedException
 from airflow.providers.cncf.kubernetes.kubernetes_helper_functions import (
     KubernetesApiException,
     PodLaunchFailedException,
@@ -208,8 +209,28 @@ class PodLaunchTimeoutException(AirflowException):
     """When pod does not leave the ``Pending`` phase within specified timeout."""
 
 
-class PodNotFoundException(AirflowException):
-    """Expected pod does not exist in kube-api."""
+@dataclass
+class PodPhaseTracker:
+    """
+    Track whether a pod ever reached Running state for safe 404 retry handling.
+
+    If pod never reached Running: safe to retry (likely preemption).
+    If pod was Running: NOT safe to retry (may cause duplicate execution).
+    """
+
+    ever_reached_running: bool = False
+    last_observed_phase: str | None = None
+
+    def update_from_pod(self, pod: V1Pod) -> None:
+        """Update tracker state from observed pod status."""
+        if pod.status and pod.status.phase:
+            self.last_observed_phase = pod.status.phase
+            if pod.status.phase == PodPhase.RUNNING:
+                self.ever_reached_running = True
+
+    def is_safe_to_retry_on_404(self) -> bool:
+        """Return True if pod never reached Running (safe to retry)."""
+        return not self.ever_reached_running
 
 
 class PodLogsConsumer:
@@ -811,13 +832,54 @@ def read_pod_events(self, pod: V1Pod, resource_version: str | None = None) -> Co
             raise KubernetesApiException(f"There was an error reading the kubernetes API: {e}")
 
     @generic_api_retry
-    def read_pod(self, pod: V1Pod) -> V1Pod:
-        """Read POD information."""
+    def read_pod(self, pod: V1Pod, *, phase_tracker: PodPhaseTracker | None = None) -> V1Pod:
+        """
+        Read POD information.
+
+        :param pod: The pod object to read.
+        :param phase_tracker: Optional tracker to monitor pod lifecycle state.
+            When provided, enables state-aware retry behavior on 404 errors
+            to handle pod preemption during node bootstrap.
+        """
         try:
-            return self._client.read_namespaced_pod(pod.metadata.name, pod.metadata.namespace)
+            result = self._client.read_namespaced_pod(pod.metadata.name, pod.metadata.namespace)
+            if phase_tracker:
+                phase_tracker.update_from_pod(result)
+            return result
+        except ApiException as e:
+            if e.status == 404:
+                self._handle_pod_not_found(pod, phase_tracker)
+            raise
         except HTTPError as e:
             raise KubernetesApiException(f"There was an error reading the kubernetes API: {e}")
 
+    def _handle_pod_not_found(self, pod: V1Pod, phase_tracker: PodPhaseTracker | None) -> None:
+        """
+        Handle 404 error with state-aware exception raising.
+
+        :raises PodNotFoundException: If pod was previously running.
+        :raises PodPreemptedException: If pod never reached running.
+        """
+        pod_name = pod.metadata.name
+        namespace = pod.metadata.namespace
+
+        if phase_tracker and not phase_tracker.is_safe_to_retry_on_404():
+            # Pod was running - don't retry to prevent duplicate execution
+            raise PodNotFoundException(
+                f"Pod '{pod_name}' in namespace '{namespace}' not found. "
+                f"The pod was previously observed in Running state (last phase: "
+                f"{phase_tracker.last_observed_phase}), so this is treated as a "
+                f"terminal failure to prevent duplicate execution of non-idempotent workloads."
+            )
+        # Pod never ran - likely preempted during node bootstrap
+        raise PodPreemptedException(
+            f"Pod '{pod_name}' in namespace '{namespace}' not found. "
+            f"The pod was never observed in Running state, which may indicate "
+            f"preemption by higher-priority pods (e.g., daemonsets) during node bootstrap. "
+            f"Check Kubernetes events: kubectl get events --field-selector "
+            f"reason=Preempted -n {namespace}"
+        )
+
     def await_xcom_sidecar_container_start(
         self, pod: V1Pod, timeout: int = 900, log_interval: int = 30
     ) -> None:
@@ -1007,12 +1069,23 @@ def __init__(
         self._callbacks = callbacks or []
         self.stop_watching_events = False
 
-    async def read_pod(self, pod: V1Pod) -> V1Pod:
-        """Read POD information."""
-        return await self._hook.get_pod(
+    async def read_pod(self, pod: V1Pod, *, phase_tracker: PodPhaseTracker | None = None) -> V1Pod:
+        """
+        Read POD information.
+
+        :param pod: The pod object to read.
+        :param phase_tracker: Optional tracker to monitor pod lifecycle state.
+            When provided, enables state-aware retry behavior on 404 errors
+            to handle pod preemption during node bootstrap.
+        """
+        result = await self._hook.get_pod(
             pod.metadata.name,
             pod.metadata.namespace,
+            phase_tracker=phase_tracker,
         )
+        if phase_tracker and result:
+            phase_tracker.update_from_pod(result)
+        return result
 
     async def read_pod_events(self, pod: V1Pod, resource_version: str | None = None) -> CoreV1EventList:
         """Get pod's events."""