KPO deferable "random" false fail

[raphaelauv]

Apache Airflow version

2.6.1

What happened

With the KPO and only in deferrable I have "random" false fail

the dag

from pendulum import today
from airflow import DAG
from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator

dag = DAG(
    dag_id="kubernetes_dag",
    schedule_interval="0 0 * * *",
    start_date=today("UTC").add(days=-1)
)

with dag:
    cmd = "echo toto && sleep 22 && echo finish"

    KubernetesPodOperator.partial(
        task_id="task-one",
        namespace="default",
        kubernetes_conn_id="kubernetes_default",
        config_file="/opt/airflow/include/.kube/config",  # bug of deferrable corrected in 6.2.0
        name="airflow-test-pod",
        image="alpine:3.16.2",
        cmds=["sh", "-c", cmd],
        is_delete_operator_pod=True,
        deferrable=True,
        get_logs=True,
    ).expand(env_vars=[{"a": "a"} for _ in range(8)])

the log of the task in error :

dag_id=kubernetes_dag_run_id=scheduled__2023-05-16T00_00_00+00_00_task_id=task-one_map_index=2_attempt=1.log

What you think should happen instead

KPO should not fail ( in deferable ) if the container succesfully run in K8S

How to reproduce

If I remove the sleep 22 from the cmd then I do not see any more random task fails

Operating System

ubuntu 22.04

Versions of Apache Airflow Providers

apache-airflow-providers-cncf-kubernetes==6.1.0

Deployment

Docker-Compose

Deployment details

No response

Anything else

No response

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

[hussein-awala]

@raphaelauv, based on the log you provided in the issue description, it appears that the KuberenetesPodTrigger is triggering the success event too quickly, before the pod has terminated. As a result, when we call the cleanup method, it fails because it expects a terminated pod.

In #31348, I made a change to address this issue by modifying it to wait for the pod to reach a terminated state before triggering a success event. Could you please test it with your DAG to verify its effectiveness?

[raphaelauv]

unfortunately I'm not able to reproduce the case . I'm not aware of how I could tune K8S to reproduce it ( something like a terminationGracePeriodSeconds that would for sure reproduce the pattern )

[potiuk]

unfortunately I'm not able to reproduce the case . I'm not aware of how I could tune K8S to reproduce it ( something like a terminationGracePeriodSeconds that would for sure reproduce the pattern )

The fix look reasonable and hypothesis/fix looks plausible - so maybe we can merge it and close that issue then hoping it will solve similar issue. We can always reopen it if it occurs again (or create a new issue)