Was generative AI tooling used to co-author this PR?

• Yes Sonnet 4.5

Why?

KPO's existing reattach_on_restart relies on a label search on every retry, which can raise FoundMoreThanOnePodFailure when more than one matching pod exists, and gives no way to reconnect to a specific pod unambiguously. On Airflow 3.3+, persisting the running pod's identity to task state store lets a retry reconnect directly to that exact pod instead, removing the ambiguity window entirely. reattach_on_restart is deprecated in favor of a new durable flag that supersedes it; existing configurations map onto durable with zero behavior change, and pre-3.3 environments keep today's label-search behavior unchanged.

Summary

• Adds a durable parameter to KubernetesPodOperator (default True), superseding the
  deprecated reattach_on_restart parameter.
• On Airflow 3.3+, durable=True persists the running pod's identity (name, namespace, uid)
  to task state store before the operator waits on it. On retry, the operator reconnects
  directly to that pod via a direct read_namespaced_pod lookup instead of a label search --
  removing the FoundMoreThanOnePodFailure ambiguity window a label search can hit when more
  than one matching pod exists.
• The existing label-search mechanism (find_pod()) is retained as a one-time bootstrap
  fallback: a pod created by a pre-upgrade provider version (no persisted identity yet) is
  still found via labels on the first retry after upgrading, and that retry starts persisting
  state for every subsequent one. No in-flight task or existing pod breaks across an upgrade.
• reattach_on_restart still works, mapping its value onto durable with zero behavior
  change; passing it emits AirflowProviderDeprecationWarning on Airflow 3.3+ only (below
  3.3, durable/task state store don't functionally exist yet, so warning would be
  premature).
• durable=False opts out of task state store entirely, preserving today's reattach_on_restart=False
  behavior exactly, including its pre-existing incidental-cleanup quirk (an orphaned pod from
  a killed prior attempt can still get found and deleted via unconditional find_pod()
  refetch calls elsewhere in execute_sync/execute_async -- unrelated to and unchanged by
  this work).
• Docs: added a "Durable execution" section to KPO's operator docs, cross-linking task state
  store.

Testing

Using this dag:

from datetime import datetime, timedelta

from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator
from airflow.sdk import DAG

with DAG(
    dag_id="try_kpo",
    schedule=None,
    start_date=datetime(2024, 1, 1),
    catchup=False,
) as dag:
    run_pod = KubernetesPodOperator(
        task_id="run_pod",
        name="try-kpo",
        namespace="default",
        image="ubuntu:24.04",
        cmds=["bash", "-cx"],
        arguments=["sleep 60 && echo done"],
        in_cluster=False,
        do_xcom_push=False,
        retries=2,
        retry_delay=timedelta(seconds=5),
    )

Before my changes:

Worker killed:

But pod still exists:

Reconnected when worker came back up:

After my changes (changing to pass durable=True)

    run_pod = KubernetesPodOperator(
        task_id="run_pod",
        name="try-kpo",
        namespace="default",
        image="ubuntu:24.04",
        cmds=["bash", "-cx"],
        arguments=["sleep 60 && echo done"],
        in_cluster=False,
        do_xcom_push=False,
        retries=2,
        retry_delay=timedelta(seconds=5),
        durable=True,
    )

WOrker killed:

Task store persisted:

[Breeze:3.10.20] root@0014bf761b80:/opt/airflow$ cat /tmp/airflow_state/ti_run_pod/pod_identifier.json
{"name": "try-kpo-vlap9a9c", "namespace": "default", "uid": "878b321f-fb21-4779-94c4-1811f7ced720"}[Breeze:3.10.20]

Pod present and reconnected to via state store:

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.