thc1006 opened a new issue, #71743:
URL: https://github.com/apache/airflow/issues/71743
### Under which category would you file this issue?
Providers
### Apache Airflow version
3.3.1
### What happened and how to reproduce it?
**Issue Description**
KPO durable reattach save the pod identity into task state store, but it
only save name and namespace, no uid.
`_persist_pod_identity_to_task_state_store` in
`providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py`:
```python
task_state_store.set(
POD_IDENTIFIER_STATE_KEY,
{"name": pod.metadata.name, "namespace": pod.metadata.namespace},
)
```
and on retry `_get_pod_from_task_state_store` read it back like this:
```python
pod = self.hook.get_pod(name, namespace)
```
In Kubernetes the name is only a slot, it is not identity. After a pod is
deleted, the same name can be take by another pod. Only `metadata.uid` is
stable. So this lookup have no fence, it only trust the name.
Old `find_pod()` path is not like this. `_build_find_pod_label_selector()`
use dag_id, task_id, run_id and map_index, so a pod from other dag run can not
match. But on Airflow 3.3+ `durable` is default True, and once the identity is
persisted the label search is skip. So the fence become weaker than before.
The only guard left is the `already_checked` label, but a pod from another
run does not have this label.
**Steps to reproduce**
Need the pod name can repeat, so `random_name_suffix=False`, or a fixed
`metadata.name` from `pod_template_file` / `full_pod_spec`.
1. dag run A create pod `my-pod`, uid A. store keep `{"name": "my-pod",
"namespace": "..."}`
2. worker die, and later `my-pod` is remove by node drain, gc, cleanup job,
or someone do `kubectl delete pod`
3. dag run B start and create pod `my-pod` again, uid B
4. dag run A retry, `_get_pod_from_task_state_store()` read `my-pod` and get
uid B
I test it on real cluster, k8s v1.37.0-rc.0, with airflow 3.3.1 and
apache-airflow-providers-cncf-kubernetes 10.21.0:
```
STEP 1 dag run A creates its pod (this is what gets persisted)
run_id = manual__2026-08-18T01:00:00+00:00
pod = kpo-uid-test/my-pod
uid = 405af140-31e7-4529-9969-38407688a061
task_state_store['pod_identifier'] = {'name': 'my-pod', 'namespace':
'kpo-uid-test'}
STEP 2 that pod is gone (node drain / GC / kubectl delete / cleanup job)
deleted
STEP 3 dag run B creates a pod, same name (random_name_suffix=False)
run_id = manual__2026-08-18T02:00:00+00:00
pod = kpo-uid-test/my-pod
uid = 366da1af-7790-4c27-9158-d548b002c2be <-- different pod
STEP 4 dag run A retries -> _get_pod_from_task_state_store()
Reconnecting to pod kpo-uid-test/my-pod via identity persisted in task state
store. loc=pod.py:685
Found matching pod my-pod with labels {'dag_id': 'uid_demo',
'kubernetes_pod_operator': 'True', 'run_id':
'manual__2026-08-18t02_00_00_00_00', 'task_id': 'my-pod', 'try_number': '1'}
loc=pod.py:646
`try_number` of task_instance: 2 loc=pod.py:647
`try_number` of pod: 1 loc=pod.py:648
returned pod uid = 366da1af-7790-4c27-9158-d548b002c2be
returned pod run_id = manual__2026-08-18t02_00_00_00_00
*** run A reattached to run B's pod ***
STEP 5 the real caller: get_or_create_pod()
Reconnecting to pod kpo-uid-test/my-pod via identity persisted in task state
store. loc=pod.py:685
Found matching pod my-pod with labels {'dag_id': 'uid_demo',
'kubernetes_pod_operator': 'True', 'run_id':
'manual__2026-08-18t02_00_00_00_00', 'task_id': 'my-pod', 'try_number': '1'}
loc=pod.py:646
Reusing existing pod 'my-pod' (phase=Running, reason=) since it is not
terminated or evicted. loc=pod.py:712
returned pod uid = 366da1af-7790-4c27-9158-d548b002c2be
STEP 6 same situation, but the OLD label search path find_pod()
returned None -> run_id label does not match, run B's pod rejected
```
Step 4 log also show run_id of the pod is run B, and try_number of the pod
is 1 but task instance is 2. The operator print them and still continue.
Step 5 is the real caller. Pod B is Running so `get_or_create_pod()` just
return it. Then run A monitor run B pod, read its log, take its exit code, and
with default `on_finish_action=delete_pod` it delete run B pod in the end. So
run B also break.
<details><summary>script I use</summary>
```python
import subprocess, time, types
from kubernetes import client as k8s, config as k8s_config
from airflow.providers.cncf.kubernetes.operators.pod import
KubernetesPodOperator
NS, POD_NAME, DAG_ID, TASK_ID = "kpo-uid-test", "my-pod", "uid_demo",
"my-pod"
RUN_A = "manual__2026-08-18T01:00:00+00:00"
RUN_B = "manual__2026-08-18T02:00:00+00:00"
def body(run_id):
return {
"apiVersion": "v1", "kind": "Pod",
"metadata": {"name": POD_NAME, "namespace": NS, "labels": {
"dag_id": DAG_ID, "task_id": TASK_ID, "try_number": "1",
"kubernetes_pod_operator": "True",
"run_id": run_id.replace(":", "_").replace("+", "_").lower()}},
"spec": {"restartPolicy": "Never", "containers": [
{"name": "base", "image": "registry.k8s.io/pause:3.9"}]},
}
subprocess.run(["kubectl", "delete", "ns", NS, "--ignore-not-found"])
subprocess.run(["kubectl", "create", "ns", NS])
k8s_config.load_kube_config()
core = k8s.CoreV1Api()
uid_a = core.create_namespaced_pod(NS, body(RUN_A)).metadata.uid
persisted = {"name": POD_NAME, "namespace": NS} # what the provider writes
today
core.delete_namespaced_pod(POD_NAME, NS,
body=k8s.V1DeleteOptions(grace_period_seconds=0))
while True:
try:
core.read_namespaced_pod(POD_NAME, NS); time.sleep(1)
except k8s.rest.ApiException:
break
uid_b = core.create_namespaced_pod(NS, body(RUN_B)).metadata.uid
print("uid A", uid_a, "/ uid B", uid_b)
op = KubernetesPodOperator(task_id=TASK_ID, name=POD_NAME, namespace=NS,
image="registry.k8s.io/pause:3.9",
random_name_suffix=False,
in_cluster=False, kubernetes_conn_id=None,
do_xcom_push=False)
class Store:
def get(self, key, default=None):
return persisted if key == "pod_identifier" else default
context = {"ti": types.SimpleNamespace(dag_id=DAG_ID, task_id=TASK_ID,
map_index=-1, try_number=2),
"run_id": RUN_A, "dag": types.SimpleNamespace(dag_id=DAG_ID),
"task_state_store": Store()}
print("from store :",
op._get_pod_from_task_state_store(context).metadata.uid)
req = k8s.V1Pod(metadata=k8s.V1ObjectMeta(name=POD_NAME, namespace=NS))
print("get_or_create :", op.get_or_create_pod(pod_request_obj=req,
context=context).metadata.uid)
print("find_pod :", op.find_pod(NS, context=context))
```
</details>
### What you think should happen instead?
The uid should be save together with name and namespace, and be check when
read back. If the uid is not the same, treat it like the pod is gone and go to
the label search fallback, same as the 404 case today.
PR #69914 description already say the uid is there:
> On Airflow 3.3+, `durable=True` persists the running pod's identity (name,
namespace, uid) to task state store
But in the merged code the uid is not written, and `operators.rst` say only
(name, namespace). So I think the uid part is just lost somewhere during the
review.
The dev list announce for durable execution also say the criteria is:
> The job has a stable tracking ID that survives the worker process
Other operator persist a server side id, like spark app id, glue job run id
or snowflake query id. In Kubernetes that id is `metadata.uid`. The pod name is
choose by the client and can repeat, so it is not this kind of id.
### Operating System
Ubuntu 26.04 LTS
### Deployment
Virtualenv installation
### Apache Airflow Provider(s)
cncf-kubernetes
### Deployment details
kubeadm cluster, k8s v1.37.0-rc.0, single node.
apache-airflow==3.3.1, apache-airflow-providers-cncf-kubernetes==10.21.0,
kubernetes==36.0.3
### Anything else?
This only happen when the pod name can repeat. Default
`random_name_suffix=True` add 8 random char so it is almost impossible. But
`random_name_suffix=False` is a public parameter and people use it.
#21169 is the same kind of problem before, `random_name_suffix=False` make
KPO delete the wrong pod. It was fixed by #22092 which add a `find_pod` check
before the delete. The durable path go around that check now.
Two more place look like the same problem, but maybe better to do in a
separate PR:
- `KubernetesPodTrigger` only serialize `pod_name` and `pod_namespace`, so
the deferrable path also have no uid
- `PodManager.delete_pod()` use `V1DeleteOptions()` without
`preconditions.uid`
### Are you willing to submit PR?
- [X] Yes I am willing to submit a PR!
### Code of Conduct
- [X] I agree to follow this project's [Code of
Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
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]
