Airflow TriggerDagRunOperator的allowed_states参数失效问题求助

你的问题核心是对TriggerDagRunOperator的参数逻辑理解有偏差：allowed_states只是定义子DAG的合法完成状态范围，但默认情况下，只要子DAG最终状态不是SUCCESS，算子仍会判定自身失败——除非你同时配置failed_states参数，明确告诉算子哪些失败状态是“预期内的”，不会触发自身报错。

修正后的代码配置 #

直接在原有基础上添加failed_states参数，将FAILED加入其中：

from airflow.utils.state import State
from airflow.operators.trigger_dagrun import TriggerDagRunOperator

dag3_trigger = TriggerDagRunOperator(
    task_id="demo_dag_3_trigger",
    trigger_dag_id="demo_dag_3",
    wait_for_completion=True,
    allowed_states=[State.SUCCESS, State.FAILED],
    failed_states=[State.FAILED],  # 关键配置：标记该失败状态为预期内
    poke_interval=5,
)

参数逻辑说明 #

当wait_for_completion=True时，算子的判定逻辑是：

• 子DAG进入allowed_states中的SUCCESS：算子直接标记为成功
• 子DAG进入failed_states中的状态：算子认为这是预期内的结果，不会抛出异常，自身标记为成功
• 子DAG进入既不在allowed_states也不在failed_states的状态（比如UP_FOR_RETRY）：才会抛出AirflowException并标记自身失败

之前你只设置了allowed_states，但未配置failed_states，所以算子看到子DAG是FAILED时，仍会将其视为“非预期失败”，进而触发报错。

额外验证点 #

如果用字符串格式的状态值，同样有效（注意Airflow内部状态为大写）：

allowed_states=['SUCCESS', 'FAILED'],
failed_states=['FAILED'],

推荐使用State类的常量，避免手动拼写错误。

内容的提问来源于stack exchange，提问作者Mario Berg