Designing Airflow DAGs with Retries, SLAs, and Idempotency

A production Airflow DAG must handle ordinary failure without corrupting data or silently missing its delivery target. Three design concerns work together:

• Retries handle transient failures such as temporary database locks, API rate limits, or network interruptions.
• SLAs make late completion visible to operators.
• Idempotency ensures that rerunning or retrying a task produces the same correct result instead of duplicates.

This tutorial uses a daily revenue aggregation pipeline as an example. It assumes:

• Apache Airflow 2.x, where classic task-level SLA monitoring is available.
• PostgreSQL as the warehouse.
• A source table named orders and a reporting table named daily_order_revenue.
• A daily DAG that processes a closed data interval.

Airflow 3 introduces newer alerting concepts such as Deadline Alerts, so verify the behavior of SLA-related settings against the version you operate.

The architecture: separate orchestration from data correctness

Consider this pipeline:

PostgreSQL orders table | | Read one Airflow data interval v daily_revenue DAG task | | Delete and rebuild one report partition v PostgreSQL daily_order_revenue table | +--> Retry policy handles temporary failures +--> SLA callback alerts when completion is late 

The task processes one interval at a time:

[2025-03-10 00:00:00 UTC, 2025-03-11 00:00:00 UTC) 

That interval is more important than the wall-clock time at which the task happens to run. A manual rerun next week should still process the same logical interval, not “today.”

This leads to a useful division of responsibility:

Retries are not a substitute for idempotency. A retry means the task may execute its write logic multiple times. If each execution appends rows without a deterministic key or replacement strategy, the retry itself can create incorrect output.

Model retries around failure types

Not every error deserves a retry.

A retry is useful when the next attempt may reasonably succeed without changing the code or input:

• Temporary network failure to the database
• Database deadlock or lock timeout
• A short-lived warehouse outage
• Rate limiting from an external API
• Object storage request timeout

A retry is usually wasteful for deterministic failures:

• Missing required source column
• Invalid SQL syntax
• Bad credentials
• A malformed input file
• A violated business rule, such as negative revenue when negatives are forbidden

Airflow retries are configured per task. In this example:

retries=3 retry_delay=timedelta(minutes=5) retry_exponential_backoff=True max_retry_delay=timedelta(minutes=30) 

This means Airflow makes the initial attempt plus up to three retry attempts. Exponential backoff avoids having many failed tasks retry at the same moment after a shared outage.

Also use execution_timeout. A retry policy cannot help a task that hangs forever waiting on a database operation.

A good starting point is:

• Keep retry counts modest.
• Use a short initial delay for likely transient failures.
• Cap exponential backoff.
• Set an execution timeout based on the expected task duration.
• Make writes safe to repeat before enabling retries.

Build an idempotent daily aggregation task

The following DAG task rebuilds a single reporting partition. It uses Airflow’s data_interval_start and data_interval_end instead of datetime.now().

The destination table has a primary key on (report_date, order_id), but the task also deletes the entire daily partition before inserting. That deletion handles a subtle case: an order that was present in a prior attempt but is no longer eligible in a later rerun.

Example schema

CREATE TABLE orders ( order_id BIGINT PRIMARY KEY, paid_at TIMESTAMPTZ NOT NULL, amount_cents INTEGER NOT NULL, status TEXT NOT NULL ); CREATE TABLE daily_order_revenue ( report_date DATE NOT NULL, order_id BIGINT NOT NULL, amount_cents INTEGER NOT NULL, loaded_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), PRIMARY KEY (report_date, order_id) ); 

Airflow DAG

Save this as dags/daily_revenue.py. It requires a PostgreSQL Airflow connection named warehouse_postgres.

from datetime import datetime, timedelta, timezone import logging from airflow.decorators import dag, task from airflow.operators.python import get_current_context from airflow.providers.postgres.hooks.postgres import PostgresHook def report_sla_miss(dag, task_list, blocking_task_list, slas, blocking_tis): logging.error( "SLA missed for DAG=%s tasks=%s blocking_tasks=%s", dag.dag_id, task_list, blocking_task_list, ) @dag( dag_id="daily_revenue", start_date=datetime(2025, 3, 1, tzinfo=timezone.utc), schedule="0 0 * * *", catchup=False, max_active_runs=1, sla_miss_callback=report_sla_miss, tags=["reporting", "postgres"], ) def daily_revenue(): @task( retries=3, retry_delay=timedelta(minutes=5), retry_exponential_backoff=True, max_retry_delay=timedelta(minutes=30), execution_timeout=timedelta(minutes=20), # See the SLA section for why this is 26 hours.  sla=timedelta(hours=26), ) def rebuild_daily_revenue(): context = get_current_context() interval_start = context["data_interval_start"] interval_end = context["data_interval_end"] report_date = interval_start.date() sql = """ DELETE FROM daily_order_revenue WHERE report_date = %s; INSERT INTO daily_order_revenue ( report_date, order_id, amount_cents ) SELECT %s AS report_date, order_id, amount_cents FROM orders WHERE paid_at >= %s AND paid_at < %s AND status = 'paid'; """ hook = PostgresHook(postgres_conn_id="warehouse_postgres") # Use one database transaction: either replacement succeeds fully,  # or the previous partition remains unchanged after rollback.  with hook.get_conn() as conn: with conn.cursor() as cursor: cursor.execute( sql, ( report_date, report_date, interval_start, interval_end, ), ) conn.commit() rebuild_daily_revenue() daily_revenue() 

Suppose the DAG run processes 2025-03-10 and the source contains:

The expected destination rows are:

Order 103 is excluded because it was cancelled. Order 104 is excluded because the interval is half-open: it starts exactly at the next day’s boundary.

If the task fails after the DELETE but before the INSERT, the database transaction rolls back. If Airflow retries after a successful insert, the retry deletes and recreates the same partition. Both outcomes are safe.

Understand SLA timing before setting an SLA

An SLA is an operational expectation, not a data-quality guarantee. In classic Airflow SLA behavior, a missed SLA can trigger a callback or notification, but it does not automatically stop a task, repair data, or create a retry.

The confusing part is the reference time.

For a daily schedule at midnight, a run that executes shortly after:

2025-03-11 00:00 UTC 

typically processes this interval:

2025-03-10 00:00 UTC through 2025-03-11 00:00 UTC 

Its logical date is associated with the start of the interval: 2025-03-10 00:00 UTC.

If the business requirement is:

“Yesterday’s revenue must be ready by 02:00 UTC on the next calendar day.”

Then the deadline is 26 hours after the interval start, not 2 hours after it. That is why the example uses:

sla=timedelta(hours=26) 

An sla=timedelta(hours=2) setting would be interpreted relative to the logical date and may already be overdue when the scheduled run begins.

Classic SLA monitoring also depends on a healthy scheduler. Treat SLA alerts as a signal to investigate, not as the only mechanism for delivery monitoring.

For a broader learning path that connects Airflow concepts with warehouse modeling and orchestration fundamentals, this optional data engineering roadmap can be a useful next step.

Avoid the “idempotent SQL but changing source” trap

The partition replacement pattern is idempotent with respect to a fixed source state. But source state can change.

For example:

1. The first attempt reads an order with status = 'paid'.
2. The task fails after committing the destination write, or is manually rerun later.
3. Someone changes the source order to status = 'refunded'.
4. The rerun produces a different daily report.

That may be correct or incorrect depending on the business definition.

Decide explicitly whether your report is:

• As-of processing time: reruns should use the latest source state.
• As-of interval close: reruns should reproduce the original result.
• Eventually corrected: late arrivals and corrections are expected, and recent partitions are intentionally rebuilt.

For reproducible historical output, use immutable source files, source snapshots, change-data-capture offsets, or a versioned bronze layer. Airflow retries alone cannot freeze mutable source data.

Also avoid using NOW() inside transformation predicates. This query is unsafe for backfills:

WHERE paid_at >= NOW() - INTERVAL '1 day' 

It produces a different answer each time it runs. Use the Airflow interval boundaries instead.

Test retries, writes, and schedules separately

Test the transformation SQL independently from Airflow. Insert representative rows around boundaries:

• Exactly at data_interval_start
• One microsecond before data_interval_end
• Exactly at data_interval_end
• Cancelled or refunded orders
• Duplicate source events, if your source allows them
• Late-arriving events

Then verify these properties:

1. One interval produces only expected rows.
2. Running the same interval twice produces the same destination state.
3. A failure between delete and insert leaves no partial partition.
4. Adjacent intervals do not overlap or leave gaps.
5. A backfill uses historical interval boundaries, not the current date.

For the transactional behavior, deliberately raise an exception after the DELETE in a non-production test environment. Confirm that the transaction rolls back and the prior partition remains available.

For DAG-level tests, Airflow can parse and validate the DAG without executing it:

airflow dags list | grep daily_revenue airflow dags test daily_revenue 2025-03-11 

The dags test command is useful for validating task execution for a logical date, but it is not a replacement for an integration test database. Use an isolated PostgreSQL schema or disposable database for tests that mutate tables.

Finally, test SLA callbacks differently from task logic. SLA behavior is scheduler-driven and time-dependent, so unit test the callback function directly and validate actual alert routing in a staging Airflow environment.

Production pitfalls to plan for

A few configuration choices often cause surprising behavior:

• catchup=False does not make backfills impossible. Manual runs can still be created, so interval-based SQL remains necessary.
• max_active_runs=1 protects a mutable daily partition table. Without it, overlapping runs for the same business date can race unless the write strategy is designed for concurrency.
• Retries can overload a failing dependency. Use exponential backoff and consider Airflow pools for shared databases or APIs.
• Do not put secrets in DAG code. Use Airflow connections, a secrets backend, or workload identity.
• A task can succeed with bad data. Add validation tasks for row counts, null rates, reconciliation totals, or expected freshness.
• SLA misses do not imply data loss. A task may finish late but correctly; conversely, a task can finish on time with incorrect results.

Takeaway

Reliable Airflow DAGs treat retries, SLAs, and idempotency as one design problem:

• Retry only failures likely to be transient.
• Use data intervals, not current timestamps, to define work.
• Make destination writes safe to repeat with transactions and deterministic partition replacement or upserts.
• Set SLAs relative to Airflow’s logical date semantics.
• Test reruns, interval boundaries, rollback behavior, and mutable-source assumptions.

When a task can be retried or manually rerun without changing the correctness of its output, Airflow becomes much safer to operate under real-world failure conditions.