Using Airflow to Orchestrate GCP Pipeline Failover Tests

Using Airflow to Orchestrate Daily GCP Pipeline Failover Tests

A pipeline that's never had its failure path tested is a pipeline you don't actually understand yet. Apache Airflow is the standard tool for orchestrating multi-step data pipelines — and one of its most underused features is scheduling regular failover tests, not just the happy-path job.

📘 Why Test Failover on a Schedule

Most teams only discover their retry logic, alerting, and fallback paths don't actually work the moment a real production failure happens — which is the worst possible time to find out. Running a scheduled, low-stakes failover drill (e.g., forcing one task to fail and confirming recovery) catches gaps before they matter.

🧪 A Simple Airflow DAG Structure

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

default_args = {
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
}

with DAG(
    "gcp_pipeline_failover_test",
    schedule_interval="@daily",
    start_date=datetime(2026, 1, 1),
    default_args=default_args,
    catchup=False,
) as dag:

    def extract():
        print("Extracting from source...")

    def simulate_failure():
        import random
        if random.random() < 0.3:  # simulate ~30% failure rate in test env
            raise Exception("Simulated transient failure")

    def load_to_bigquery():
        print("Loading to BigQuery...")

    extract_task = PythonOperator(task_id="extract", python_callable=extract)
    failover_task = PythonOperator(task_id="simulate_failure", python_callable=simulate_failure)
    load_task = PythonOperator(task_id="load_to_bigquery", python_callable=load_to_bigquery)

    extract_task >> failover_task >> load_task

The retries and retry_delay in default_args are doing the real work here — this is what you're actually testing: does the DAG recover automatically within an acceptable time window, or does it silently stall?

🎯 What "Good" Failover Looks Like

• The DAG retries the failed task automatically, without manual intervention
• An alert fires (Slack, email, PagerDuty) if retries exhaust — silence on failure is the real danger, not the failure itself
• Downstream tasks don't run with partial/incomplete upstream data — Airflow's dependency graph should block them until the upstream task truly succeeds
• Recovery time is measured and tracked over time, not just checked once

🛠️ Alerting on Failure

def alert_on_failure(context):
    task_id = context["task_instance"].task_id
    print(f"ALERT: {task_id} failed after all retries exhausted")
    # send_to_slack(...) in a real pipeline

failover_task = PythonOperator(
    task_id="simulate_failure",
    python_callable=simulate_failure,
    on_failure_callback=alert_on_failure,
)

⚠️ Common Mistakes

• Setting retries but never testing whether the retry actually resolves the underlying failure condition
• No alerting configured, so a failed DAG just sits there until someone happens to check the Airflow UI
• Testing failover only in a way that's too "soft" to be realistic — e.g., never actually killing a task mid-execution
• Running failover tests directly against production data instead of an isolated test dataset

🔗 Related Posts

• How I Replaced Legacy SSIS with Real-Time GCP Pipelines
• Change Data Capture (CDC): Real-Life Use Cases and Pitfalls