[Python] How to find failed tasks in DAG level failure callback with KubernetesExecutor

Is it possible to find failed tasks from a DAG level failure callback when using the Kubernetes executor? All my attempts have failed, because failed tasks still show as 'running' when the failure callback is called.

For example:

import pendulum

from airflow import DAG
from airflow.operators.python import PythonOperator


def succeed(*args, **kwargs):
return 1


def fail(*args, **kwargs):
1 / 0


def failure_callback(context):
dag_run = context["dag_run"]
tasks = dag_run.get_task_instances()

for ti in tasks:
print(f"{ti.task_id}: {ti.state}")


with DAG(
"test_email",
start_date=pendulum.datetime(2023, 1, 1, tz="UTC"),
schedule="@daily",
catchup=False,
on_failure_callback=failure_callback,
) as dag:
f1 = PythonOperator(task_id="failing-task", python_callable=fail)
f2 = PythonOperator(task_id="failing-task2", python_callable=fail)
s2 = PythonOperator(task_id="succeeding-task", python_callable=succeed)
skipped = PythonOperator(task_id="skipped-task", python_callable=fail)

s2 >> [f1, f2] >> skipped


If the above code runs with dag.test(), it prints:

failing-task: failed
failing-task2: failed
succeeding-task: success
skipped-task: upstream_failed


However, when run with the Kubernetes executor, it prints something like:

failing-task: running
failing-task2: running
succeeding-task: success
skipped-task: None


I thought this might be a timing issue, but even if I sleep for 2 minutes in my failure callback the task instances still show as running. Is there any way to find failed tasks from a dag level failure callback?

Continue reading...