Validate KubernetesExecutor launches with external executor IDs

[wolvery]

What

Fixes #69760.

This adds DB-backed launch-token validation for KubernetesExecutor task launches using Airflow's existing external_executor_id field:

• opt KubernetesExecutor into scheduler pre-assigned external_executor_id launch tokens
• serialize the launch token into Airflow 3 ExecuteTask workloads and worker /run requests
• preserve the launch token in KubernetesExecutor queued events so the scheduler does not overwrite it with the scheduler job id
• annotate Kubernetes pods with the launch token for observability and future reconciliation checks
• reject stale /execution/task-instances/{id}/run requests when the current DB token no longer matches the worker/pod token
• revalidate queued Kubernetes jobs against the current DB row/token immediately before pod creation

This prevents a stale Kubernetes workload from validly transitioning an obsolete task-instance launch to RUNNING after a newer queued launch has superseded it.

Token semantics

This is equality-based launch-token validation, not a classic monotonic fencing token.

Airflow already has a scheduler path for executors that set pre_assigns_external_executor_id = True. When such a task instance is queued, the scheduler writes a fresh external_executor_id UUID to the task_instance row. This PR makes KubernetesExecutor use that existing capability and preserve that token through the Kubernetes workload lifecycle.

The intended invariant is:

A worker may start a task only when:
  request.ti_id == current task_instance.id
  request.external_executor_id == current task_instance.external_executor_id
  request.try_number == current task_instance.try_number
  current task_instance.state is QUEUED or RESTARTING

So this is a per-launch ownership token / capability token. It is rotated when Airflow queues a new launch by assigning a new external_executor_id; it is not ordered, and it should not be interpreted as a monotonic distributed-systems fencing counter.

Behavior notes

• Already-created stale pods: This cannot prevent a Kubernetes pod that was already created from calling the API after its immutable task-instance UUID has been deleted/replaced. In that case /run still returns 404 not_found; the worker should treat that as a stale launch and exit without running user code.
• Stale-token rejection: If the task-instance row still exists but now belongs to a different launch token, /run returns 409 stale_executor_launch and the worker should treat it as stale.
• Queued event ownership: KubernetesExecutor must publish the launch token as queued event info for ExecuteTask workloads. Otherwise the scheduler interprets the event info as external_executor_id and can overwrite the UUID launch token with the numeric scheduler job id.
• Stale-drop recovery: When a stale Kubernetes workload is dropped before pod creation, its task instance is left in place rather than failed. The newer launch's own workload creates the pod, or the scheduler's queued-task timeout / task-instance adoption reclaims the row. Failing it here could clobber a newer launch.
• Rolling upgrades: Token validation only applies when the DB row has an external_executor_id. Rows without a token keep the existing state-based validation so rolling upgrades and legacy queued work are not rejected solely because an older worker omitted the new field.

Tests

• python3 -m compileall ... on modified source and test files
• uv run ruff check ...
• uv run ruff format --check ...
• git diff --check
• pytest (isolated sqlite backend):
  • airflow-core/.../execution_api/versions/head/test_task_instances.py -k ti_run — 25 passed, 1 skipped (postgres-only)
  • providers/cncf/kubernetes/.../executors/test_kubernetes_executor.py -k "sync or run_next or stale or external_executor" — all passed
  • Covers stale-token rejection, matching-token acceptance, missing-token acceptance (old clients), matching-token pod creation, and preserving the launch token in queued executor events.

Gen-AI disclosure

• Yes: GPT was used to assist in authoring this PR. The changes were reviewed and validated by the contributor before submission.

Generated-by: GPT following the guidelines: https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions

---

Important

🛠️ Maintainer triage note for @wolvery · by @potiuk · 2026-07-15 16:00 UTC

Helpful heads-up from the maintainers — please address before this PR can be reviewed (see our Pull Request quality criteria):

• ❌ Pre-commit / static checks. See docs.
• ❌ Provider tests. See docs.

The ball is in your court — you've been assigned to this PR. Fix the above, then mark it Ready for review.

_{Automated triage — may be imperfect; a maintainer takes the next look.}

[wolvery]

Hey! I've tested it locally and it seems to have no other race condition happening.
For detecting the issue, you need to use not_found as a log filter:

Before 11, I was running Airflow without this patch, and after 11, I switched off this fix.

[potiuk]

@wolvery This PR has been converted to draft because it does not yet meet our Pull Request quality criteria.

Issues found:

• ❌ Pre-commit / static checks: Failing: CI image checks / Static checks. Run prek run --from-ref main --stage pre-commit locally to reproduce and fix.
• ❌ Provider tests: Failing: Provider distributions tests / Compat 2.11.1:P3.10:, Provider distributions tests / Compat 3.0.6:P3.10:, Provider distributions tests / Compat 3.1.8:P3.10:. Run breeze testing providers-tests --test-type "Providers[<provider>]" locally.

Note: Your branch is 158 commits behind main. Some check failures may be caused by changes in the base branch rather than by your PR. Please rebase your branch and push again to get up-to-date CI results.

What to do next:

• Fix each issue listed above.
• Make sure static checks pass locally (prek run --from-ref main --stage pre-commit).
• Mark the PR as "Ready for review" when you're done.

Converting a PR to draft is not a rejection — it is an invitation to bring the PR up to the project's standards so that maintainer review time is spent productively. There is no rush — take your time and work at your own pace. We appreciate your contribution and are happy to wait for updates. If you have questions, feel free to ask on the Airflow Slack.

---

Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you.

[wolvery]

Ok, fixes done!