Airflow Error Handling and Retries

Errors happen in every real pipeline — APIs go down, databases time out, files arrive late. Airflow gives you multiple tools to handle failures automatically so your pipelines recover without manual intervention.

What Happens When a Task Fails?

Normal flow:
[task_a: SUCCESS] → [task_b: SUCCESS] → [task_c: SUCCESS]

When task_b fails:
[task_a: SUCCESS] → [task_b: FAILED] → [task_c: NOT RUN]

Airflow marks task_b red in the UI.
task_c does not run (default behavior).
An alert email is sent (if configured).

Retries: Automatic Failure Recovery

Set retries to tell Airflow how many times to retry a failed task before giving up. Set retry_delay to add a wait between each attempt.

from datetime import timedelta
from airflow.operators.python import PythonOperator

def call_external_api():
    import requests
    response = requests.get("https://api.example.com/data", timeout=10)
    response.raise_for_status()
    return response.json()

task = PythonOperator(
    task_id="fetch_api_data",
    python_callable=call_external_api,
    retries=3,                              # retry up to 3 times
    retry_delay=timedelta(minutes=5),       # wait 5 minutes between retries
    retry_exponential_backoff=True,         # double the wait each time
    max_retry_delay=timedelta(minutes=30),  # never wait more than 30 minutes
)

Retry Timeline with exponential backoff:
Attempt 1: FAILS at 10:00
  Wait: 5 minutes
Attempt 2: FAILS at 10:05
  Wait: 10 minutes (doubled)
Attempt 3: FAILS at 10:15
  Wait: 20 minutes (doubled)
Attempt 4: FAILS at 10:35
  → Task marked as FAILED permanently

Setting Retries in default_args (DAG-Wide)

default_args = {
    "retries": 2,
    "retry_delay": timedelta(minutes=3),
    "email_on_failure": True,
    "email_on_retry": False,
    "email": ["data-team@company.com"],
}

with DAG("sales_pipeline", default_args=default_args, ...) as dag:
    # Every task inherits retries=2 and retry_delay=3min
    task1 = PythonOperator(...)
    task2 = PythonOperator(...)

Individual task settings override default_args. A task with retries=5 ignores the DAG-level retries=2.

Execution Timeout

Set a maximum run time for a task. If the task runs longer than the timeout, Airflow kills it and marks it failed:

slow_task = PythonOperator(
    task_id="run_heavy_query",
    python_callable=run_query,
    execution_timeout=timedelta(minutes=30),  # kill if it runs > 30 minutes
)

Use Case: A database query normally takes 5 minutes.
If it suddenly takes 45 minutes, something is wrong.
The timeout catches it and triggers a retry or alert.

Handling Errors Inside Task Code

Catch expected errors inside your function and decide what to do:

import requests
from airflow.exceptions import AirflowSkipException, AirflowFailException

def fetch_daily_report(date_str):
    try:
        response = requests.get(f"https://api.example.com/report/{date_str}")

        if response.status_code == 404:
            # Report not available yet — skip this task gracefully
            raise AirflowSkipException(f"No report for {date_str} yet")

        response.raise_for_status()
        return response.json()

    except requests.exceptions.ConnectionError:
        # Network is down — let Airflow retry
        raise  # re-raise so Airflow sees the failure

    except ValueError as e:
        # Bad data format — this is a real failure, don't retry
        raise AirflowFailException(f"Bad data format: {e}")

On-Failure Callbacks

Run custom code automatically when a task fails — post to Slack, open a ticket, or log to a monitoring system:

def notify_slack_on_failure(context):
    dag_id  = context["dag"].dag_id
    task_id = context["task_instance"].task_id
    run_id  = context["run_id"]
    print(f"ALERT: {dag_id} / {task_id} failed in run {run_id}")
    # In production: call Slack API or PagerDuty here

def process_data():
    raise ValueError("Something went wrong!")

task = PythonOperator(
    task_id="process_data",
    python_callable=process_data,
    on_failure_callback=notify_slack_on_failure,
)

Available Callback Hooks

SLA (Service Level Agreement) Deadlines

An SLA sets the maximum time a task is allowed to take from the start of the DAG run. Breaching the SLA triggers a callback and logs a warning:

from datetime import timedelta

task = PythonOperator(
    task_id="morning_report",
    python_callable=generate_report,
    sla=timedelta(hours=2),  # must complete within 2 hours of the DAG run starting
)

Manually Marking Tasks in the UI

When you fix a bug and want to re-run or skip a task without triggering the whole DAG:

• Clear: Resets the task to "no status" so the Scheduler re-runs it
• Mark Success: Marks a failed task as succeeded so downstream tasks unblock
• Mark Failed: Forces a running or queued task to fail immediately

Access these options by clicking any task box in Grid View or Graph View, then choosing the action from the popup panel.

Error Handling Strategy: A Decision Diagram

Task Fails
    │
    ├─ Is the error temporary? (network timeout, API rate limit)
    │        └── YES → Set retries + retry_delay
    │
    ├─ Is the error permanent? (bad data format, missing config)
    │        └── YES → Raise AirflowFailException (no retries)
    │
    ├─ Is it OK to skip and continue?
    │        └── YES → Raise AirflowSkipException
    │
    └─ Does the team need to know immediately?
              └── YES → Add on_failure_callback to notify Slack / PagerDuty