wolvery opened a new issue, #69760:
URL: https://github.com/apache/airflow/issues/69760
### Under which category would you file this issue?
Airflow Core
### Apache Airflow version
3.3.0
### What happened and how to reproduce it?
When running Airflow 3.3.0 with `KubernetesExecutor` and multiple scheduler
replicas, we observed stale Kubernetes worker pods starting after the
corresponding `TaskInstance` row had already been replaced/retried.
The stale worker pod then calls:
```http
PATCH /execution/task-instances/{task_instance_id}/run
and the execution API returns:
{
"detail": {
"reason": "not_found",
"message": "Task Instance not found"
}
}
The worker logs this as a generic ServerResponseError.
The important detail is that the missing {task_instance_id} is an immutable
TaskInstance.id UUID from an older queued workload. Looking up the same logical
task by (dag_id, task_id, run_id, map_index) shows a newer TaskInstance.id with
a higher try_number already queued.
Observed sequence:
1. Scheduler A queues a task instance with UUID A, try 1.
2. Kubernetes pod creation/start is delayed, for example due to cluster
quota/resource pressure.
3. Another scheduler/retry path moves the same logical task forward and
queues a replacement task instance with UUID B, try 2.
4. The old Kubernetes pod for UUID A eventually starts.
5. The worker calls /execution/task-instances/A/run.
6. The API returns 404 not_found because UUID A no longer exists.
7. The worker fails during startup with ServerResponseError.
This appears to be a race between the scheduler/executor queue, Kubernetes
pod startup latency, and HA scheduler retry/reconciliation.
Relevant source behavior in Airflow 3.3.0:
- The /execution/task-instances/{task_instance_id}/run route looks up the
row by immutable TaskInstance.id.
- KubernetesExecutor builds an ExecuteTask workload containing the current
TaskInstance.id.
- There does not appear to be a DB revalidation immediately before
AirflowKubernetesScheduler.run_next(...) creates the Kubernetes pod.
- Kubernetes pod annotations and watcher/adoption paths primarily use
logical task identity (dag_id, task_id, run_id, map_index, try_number), not the
immutable TaskInstance.id or a launch fencing token.
- TaskInstanceOperations.start() only special-cases 409 invalid_state with
previous_state=running; it does not handle 404 not_found as a stale/obsolete
worker.
A minimal reproduction should be possible with:
1. Airflow 3.3.0.
2. KubernetesExecutor.
3. At least two scheduler replicas.
4. A high-fanout DAG that creates many Kubernetes worker pods.
5. Artificial pod startup delay or Kubernetes quota/resource pressure.
6. Retries enabled.
The failure becomes visible when an old pending pod starts after the logical
task has already been retried/requeued with a different TaskInstance.id.
### What you think should happen instead?
A stale Kubernetes worker pod should not be able to start an obsolete task
instance.
At minimum, before creating a Kubernetes pod from a serialized executor
workload, KubernetesExecutor should revalidate that the workload is still
current in the metadata DB, for example:
TaskInstance.id == workload.ti.id
TaskInstance.state == QUEUED
TaskInstance.try_number == workload.ti.try_number
TaskInstance.queued_by_job_id == current scheduler job id
If the row is missing or no longer matches, the executor should drop the
stale workload and not create the pod.
A more robust fix would be to introduce/propagate a launch fencing token,
possibly using external_executor_id or a dedicated launch generation:
1. Scheduler queues the task with a fresh launch token.
2. Kubernetes pod annotations include:
- TaskInstance.id
- launch token / external_executor_id
- try_number
- scheduler_job_id
3. Worker /run, Kubernetes watcher, and pod adoption paths validate that the
pod token still matches the current DB row.
4. Stale pods are ignored/deleted instead of being adopted or allowed to
update task state.
This follows the usual distributed-systems pattern of fencing tokens/CAS
claims for delayed workers.
The API returning 404 for a missing immutable TaskInstance.id seems
reasonable. The issue is that the executor can still launch stale work after
the DB state has moved on.
### Operating System
Linux container image based on the official Apache Airflow image.
### Deployment
Other Docker-based deployment
### Apache Airflow Provider(s)
cncf-kubernetes
### Versions of Apache Airflow Providers
apache-airflow-providers-cncf-kubernetes==10.19.0
apache-airflow-task-sdk==1.3.0
### Official Helm Chart version
Not Applicable
### Kubernetes Version
_No response_
### Helm Chart configuration
The issue was observed with a vanilla Airflow 3.3.0 image:
apache/airflow:3.3.0-python3.10
### Docker Image customizations
_No response_
### Anything else?
Related behavior seems similar to prior discussions around
KubernetesExecutor and multiple schedulers, especially stale/duplicate task
execution during scheduler failover or retry.
This is not the same as terminal /state duplicate update handling. The
observed failure happens earlier, during worker startup, at:
PATCH /execution/task-instances/{task_instance_id}/run
with:
404 Task Instance not found
Potential patch areas:
- providers/cncf/kubernetes/.../kubernetes_executor.py
- preflight DB validation before pod creation
- providers/cncf/kubernetes/.../kubernetes_executor_utils.py
- include immutable TI identity / launch token in pod annotations
- airflow-core/.../execution_api/routes/task_instances.py
- optional launch-token validation on /run
- task-sdk/.../api/client.py
- classify /run 404 not_found as stale worker startup rather than opaque
ServerResponseError
- scheduler/DagRun scheduling paths
- ensure CAS-style transitions so concurrent schedulers cannot advance the
same logical task unexpectedly
### 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]
