diff --git a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py index 1656b8a825ead..a2fc60283d2a5 100644 --- a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py +++ b/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: diff --git a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py index 154af0708e242..98e0cc64f4510 100644 --- a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py +++ b/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):