KubernetesExecutor can start stale worker pods for obsolete TaskInstance UUIDs under HA schedulers

[wolvery]

Under which category would you file this issue?

Airflow Core

Apache Airflow version

3.3.0

What happened and how to reproduce it?

Basically, we got 19 failed retries for example in a single Task all of them are failed because another Scheduler created the same task, so, using HA increases this issue.

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:

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?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct