Airflow KubernetesPodOperator任务失败后仍显示运行状态求助

在Airflow项目中使用KubernetesPodOperator时，部分任务在失败且耗尽设定的重试次数后，仍处于Running状态——UI显示为Running，但实际无对应的Worker Pod运行，该状态可能持续长达一天。

以下是简化后的DAG代码：

from datetime import datetime, timedelta
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator

# 假设DBT_POD_TEMPLATE_PATH已提前定义
DBT_POD_TEMPLATE_PATH = "/path/to/your/pod/template.yaml"

with DAG(
        dag_id='dbt',
        dagrun_timeout=timedelta(hours=4),
        start_date=datetime(2023, 1, 1),
        schedule_interval="25 * * * *",
        catchup=True,
        max_active_runs=1,
) as dag:

    dbt_test_model = KubernetesPodOperator(
        task_id="dbt_test",
        name="dbt-run-test",
        cmds=["sh", "-c"],
        arguments=["dbt source model_test"],
        get_logs=True,
        retries=3,
        in_cluster=True,
        is_delete_operator_pod=False,
        pod_template_file=DBT_POD_TEMPLATE_PATH,
        dag=dag
    )

    dbt_run_model = KubernetesPodOperator(
        task_id="dbt_run",
        name="dbt-run",
        cmds=["sh", "-c"],
        arguments=["dbt run"],
        get_logs=True,
        retries=3,
        in_cluster=True,
        is_delete_operator_pod=False,
        pod_template_file=DBT_POD_TEMPLATE_PATH,
        dag=dag
    )

已尝试调试但未找到解决方案，请问有谁遇到过类似问题或有处理建议吗？

内容的提问来源于stack exchange，提问作者David Shpilfoygel