BranchPythonOperator not running with past skipped state task

user_13514743 0
Viewed 301

This is how my airflow dag looks like

1:Airflow dag

There is a branch task which checks for a condition and then either :

Runs Task B directly, skipping task A or

Runs task A and then runs task B

When task A is skipped, in the next(future) run of the dag, branch task never runs(execution stops at main task) although default trigger rule is 'none_failed' and no task is failed in the dag only skipped.

default_args = {
    'owner': 'airflow',
    'depends_on_past': True,
    'wait_for_downstream': True, 
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=2),
    'trigger_rule': 'none_failed'
}

dag = DAG(
        dag_id='main_task',
        default_args=default_args,
        schedule_interval='0 2 * * *',
        start_date=datetime(2021,6,2),
        max_active_runs=8,
        )
def check_condition():
    if(conditionA == conditionB):
        return ['task_A','task_B']
    else :
        return 'task_B'

branch_task = BranchPythonOperator(
    task_id='branching',
    python_callable=check_condition,
    dag=dag,
    depends_on_past=False,
)

Using Airflow 1.10.12. Could someone guide me why branch task never runs after task A is skipped in the past run.

airflow python
1 Answers

The reason is happens isn't related to trigger rules. It happens because default_args in DAG constructor contains wait_for_downstream=True so when you do:

branch_task = BranchPythonOperator(
    task_id='branching',
    python_callable=check_condition,
    dag=dag,
    depends_on_past=False,
)

What actually happens is that depends_on_past is set to True by the constructor of BaseOpeartor. Since wait_for_downstream=True will cause a task instance to also wait for all task instances immediately downstream of the previous task instance to succeed this causes the BranchPythonOperator to not start running. This is a problem as branch operator usually have direct downstream tasks in Skip status.

You can fix it by:

branch_task = BranchPythonOperator(
    task_id='branching',
    python_callable=check_condition,
    dag=dag,
    depends_on_past=False,
    wait_for_downstream=False
)

I'd like to note that this is an issue only in Airlfow<2.0.0 because wait_for_downstream consider only Success status as accepted (1.10 operator description).

For Airflow > 2.0.0 this issue won't happen as the behavior was changed in PR by making wait_for_downstream consider both Successful and Skipped tasks as accepted statuses (2.0 operator description).

user_146244090
Related