Prevent KubernetesExecutor from launching stale workloads

[wolvery]

Fixes #69760

This adds a final DB preflight before KubernetesExecutor creates a worker pod for an Airflow 3 ExecuteTask workload. A workload can sit in the executor queue while Kubernetes pod creation is delayed, and in HA scheduler deployments the task instance can be retried, cleared, or otherwise moved on before the pod is eventually created.

The executor now verifies that the workload still matches the current queued task instance before calling run_next:

• the immutable TaskInstance id still exists
• the TaskInstance is still QUEUED
• try_number still matches the workload
• queued_by_job_id still belongs to this scheduler job

If the workload is stale, it is dropped before pod creation and executor bookkeeping for that launch is cleared.

Performance trade-off: this adds one indexed metadata DB query and one database round-trip per pod launch. The query uses the TaskInstance primary key, and the additional round-trip is an intentional correctness cost to prevent stale workloads from creating duplicate or orphaned pods.

Tests added:

• KubernetesExecutor drops a stale ExecuteTask workload before pod creation when the DB task state has moved on

Local verification:

• python3 -m compileall providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py
• AIRFLOW_HOME=/var/folders/_l/8htlzsws7gjg8_p6ht_hl0tc0000gp/T/opencode/airflow-test-home uv run pytest providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py::TestKubernetesExecutor::test_sync_drops_stale_execute_task_workload_before_pod_creation -q --with-db-init
• AIRFLOW_HOME=/var/folders/_l/8htlzsws7gjg8_p6ht_hl0tc0000gp/T/opencode/airflow-test-home uv run pytest providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py::TestKubernetesExecutor::test_run_next_exception_requeue providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py::TestKubernetesExecutor::test_skip_pod_creation_on_create_pods_after -q
• uv run ruff check providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py
• uv run ruff format --check providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py

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

[potiuk]

Thanks — the bug is real and well-described. A workload can sit in the executor queue while pod creation is delayed, and in an HA scheduler deployment the task instance can be retried, cleared or re-queued in the meantime, so launching the pod anyway produces a duplicate or orphaned run. Revalidating against the immutable TI id before creating the pod is the right shape, and checking all four of (exists, still QUEUED, try_number, queued_by_job_id) is the correct set — queued_by_job_id in particular is what catches the HA case.

I also checked whether this crosses an architecture boundary and it doesn't: this executor already queries the metadata DB (@provide_session at line 547, select(TaskInstance.state) at 699), so it's consistent with the existing design rather than new access.

Failing open on ImportError and on a non-numeric scheduler_job_id is the right call too — a preflight check should never be the thing that stops tasks running.

One blocking issue with the query itself, inline.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting