From 0aa5a5351c2f6ac0687bfafe134762846347a83a Mon Sep 17 00:00:00 2001 From: rjgoyln Date: Mon, 14 Sep 2026 22:01:21 +0800 Subject: [PATCH] Fail deferrable KPO when a GC'd pod takes its XCom with it A pod reclaimed between the trigger firing and the worker resuming the task takes the XCom sidecar with it. Returning normally in that case reports the task as successful while pushing no return_value, so downstream tasks that template on it fail with an error that points nowhere near the cause and no retry of the downstream can ever recover. Restore the loud failure for tasks that are contracted to produce a value: only a task with do_xcom_push unset can still succeed on a vanished pod, where the pod logs are the only thing lost. closes: #73117 --- .../cncf/kubernetes/operators/pod.py | 15 ++++++++--- .../cncf/kubernetes/operators/test_pod.py | 26 +++++++++++++++++-- 2 files changed, 35 insertions(+), 6 deletions(-) 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):