kaxil commented on code in PR #69914:
URL: https://github.com/apache/airflow/pull/69914#discussion_r3621751114
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py:
##########
@@ -614,9 +643,53 @@ def log_matching_pod(self, pod: k8s.V1Pod, context:
Context) -> None:
self.log.info("`try_number` of task_instance: %s",
context["ti"].try_number)
self.log.info("`try_number` of pod: %s",
pod.metadata.labels["try_number"])
+ def _get_pod_from_task_state_store(self, context: Context) -> k8s.V1Pod |
None:
+ task_state_store = context.get("task_state_store")
+ if task_state_store is None:
+ return None
+ stored = task_state_store.get(POD_IDENTIFIER_STATE_KEY)
+ if not isinstance(stored, dict):
+ return None
+ name, namespace = stored.get("name"), stored.get("namespace")
+ if not isinstance(name, str) or not isinstance(namespace, str):
+ self.log.warning(
+ "Pod identity stored in task state store is malformed (%s),
falling back to label search.",
+ stored,
+ )
+ return None
+ try:
+ pod = self.hook.get_pod(name, namespace)
Review Comment:
This reconnect fetches the stored pod by name with no `already_checked`
filter, which reintroduces reattach-on-retry that the base path deliberately
prevents.
The task state store is scoped to `(dag_run_id, task_id, map_index, key)`
with no `try_number` (`models/task_state_store.py`: "Retries of the same task
share the same rows"), and rows are cleared only on SUCCESS +
`clear_on_success` (default `False`). So on a retry after failure, the
persisted identity still points at the previous attempt's now-terminal pod, and
`get_pod` returns it. In `get_or_create_pod`, `process_pod_deletion` returns
`False` for `on_finish_action` in {`keep_pod`, `delete_succeeded_pod` (failed
pod), `delete_active_pod` (terminal pod)}, so `if not deleted_pod: return pod`
hands back the dead pod. `execute_sync` then awaits it and re-raises the old
failure, so the retry never actually runs. Manual-clear-and-rerun of a kept
SUCCEEDED pod returns its stale XCom without re-running.
Base avoids this because `cleanup()` stamps `already_checked=True`
(`patch_already_checked` docstring: "ensure we don't reattach on retries") and
`find_pod(exclude_checked=True)` then skips it, forcing a fresh pod. The
durable reconnect drops that guard.
Scope: the pure default (`durable=True` + `on_finish_action="delete_pod"`)
is safe, since the deleted pod 404s and falls back to the label search. It
bites when a keep/conditional-delete action is set (all documented configs).
Suggested fix: in `_get_pod_from_task_state_store`, treat a terminal or
`already_checked`-labelled pod as not reattachable (return `None`) so the
fresh-create path runs and re-persists the new identity. Note
`test_durable_reconnect_terminal_state_runs_existing_handling` forces
`process_pod_deletion.return_value = True`, which skips exactly this branch, so
a `keep_pod` + terminal-pod test asserting a fresh pod is created would catch
it.
##########
providers/cncf/kubernetes/docs/operators.rst:
##########
@@ -187,6 +187,45 @@ for debugging), set ``on_kill_action="keep_pod"``:
The ``termination_grace_period`` parameter is also respected during cleanup,
giving the
pod time to shut down gracefully before being forcefully terminated.
+Durable execution
+^^^^^^^^^^^^^^^^^
+
+If the worker running ``KubernetesPodOperator`` dies while the pod is still
running (e.g. the
+worker is preempted or crashes) and the task is retried, the operator can
reattach to the pod
+that is already running instead of creating a duplicate. This is controlled by
the ``durable``
+parameter, which defaults to ``True``.
+
+On Airflow 3.3+, ``durable=True`` persists the running pod's identity (name,
namespace) to
+:doc:`task state store <apache-airflow:core-concepts/task-state-store>` before
the operator starts
+waiting on it. On retry, the operator reads this identity back and reconnects
directly to that
+specific pod -- no label search, and no possibility of the ambiguity failure
Review Comment:
This overstates it slightly. The reconnect removes the label search only
inside `get_or_create_pod`. `execute_sync` still calls `self.find_pod(...)`
unconditionally right after (to populate `remote_pod`, plus the callback
paths), and `find_pod` raises `FoundMoreThanOnePodFailure` when more than one
unchecked pod matches. So a task with a duplicate orphan pod can still hit the
ambiguity failure on the durable path.
Suggest scoping the wording to something like "the reconnect step uses the
persisted identity instead of a label search" rather than "no possibility of
the ambiguity failure," or reusing the reconnected pod for `remote_pod` so the
claim actually holds end to end.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
