Fail deferrable KPO when a GC'd pod takes its XCom with it

providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py

@@ -1116,11 +1116,18 @@ def trigger_reentry(self, context: Context, event: dict[str, Any]) -> Any:
                 pod_name,
             )
             if event["status"] == "success":
-                # Trigger already observed the pod completed successfully;
-                # logs/XCom are unrecoverable but the task itself succeeded.
-                return
+                if not self.do_xcom_push:
+                    # The pod ran to completion and the task returns nothing, so only
+                    # its logs are lost.
+                    return
+                # The XCom sidecar went with the pod, so the value the task is contracted
+                # to produce is unrecoverable — fail so a retry re-runs the pod instead of
+                # leaving downstreams to pull a missing XCom.
+                reason = " — its XCom result can no longer be retrieved"
+            else:
+                reason = ""
             raise PodNotFoundException(
-                f"Pod {pod_namespace}/{pod_name} not found after resuming from deferral"
+                f"Pod {pod_namespace}/{pod_name} not found after resuming from deferral{reason}"
             ) from e
 
         if not self.pod:

providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py

@@ -4009,9 +4009,9 @@ def test_skip_deferral_on_terminated_pod(
     @patch(POD_MANAGER_CLASS)
     @patch(HOOK_CLASS)
     def test_async_trigger_reentry_returns_when_pod_gcd_on_success(self, mocked_hook, mock_manager):
-        """Pod GC'd between trigger firing and reentry should not fail a successful task."""
+        """Pod GC'd between trigger firing and reentry should not fail a successful XCom-less task."""
         mocked_hook.return_value.get_pod.side_effect = ApiException(status=404, reason="Not Found")
-        k = KubernetesPodOperator(task_id="task", deferrable=True)
+        k = KubernetesPodOperator(task_id="task", deferrable=True, do_xcom_push=False)
         context = create_context(k)
         context["ti"] = MagicMock()
 
@@ -4030,6 +4030,28 @@ def test_async_trigger_reentry_returns_when_pod_gcd_on_success(self, mocked_hook
         # await_pod_completion in _clean must not be called with self.pod=None
         mock_manager.return_value.await_pod_completion.assert_not_called()
 
+    @patch(POD_MANAGER_CLASS)
+    @patch(HOOK_CLASS)
+    def test_async_trigger_reentry_raises_when_pod_gcd_on_success_with_xcom_push(
+        self, mocked_hook, mock_manager
+    ):
+        """A GC'd pod fails an XCom-producing task rather than succeeding with no return_value."""
+        mocked_hook.return_value.get_pod.side_effect = ApiException(status=404, reason="Not Found")
+        k = KubernetesPodOperator(task_id="task", deferrable=True, do_xcom_push=True)
+        context = create_context(k)
+        context["ti"] = MagicMock()
+
+        with pytest.raises(PodNotFoundException, match="XCom result can no longer be retrieved"):
+            k.trigger_reentry(
+                context=context,
+                event={
+                    "status": "success",
+                    "message": TEST_SUCCESS_MESSAGE,
+                    "name": TEST_NAME,
+                    "namespace": TEST_NAMESPACE,
+                },
+            )
+
     @patch(POD_MANAGER_CLASS)
     @patch(HOOK_CLASS)
     def test_async_trigger_reentry_raises_pod_not_found_on_failure(self, mocked_hook, mock_manager):