Airflow Task States: Stuck in Queued? Here's the Fix

A transient DB blip failed one task at 3 AM. With retries=0 the whole pipeline died, and the page woke a human for an incident a retry would have healed. The fix was three lines of policy.

Task states tell you where every run stands, from queued to deferred. You'll learn which states signal trouble and which mean patience.

We'll map the state machine, honest retries, and concurrency knobs. Queued stops being a mystery.

The Full State Machine (Why It Matters)

Every task instance carries a state: scheduled, queued, running, success, failed, up_for_retry, skipped, up_for_reschedule, or deferred. Scheduled means the dependencies are met. Queued means it waits for a slot. Running means a worker owns it.

Terminal states end the story for that attempt. Success, failed, and skipped close the instance. Up_for_retry and deferred are waiting rooms with different landlords: the clock and the triggerer.

Learn to read the Grid view as sentences. Green success, red failed, grey queued-aging means capacity trouble, not code trouble.

Full 3.x list is 13 states: none, scheduled, queued, running, success, failed, skipped, upstream_failed, up_for_retry, up_for_reschedule, deferred, restarting, removed — plus awaiting_input for human-in-the-loop tasks. You'll clear failed instances (try_number bumps, max_tries resets) from Grid or airflow tasks clear, and history with per-try logs stays inspectable.

Retries: Honest Retry vs Hiding Bugs

Retries admit that networks blip. retries=3 gives a task three extra chances after the first failure. retry_delay spaces attempts so the dependency can recover. The defaults live in default_args and apply fleet-wide.

Honest retries pair with idempotent tasks. A safe rerun heals; an unsafe rerun duplicates. Set retries only on tasks you have proven re-runnable with a clear-and-rerun test.

Keep alerts on exhaustion, not on attempts. Paging on every first failure pages humans for blips. Paging when retries run out pages humans for real bugs.

Go finer with 3.x ExceptionRetryPolicy: map HTTPError to RETRY with a 5-minute delay, auth errors to FAIL fast, ConnectionError to 30-second RETRY. Rules run in the worker, first match wins, and retries still caps the total. Raise AirflowSkipException for no-data days and AirflowFailException to skip remaining retries on bad credentials.

import pendulum
from datetime import timedelta
from airflow.sdk import dag, task

@dag(
    schedule="@daily",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    default_args={
        "owner": "platform",
        "retries": 3,
        "retry_delay": timedelta(minutes=5),
        "max_retry_delay": timedelta(minutes=30),
    },
    tags=["resilience"],
)
def resilient_sync():
    @task
    def pull_warehouse() -> str:
        return "partition-2026-09-03"

    @task
    def publish(partition: str) -> None:
        print(f"publishing {partition}")

    publish(pull_warehouse())

resilient_sync()

Backoff and max_retry_delay

Backoff spaces retries exponentially: 5 minutes, then 10, then 20. Without it, every attempt hammers a dependency that is already down. With it, recovery gets breathing room.

max_retry_delay caps the growth so waits stay sane. A 6-hour cap on an hourly pipeline means attempts never sleep past usefulness. Tune caps to the schedule cadence.

Add jitter by staggering DAG start times across teams. Synchronized retries after a shared outage arrive as a stampede; spread starts turn the stampede into a queue.

Backoff knobs are retry_delay, retry_exponential_backoff, and max_retry_delay — keep caps under one schedule interval so hourly DAGs never sleep past usefulness. You'll also set execution_timeout (AirflowTaskTimeout on breach) per task and timeout on reschedule-mode sensors (AirflowSensorTimeout, no retry). Stagger starts to dodge herds.

What Queued Means and Why Tasks Get Stuck

Queued means waiting for a worker slot, a pool slot, or a concurrency limit. The task is ready but homeless. Short queues are normal; aging queues are capacity alarms.

Diagnose in order: pool occupancy, executor parallelism, per-DAG max_active_tasks, then greedy neighbors. One DAG with unbounded concurrency can starve ten polite ones.

Fix structurally. Raise pool slots deliberately, cap greedy DAGs, or move long waits to deferred operators. Restarting the scheduler treats the symptom for an hour.

Stuck running often means heartbeat timeout: OOMKill, liveness-probe restart, or node scale-down leaves a zombie the scheduler reaps and fails or retries. Tune task_instance_heartbeat_sec and watch for it after K8s churn. Check airflow tasks states-for-dag-run <dag> <date> to list states fast.

# find who sits queued and why
AIRFLOW_HOME=~/airflow airflow pools list
AIRFLOW_HOME=~/airflow airflow dags list-runs -d sales_daily --limit 10
AIRFLOW_HOME=~/airflow airflow tasks states-for-dag-run sales_daily 2026-09-03

Per-DAG vs Per-Task Concurrency Knobs

Three knobs bound concurrency at different scopes. parallelism caps the whole scheduler. max_active_tasks caps one DAG. Pools cap arbitrary task groups like warehouse writers.

Set them as a stack. parallelism guards the cluster, max_active_tasks guards the noisy DAG, pools guard the shared downstream. Each layer stops a different stampede.

Review quarterly as DAG counts grow. Defaults that fit 20 DAGs strangle 200. Capacity planning is scheduling work, not hardware shopping.

Trigger Rules: The Quiet Killer

Trigger rules decide which upstream states release a task. all_success needs every upstream green. none_failed tolerates skips. all_done runs no matter what. The default is all_success.

That default quietly kills joins after branches. One skipped path means not all succeeded, so the join fails on a healthy run. The DAG goes red over a path that was designed to be optional.

Set join rules deliberately and test skip scenarios with dags test. Branching gets its deep dive later; for now, remember the join is where defaults bite.

import pendulum
from airflow.sdk import dag, task
from airflow.sdk.bases.operator import chain

@dag(
    schedule="@daily",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    tags=["branching"],
)
def guarded_join():
    @task.branch
    def pick() -> str:
        return "heavy_path"

    @task
    def heavy_path() -> str:
        return "heavy done"

    @task
    def light_path() -> str:
        return "light done"

    @task(trigger_rule="none_failed")
    def join(results: str) -> None:
        print(f"joined: {results}")

    branch = pick()
    join(branch)

    heavy = heavy_path()
    light = light_path()
    chain(branch, heavy, light)

guarded_join()