Why did my join task get skipped?

The default trigger rule is all_success, and a skipped upstream does not count as success. Skips propagate down the graph, so a task after a branch is skipped unless you give it a rule such as none_failed_min_one_success.

What is the difference between none_failed and none_failed_min_one_success?

none_failed runs when no upstream failed, including when every upstream was skipped. none_failed_min_one_success additionally requires at least one upstream to have actually succeeded, which is almost always what you want after a branch.

Can a branch return more than one task id?

Yes. Return a list and every task in it runs, while all other direct downstream tasks are skipped. Returning an empty list skips all of them, which is a clean way to express do nothing today.

How do I run a cleanup task whether or not the pipeline succeeded?

Give it trigger_rule=all_done, which fires once every upstream has reached a terminal state regardless of what that state was. Pair it with a separate all_failed task if you want alerting only on failure.

Branching and Trigger Rules in Airflow

Real pipelines choose. A DAG is a static graph, so choosing means marking some branches skipped at runtime — and skips travel further down the graph than most people expect.

Branching

from datetime import datetime
from airflow import DAG
from airflow.operators.python import BranchPythonOperator
from airflow.operators.bash import BashOperator
from airflow.operators.empty import EmptyOperator


def choose_path(**context):
    """Must return the task_id (or list of ids) to run."""
    row_count = 1284                       # in reality: an XCom or a query
    if row_count == 0:
        return "no_data"
    if row_count > 1000:
        return "full_rebuild"
    return "incremental_load"


with DAG(
    dag_id="branching",
    start_date=datetime(2026, 2, 1),
    schedule=None,
    catchup=False,
) as dag:

    check = BranchPythonOperator(task_id="check_volume", python_callable=choose_path)

    full = BashOperator(task_id="full_rebuild",      bash_command="echo 'full rebuild'")
    incr = BashOperator(task_id="incremental_load",  bash_command="echo 'incremental'")
    none = EmptyOperator(task_id="no_data")

    publish = BashOperator(task_id="publish", bash_command="echo 'publishing'")

    check >> [full, incr, none] >> publish

airflow dags trigger branching && sleep 20
airflow tasks states-for-dag-run branching manual__2026-02-14T14:02:00+00:00

dag_id    | task_id          | state
==========+==================+=========
branching | check_volume     | success
branching | full_rebuild     | success
branching | incremental_load | skipped
branching | no_data          | skipped
branching | publish          | skipped

The branch worked — full_rebuild ran, the other two were skipped. But publish was skipped too, and that is not what anyone intends.

Why the skip propagated

publish uses the default trigger rule all_success, which requires every direct upstream to have succeeded. Two of its three upstreams were skipped, so the rule was not satisfied, so publish was skipped as well — and anything after it would be too.

Fix it with a rule that tolerates skips:

    publish = BashOperator(
        task_id="publish",
        bash_command="echo 'publishing'",
        trigger_rule="none_failed_min_one_success",
    )

branching | check_volume     | success
branching | full_rebuild     | success
branching | incremental_load | skipped
branching | no_data          | skipped
branching | publish          | success

none_failed_min_one_success means: nothing upstream failed, and at least one upstream succeeded. That is the correct rule for a join after a branch, and it is the single most common Airflow fix there is.

The trigger rules

trigger_rule="all_success"                  # default — every upstream succeeded
trigger_rule="all_failed"                   # every upstream failed
trigger_rule="all_done"                     # every upstream finished, any state
trigger_rule="one_success"                  # at least one succeeded (runs early)
trigger_rule="one_failed"                   # at least one failed (runs early)
trigger_rule="none_failed"                  # none failed; all-skipped still runs
trigger_rule="none_failed_min_one_success"  # none failed and one succeeded
trigger_rule="none_skipped"                 # no upstream was skipped
trigger_rule="always"                       # run regardless, even with no upstream

The two that matter most in practice are none_failed_min_one_success for joins after branches, and all_done for cleanup.

Cleanup that always runs

    cleanup = BashOperator(
        task_id="cleanup",
        bash_command="echo 'removing temp files'; rm -rf /tmp/staging/*",
        trigger_rule="all_done",
    )

    alert = BashOperator(
        task_id="alert",
        bash_command="echo 'PIPELINE FAILED' >&2",
        trigger_rule="one_failed",
    )

    [full, incr, none] >> publish >> [cleanup, alert]

With everything succeeding:

branching | publish | success
branching | cleanup | success
branching | alert   | skipped

With a failure injected upstream:

branching | publish | upstream_failed
branching | cleanup | success
branching | alert   | success

cleanup ran both times; alert only fired on the failure. Note one_failed triggers as soon as any upstream fails, without waiting for the others — useful for alerting quickly, and worth knowing if your alert task assumes the run is over.

Branching with TaskFlow

from airflow.decorators import dag, task


@dag(dag_id="branch_taskflow", start_date=datetime(2026, 2, 1),
     schedule=None, catchup=False)
def pipeline():

    @task
    def count_rows() -> int:
        return 1284

    @task.branch
    def route(rows: int) -> str:
        print(f"routing on {rows} rows")
        return "full_rebuild" if rows > 1000 else "incremental_load"

    @task
    def full_rebuild() -> str:
        return "rebuilt"

    @task
    def incremental_load() -> str:
        return "incremented"

    @task(trigger_rule="none_failed_min_one_success")
    def publish() -> None:
        print("publishing")

    rows = count_rows()
    branch = route(rows)
    branch >> [full_rebuild(), incremental_load()] >> publish()


pipeline()

INFO - routing on 1284 rows
INFO - Following branch full_rebuild
INFO - Marking task incremental_load as SKIPPED
INFO - publishing

@task.branch returns the task id to follow. The log line Following branch full_rebuild is the one to check when a branch does not do what you expect — it names exactly what the callable returned.

Returning an invalid task id

    @task.branch
    def bad_route() -> str:
        return "does_not_exist"

airflow.exceptions.AirflowException:
Branch callable must return valid task_ids. Invalid tasks found: {'does_not_exist'}

A clear failure rather than a silent skip-everything, which is the right behaviour. Note it must be a direct downstream task — returning the id of a task two hops away raises the same error.

Branching to nothing

    @task.branch
    def maybe_run(rows: int) -> list[str]:
        return [] if rows == 0 else ["process"]

INFO - Following branch []
INFO - Marking task process as SKIPPED
publish | success

An empty list skips every branch cleanly, and publish still runs under none_failed_min_one_success… except it does not, because nothing succeeded upstream. For a DAG that may legitimately do nothing, use none_failed instead, which permits the all-skipped case.

Short-circuiting

When the choice is “continue or stop” rather than “which path”, a branch is overkill:

from airflow.operators.python import ShortCircuitOperator

    gate = ShortCircuitOperator(
        task_id="only_on_weekdays",
        python_callable=lambda **c: c["logical_date"].weekday() < 5,
    )

    gate >> transform >> publish

[2026-02-14 14:30:02] {python.py:240} INFO - Condition result is False
[2026-02-14 14:30:02] {python.py:245} INFO - Skipping downstream tasks...

only_on_weekdays | success
transform        | skipped
publish          | skipped

The gate itself succeeds and everything after it skips. This reads better than a branch with a dummy no-op path.

Trigger rules and retries

A subtlety worth knowing:

    flaky = BashOperator(task_id="flaky", bash_command="exit 1", retries=2)
    after  = BashOperator(task_id="after", bash_command="echo hi", trigger_rule="all_done")
    flaky >> after

flaky | up_for_retry   (attempt 1)
flaky | up_for_retry   (attempt 2)
flaky | failed         (attempt 3)
after | success

all_done waits for a terminal state, so it does not fire on up_for_retry. Retries are exhausted first. That is what you want — a cleanup task should not run between attempts.

Practice

1. Build a branch with a join using the default trigger rule.

chosen_path | success
other_path  | skipped
join        | skipped

The join skips because all_success counts a skip as not-success. Change it to none_failed_min_one_success and it runs. If you only remember one thing about trigger rules, make it this.

2. Add an all_done cleanup and fail an upstream task.

extract | failed
load    | upstream_failed
cleanup | success

cleanup ran despite two upstream problems. Note it fires on upstream_failed as well as failed — both are terminal states.

3. Return a list of two task ids from a branch.

INFO - Following branch ['load_a', 'load_b']
load_a | success
load_b | success
load_c | skipped

Both run in parallel. This is the clean way to express “process these regions today” without building the DAG dynamically.

4. Use none_failed where every branch was skipped.

none_failed:                  join | success
none_failed_min_one_success:  join | skipped

The difference only shows in the all-skipped case. Pick none_failed when a no-op run is valid, and none_failed_min_one_success when the join genuinely needs data from somewhere.

Next: connecting to the systems your tasks actually talk to.