Airflow Deferrable Operators & Triggerer: Cutting Idle Worker Costs by 90%

airflow deferrable operator is the single Airflow feature that turns an idle-sensor line item on a monthly cloud bill from a five-digit number into rounding-error noise — and the piece of the 2.2+ architecture that most senior data engineers still explain wrong in interviews. A classic S3KeySensor that pokes every five minutes for the eight hours between a scheduled DAG start and the file's actual landing time holds a full worker slot for those eight hours, doing effectively nothing 99% of the time. Multiply that across a hundred sensor tasks per day, ten worker replicas, and a KubernetesExecutor that bills a pod-hour for every held slot, and the idle-worker cost line becomes the loudest number in the FinOps review. The airflow triggerer service, introduced in 2.2 and hardened through 2.10, exists specifically to end that waste — the sensor "defers" itself to a shared async event loop, the worker slot is released back to the pool the moment the wait begins, and the task only re-hydrates a worker once the event actually fires.

This guide is the senior-DE walkthrough you wished existed the first time an interviewer asked "walk me through the deferrable operator lifecycle from execute() through TriggerEvent back to method_name" or "how do you run the airflow triggerer for HA, and what happens when one triggerer instance crashes mid-wait?" or "write me a custom Trigger class that polls an internal REST API every 30 seconds with an exponential backoff." It walks through why the classic poke-based sensor is the wrong architecture for any wait longer than 60 seconds, the defer method contract that pairs BaseTrigger subclasses with asyncio coroutines, the Triggerer service's role alongside scheduler, webserver, and workers, the HA topology that lets two triggerer pods share a claim table, the deferrable=True migration path across the modern Provider catalogue (S3, GCS, HTTP, Databricks, Snowflake, BigQuery), and the production cost-audit query that quantifies the idle-slot-hours you actually save. Each section pairs a teaching block with a Solution-Tail interview answer — code, a step-by-step trace, an output table, then a concept-by-concept breakdown of why it works.

When you want hands-on reps immediately after reading, drill the ETL practice library → for the pipeline-orchestration problems, rehearse on the SQL practice library → for the sensor-driven data-wait patterns, and sharpen the tuning axis with the optimization practice library → for the cost-audit and slot-utilisation problems.

---

On this page

• Why deferrable operators cut idle worker costs by 90%
• Deferrable operator lifecycle
• Writing a custom Trigger class
• Triggerer service — deployment + HA
• Migration + operator inventory
• Cheat sheet — deferrable recipes
• Frequently asked questions
• Practice on PipeCode

---

1. Why deferrable operators cut idle worker costs by 90%

Every sensor and every long-running task holds a worker slot idle — the deferrable model releases the slot and hands the wait to a shared async loop

The one-sentence invariant: the classic Airflow sensor holds a full worker slot for the entire wait, but a deferrable operator releases the slot on .defer(), hands the wait to a shared async loop in the airflow triggerer service, and only re-hydrates a worker when the awaited event fires. Every question about airflow deferrable operator, the defer method, airflow async internals, airflow sensor cost, deferrable sensor migration, airflow trigger class design, asyncio airflow semantics, airflow poke interval tuning, and long-running task airflow architecture is a downstream consequence of that one architectural swap. Understand the swap and every subsequent design question answers itself.

The four "must-answer" axes interviewers actually probe.

• Trigger class. The subclass of BaseTrigger that hosts the async wait. serialize() returns the class path + kwargs needed to rebuild the Trigger on the Triggerer; run() is the async coroutine that polls, waits, and finally yields a TriggerEvent. The senior signal is being able to write one from memory in ten lines.
• .defer(). The operator-side call that hands control from the worker to the Triggerer. self.defer(trigger=..., method_name="execute_complete") records the Trigger in the metadata DB, marks the task as deferred, and returns the worker slot immediately. The task's next hop is not execute() finishing — it's execute_complete(context, event) running on a worker after the TriggerEvent fires.
• Triggerer service. The separate long-running Airflow process that hosts an asyncio event loop and runs every registered Trigger's run() concurrently. Alongside scheduler / webserver / worker, the Triggerer is its own Airflow component in 2.2+. airflow triggerer is the command; HA is achieved by running multiple triggerer instances that share the trigger table.
• Backoff. The polling cadence inside the Trigger's async run(). A well-written Trigger polls at 30–60 seconds with jitter and exponential backoff on transient errors; a poorly-written one hot-loops with while True: pass and negates the entire cost benefit. Backoff is where senior engineers separate the async coroutine from the poke-based sensor of the past.

Why the classic sensor is the wrong architecture for any wait longer than 60 seconds.

• Slot cost. A PokeSensor mode poke holds the worker slot for the entire wait — the slot is running, the pod is scheduled, the memory is reserved. mode = reschedule gives some relief by exiting the task after each poke, but each retry still restarts the pod, re-hydrates the DAG, re-parses the task, and pays a 3–10-second warmup on every cycle.
• Concurrency ceiling. A worker with 8 slots that runs 8 sensor tasks in poke mode does zero real work. The rest of the DAG queues behind an artificial cap. In KubernetesExecutor terms, you scale pods to serve waits — the definition of waste.
• Cloud billing model. Managed Airflow platforms (MWAA, Composer, Astronomer) bill by worker-node-hour. Idle sensor slots translate directly into billed compute. A 200-task-per-day sensor workload where 80% of tasks wait more than an hour each becomes the single largest line item on the bill.
• Retry churn. Sensor timeouts that fail and retry re-hydrate the worker, re-download the DAG, re-parse the task, re-authenticate against the cloud provider — for something that ultimately just polls an S3 bucket. The waste compounds.
• Deferrable fix. .defer() releases the slot the moment the wait begins. A single Triggerer with 1 CPU hosts hundreds to thousands of concurrent Trigger run() coroutines. Idle-worker cost drops from slot_hours = wait_hours × concurrency to essentially zero.

The 2026 reality — deferrable is the default, not the exception.

• First-party Providers ship deferrable versions. The Amazon Provider ships S3KeySensorAsync, the Google Provider ships GCSObjectExistenceSensorAsync, the HTTP Provider ships HttpSensorAsync, and modern Snowflake / Databricks / BigQuery operators accept a deferrable=True kwarg that flips the classic operator into deferrable mode without a code rename.
• deferrable=True toggle is the migration path. New Airflow operators expose a deferrable: bool = False config knob (or a global [operators] default_deferrable = True setting in airflow.cfg). Flipping the flag re-routes the operator through the Trigger path without further code changes.
• Triggerer is not optional infrastructure. Any DAG that uses a deferrable operator requires at least one airflow triggerer process running. On managed platforms the triggerer is provisioned as a first-class component; on self-managed deployments the Helm chart, docker-compose, or systemd unit must include it.
• Interview signal. Senior candidates in 2026 who don't reach for deferrable=True when asked about a long-running sensor are giving a 2020 answer to a 2026 question. The default answer is deferrable; the burden of proof is on not using it.

What interviewers listen for.

• Do you say "the sensor holds a full worker slot idle" in the first sentence when asked about airflow sensor cost? — senior signal.
• Do you name BaseTrigger.serialize + async run + .defer + Triggerer as the four moving parts of a deferrable operator? — required answer.
• Do you push back on "just increase parallelism" with the "you're just adding more idle slots" counter? — required answer.
• Do you describe the Triggerer as "a separate service that hosts an asyncio event loop over hundreds of concurrent Trigger coroutines" rather than as vague "async magic"? — required answer.

Worked example — the idle-sensor cost problem

Detailed explanation. The textbook cost story: a data platform runs 100 daily DAGs, each with one file-arrival sensor that waits an average of 6 hours for an upstream export. The sensors are configured in poke mode with poke_interval=300. The worker fleet is sized to accommodate the peak concurrent sensor count (100), so at any moment there are 100 workers holding 100 slots idle. Walk the interviewer through the cost model.

• The workload. 100 sensors × 6-hour average wait = 600 sensor-slot-hours per day.
• The slot cost. At $0.10/pod-hour on Kubernetes (a mid-range managed price), 600 sensor-slot-hours × $0.10 = $60/day = $1800/month.
• The waste multiplier. The sensors do zero work during those 6 hours. Every dollar is pure idle overhead.
• The deferrable fix. After migration to S3KeySensor(deferrable=True), the worker slots release on .defer(). A single Triggerer with 1 CPU hosts all 100 concurrent Trigger coroutines. Idle-slot-hours per day drops from 600 to ~2.

Question. A data platform runs 200 daily DAGs, each with 2 sensor tasks (400 total sensors per day). The average wait per sensor is 4 hours, poke_interval is 300 seconds, and each worker slot is billed at $0.08/hour on Kubernetes. Quantify the current idle-slot cost, the after-migration cost, and the payback on migration engineering.

Input.

Parameter	Current (poke mode)	After migration (deferrable)
Daily sensor tasks	400	400
Average wait per sensor	4 hours	4 hours
Worker slot cost	$0.08/hour	$0.08/hour
Slot-hours per sensor	4 (held idle)	~0.02 (only execute + resume)
Triggerer cost	0	1 CPU × 24 h × $0.05 = $1.20/day

Code.

# The problem — a classic poke-mode sensor holding a slot idle
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from airflow.decorators import dag
from datetime import datetime, timedelta

@dag(
    dag_id="daily_export_wait_v1",
    start_date=datetime(2026, 1, 1),
    schedule="0 6 * * *",
    catchup=False,
)
def pipeline():
    wait_for_export = S3KeySensor(
        task_id="wait_for_export",
        bucket_key="s3://exports/dt={{ ds }}/data.parquet",
        aws_conn_id="aws_default",
        # Classic poke — worker slot held for the full wait
        mode="poke",
        poke_interval=300,      # 5 minutes
        timeout=6 * 3600,       # 6 hour cap
    )
    # ...downstream tasks...

pipeline()

# The fix — one keyword argument
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from airflow.decorators import dag
from datetime import datetime

@dag(
    dag_id="daily_export_wait_v2",
    start_date=datetime(2026, 1, 1),
    schedule="0 6 * * *",
    catchup=False,
)
def pipeline():
    wait_for_export = S3KeySensor(
        task_id="wait_for_export",
        bucket_key="s3://exports/dt={{ ds }}/data.parquet",
        aws_conn_id="aws_default",
        # One flag — task now uses the deferrable Trigger path
        deferrable=True,
        poke_interval=60,       # inner polling cadence inside the Trigger
        timeout=6 * 3600,
    )
    # ...downstream tasks...

pipeline()

Step-by-step explanation.

1. The current v1 DAG uses the classic S3KeySensor with mode="poke". When the DAG runs at 06:00, the worker slot is acquired at 06:00 and held until the file lands at (say) 10:00. Four hours of billed worker time — pure idle.
2. The cost math: 400 sensors/day × 4 hours idle × $0.08/slot-hour = $128/day = ~$3840/month. On a 200-DAG platform this is one of the top three cost lines in the FinOps review.
3. The v2 DAG flips one flag: deferrable=True. On execute, the operator immediately calls .defer(trigger=S3KeyTrigger(...), method_name="execute_complete"). The worker slot is released back to the pool inside a few hundred milliseconds.
4. The Triggerer service picks up the Trigger, runs its async run() coroutine, and polls S3 every 60 seconds (the poke_interval). Hundreds of Triggers share the same 1-CPU Triggerer — the memory and CPU cost is roughly constant regardless of concurrency.
5. When the file lands, the Trigger yields a TriggerEvent. The scheduler re-queues the task, a worker picks it up, runs execute_complete(context, event) — which is basically a no-op in this case — and marks the task success. The worker slot was held for ~200ms of execute + ~200ms of execute_complete instead of 4 hours.

Output.

Metric	Poke mode (v1)	Deferrable (v2)	Delta
Slot-hours per sensor per day	4.0	0.02	-99.5%
Total slot-hours per day (400 sensors)	1600	8	-99.5%
Slot cost per day @ $0.08/hr	$128	$0.64	-$127.36
Triggerer cost per day (1 CPU)	$0	$1.20	+$1.20
Net cost per day	$128	$1.84	-$126.16 (~98.5%)
Payback on migration effort	—	~1 week engineering	~2 weeks calendar

Rule of thumb. Any sensor with a p50 wait longer than 60 seconds should be deferrable. The Triggerer overhead is O(1) in concurrency; the classic poke-mode cost is O(N × wait_hours). The break-even is essentially at zero waits.

Worked example — the 5-minute poke_interval trap

Detailed explanation. A second, subtler cost story: a team hears "poke every 5 minutes is fine" from a stale blog post and hard-codes poke_interval=300 everywhere. The team then discovers that mode=reschedule sensors incur a 3–10-second warmup on every cycle (DAG re-parse, task instance re-hydrate, cloud auth handshake). Over a 6-hour wait, 72 poke cycles × 5 seconds = 6 minutes of billed warmup on top of the intended sensor logic. Multiply across 400 sensors per day and you burn 40+ hours of pure warmup cost.

• The naive belief. mode=reschedule "releases" the worker and is therefore free.
• The reality. The task instance is re-created on every poke; re-parsing the DAG file, re-authenticating, re-instantiating the operator all cost billable CPU.
• The fix. deferrable=True runs the poll loop inside the Trigger's async run() — no DAG re-parse per cycle, no re-auth, no warmup.

Question. A DAG uses S3KeySensor with mode="reschedule" and poke_interval=300. The average wait is 3 hours and each reschedule cycle incurs 4 seconds of warmup cost. Quantify the wasted warmup CPU-seconds per day for 400 sensors, and show the deferrable fix.

Input.

Parameter	Value
Sensors per day	400
Average wait per sensor	3 hours
poke_interval	300 seconds
Poke cycles per sensor	36 (10800s / 300s)
Warmup cost per cycle	4 seconds
Worker CPU cost	$0.08/hour = ~$0.000022/second

Code.

# Classic reschedule mode — hidden warmup on every cycle
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor

wait_for_export = S3KeySensor(
    task_id="wait_for_export",
    bucket_key="s3://exports/{{ ds }}/data.parquet",
    aws_conn_id="aws_default",
    mode="reschedule",     # frees the worker between pokes
    poke_interval=300,     # but pays 4s warmup on every poke
    timeout=6 * 3600,
)

# 36 pokes × 4 s = 144 s wasted per sensor per day
# 400 sensors × 144 s = 57600 s = 16 CPU-hours/day of pure warmup

# Deferrable equivalent — Trigger owns the polling loop in the Triggerer's async loop
wait_for_export = S3KeySensor(
    task_id="wait_for_export",
    bucket_key="s3://exports/{{ ds }}/data.parquet",
    aws_conn_id="aws_default",
    deferrable=True,       # one flag flips the whole path
    poke_interval=60,      # inner cadence inside the async run(); no warmup per cycle
    timeout=6 * 3600,
)

# 0 s warmup per cycle × N cycles = 0 warmup cost
# Triggerer hosts the poll loop on a shared async event loop

Step-by-step explanation.

1. Under mode="reschedule", the sensor exits after each poke, releasing the worker. On the next scheduled poke, the scheduler re-launches the task instance — which re-parses the DAG file, re-instantiates the operator, re-authenticates against AWS. That's the 4 seconds of warmup.
2. The math: 36 pokes × 4 seconds × 400 sensors = 57,600 seconds of pure warmup per day. That's 16 CPU-hours/day of billable worker time doing nothing but starting up.
3. Under deferrable=True, the sensor's execute() is called once. It builds a Trigger, calls .defer(), releases the worker. The Trigger's async run() runs inside the Triggerer's persistent asyncio event loop. No DAG re-parse, no re-auth, no warmup per cycle.
4. The Triggerer's async run() might poll every 60 seconds (or even 30 seconds — cadence is now cheap). The polling is a single await asyncio.sleep(60) + a single S3 head call. The CPU cost per poll is dominated by the AWS API call, not any Airflow orchestration cost.
5. The saved 16 CPU-hours/day at $0.08/CPU-hour is $1.28/day — modest on its own, but stacked with the slot-hour saving from the previous example the deferrable path wins on every axis.

Output.

Cost line	reschedule mode	deferrable mode
Slot-hours held idle per day	0 (freed between pokes)	0
Warmup seconds per day	57,600	0
Warmup CPU-hours per day	16	0
Warmup cost per day	$1.28	$0
Poke API calls per day	14,400 (36 × 400)	14,400
Total wasted CPU per day	16 hrs	0 hrs

Rule of thumb. mode="reschedule" is not free; every poke pays a re-hydration cost. deferrable=True moves the poll loop inside a persistent async event loop where the per-poll cost is ~zero. Even at short polling intervals (30–60 s), deferrable dominates.

Worked example — sizing the Triggerer for 500 concurrent Triggers

Detailed explanation. A senior data engineer sizes the Triggerer for a platform that runs 500 concurrent sensors at peak. Rule of thumb: a single Triggerer with 1 CPU handles 100s to 1000s of concurrent Triggers because the workload is I/O-bound (mostly await asyncio.sleep(...) and network calls). Show the sizing math and the failure-mode considerations.

• Peak concurrent Triggers. 500 during the morning batch window.
• Per-Trigger CPU. ~10 ms per poll (AWS API round-trip) × 1 poll per minute = ~0.17% of one core per Trigger.
• Aggregate CPU. 500 × 0.17% = 85% of one core. Fits comfortably on 1 CPU with headroom.

Question. Produce the Helm-values sizing for a Triggerer that serves 500 concurrent Triggers with HA, and show the failure mode when a Triggerer crashes mid-wait.

Input.

Parameter	Value
Peak concurrent Triggers	500
Per-Trigger CPU cost	~0.17% of one core
Aggregate CPU at peak	~85% of one core
HA replica count	2
Memory per Trigger	~2 MB

Code.

# Helm values — Airflow official chart, triggerer sub-values
triggerer:
  # Two replicas for HA — the `trigger` metadata table de-dupes claims
  replicas: 2

  resources:
    requests:
      cpu: 500m       # 0.5 CPU baseline (headroom)
      memory: 1Gi
    limits:
      cpu: 2000m      # 2 CPU burst limit
      memory: 4Gi

  # The critical config knob — how many concurrent Triggers per instance
  extraArgs: ["--capacity=1000"]

  livenessProbe:
    initialDelaySeconds: 30
    periodSeconds: 30
    timeoutSeconds: 10
    failureThreshold: 3

  # Graceful termination — allow triggers to migrate before pod dies
  terminationGracePeriodSeconds: 60

  # PodDisruptionBudget — never let both replicas restart at once
  podDisruptionBudget:
    enabled: true
    maxUnavailable: 1

# Airflow config equivalent (airflow.cfg)
# [triggerer]
# default_capacity = 1000
# job_heartbeat_sec = 5
# triggerer_health_check_threshold = 30

Step-by-step explanation.

1. Two replicas give HA — the trigger table in the metadata DB holds a row per active Trigger with a triggerer_id claim column. If Triggerer 1 crashes, its heartbeat times out, its claims are released, and Triggerer 2 picks up the abandoned Triggers on the next poll.
2. --capacity=1000 sets the maximum number of concurrent Triggers per instance. With 500 peak Triggers and 2 replicas, each instance holds ~250 in steady state and can absorb up to 1000 during an incident. Cheap ceiling to raise; expensive to hit.
3. The CPU sizing (500m request, 2000m limit) allows for the ~85% steady-state CPU plus a 2× burst headroom for polling storms. Memory (1Gi request, 4Gi limit) accommodates ~2 MB per Trigger × 500 Triggers = 1 GB baseline.
4. terminationGracePeriodSeconds: 60 gives the Triggerer 60 seconds to gracefully hand off its claimed Triggers before Kubernetes forcibly kills the pod. Without this, a pod restart can orphan hundreds of Triggers for the 30-second heartbeat timeout.
5. The failure mode: Triggerer 1 OOM-killed at 03:00 with 250 claimed Triggers. Triggerer 2 detects the missing heartbeats within 30 seconds, adopts the 250 Triggers on its next scheduling cycle, and continues polling. From the DAG's perspective, the wait was extended by ~30 seconds — well inside SLA.

Output.

Sizing dimension	Value	Reasoning
Replicas	2	HA via shared trigger table
CPU request	500m	85% peak / 2 replicas = 42% each; 500m has headroom
CPU limit	2000m	2× burst for polling storms
Memory request	1Gi	2 MB × 500 Triggers
Memory limit	4Gi	Long-tail Triggers with large state
Capacity per instance	1000	2× peak per replica; cheap to raise
PodDisruptionBudget	maxUnavailable=1	Never lose both replicas at once

Rule of thumb. For most self-managed Airflow deployments, 2 Triggerer replicas × 1 CPU each is enough for 1000+ concurrent Triggers. Scale replica count only when a single instance's CPU exceeds 70% at peak — the async event loop is I/O-bound; CPU is rarely the bottleneck.

Senior interview question on the deferrable cost model

A senior interviewer often opens with: "Walk me through why an idle-sensor-heavy Airflow deployment burns 90% of its worker cost, what a deferrable operator changes at the runtime level, and how you'd quantify the savings for a FinOps review."

Solution Using the deferrable cost audit + Triggerer sizing

# Step 1 — audit the current idle-sensor cost
# Query the metadata DB (airflow_db) for classic-sensor slot-hours
"""
SELECT
    dag_id,
    task_id,
    operator,
    date(execution_date)                        AS run_date,
    EXTRACT(EPOCH FROM (end_date - start_date))/3600.0 AS slot_hours,
    state
FROM   task_instance
WHERE  operator LIKE '%Sensor%'
  AND  state = 'success'
  AND  execution_date >= now() - interval '30 days'
ORDER  BY slot_hours DESC
LIMIT  100;
"""

# Step 2 — migrate each hot sensor
# Before
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
S3KeySensor(
    task_id="wait_export",
    bucket_key="s3://exports/{{ ds }}/data.parquet",
    mode="reschedule",
    poke_interval=300,
    timeout=6*3600,
)

# After — one keyword
S3KeySensor(
    task_id="wait_export",
    bucket_key="s3://exports/{{ ds }}/data.parquet",
    deferrable=True,        # THE change
    poke_interval=60,       # cheap now; runs in the async loop
    timeout=6*3600,
)

# Step 3 — provision the Triggerer via Helm (2 replicas for HA)
# See the Helm sizing example above.

# Step 4 — post-migration audit
"""
SELECT
    dag_id,
    task_id,
    operator,
    date(execution_date)                        AS run_date,
    EXTRACT(EPOCH FROM (end_date - start_date))/3600.0 AS wall_hours,
    duration                                    AS worker_seconds
FROM   task_instance
WHERE  operator LIKE '%Sensor%'
  AND  state = 'success'
  AND  execution_date >= now() - interval '30 days'
ORDER  BY wall_hours DESC;
"""

# Step 5 — expected numbers
# slot_hours per sensor: 4.0 → 0.02 (2 sec execute + 2 sec execute_complete)
# monthly slot cost: $3840 → $19 for 400 sensors × 30 days
# Triggerer cost: +$36/month (2 replicas × 24 h × 30 d × $0.05)
# Net savings: $3785/month = 98.5% reduction

Step-by-step trace.

Step	Before (poke/reschedule)	After (deferrable)
Slot-hours per sensor	4.0 (idle wait)	0.02 (execute + execute_complete)
Sensors per day	400	400
Daily slot-hours	1600	8
Daily slot cost @ $0.08/hr	$128	$0.64
Daily Triggerer cost	$0	$1.20
Net daily cost	$128	$1.84
Monthly savings	—	~$3785
Migration effort	—	1 engineer-week
Payback	—	~2 days

After the rollout, the metadata-DB audit query confirms slot-hours have collapsed from 1600/day to 8/day. The Triggerer cost adds a fixed $1.20/day. The FinOps review captures a ~98.5% reduction on the sensor line, and the engineering time to migrate 400 sensors (mostly a codemod that flips deferrable=True) is one senior-week. Payback is 2 calendar days.

Output:

Cost dimension	Before	After
Idle slot-hours/day	1600	8
Slot cost/day	$128	$0.64
Triggerer cost/day	$0	$1.20
Net cost/day	$128	$1.84
Monthly cost	$3840	$55
Reduction	—	98.5%

Why this works — concept by concept:

• Slot release on defer — .defer() unblocks the worker before the wait begins; the worker returns to the pool and is available to run actual work. This is the architectural lever that turns O(concurrency × wait_hours) into O(1).
• Shared async event loop — the Triggerer hosts a single asyncio event loop over hundreds to thousands of Trigger run() coroutines. Because the workload is I/O-bound (await asyncio.sleep(...)), the CPU cost is roughly constant regardless of concurrency.
• HA via shared claim table — the trigger table in the metadata DB claims each Trigger to a specific triggerer_id. Multiple Triggerer instances share the table; when one dies, its claims are released and another picks them up on the next scheduling cycle.
• Poke_interval decoupled from worker cost — under deferrable, the inner poll cadence is cheap. Polling every 30 seconds is fine; polling every 300 seconds saves nothing meaningful. Choose the cadence based on the freshness requirement of the awaited event, not on cost.
• Cost — the migration cost is O(sensors) engineering time (usually a codemod), the Triggerer infrastructure cost is O(1) per platform, and the runtime cost per Trigger drops from O(wait_hours) to O(execute + resume). The FinOps math is unambiguous — the deferrable path wins by 90–99% on any workload where the average wait exceeds 60 seconds.

ETL
Topic — etl
ETL problems on sensor-driven orchestration and pipeline cost

Practice →

Optimization Topic — optimization Optimization problems on worker-slot cost and idle-time reduction

Practice →

---

2. Deferrable operator lifecycle

execute() → .defer() → Triggerer async loop → TriggerEvent → worker method_name(event) — five phases, three components

The mental model in one line: a deferrable operator's execute() runs briefly on a worker, hits a wait, calls .defer() to hand off to the Triggerer's async event loop, sleeps for hours or days as an asyncio coroutine, yields a TriggerEvent when the awaited condition fires, and re-hydrates on a worker to run method_name(event) for the final result-handling work. Every question about airflow deferrable operator internals, the defer method semantics, airflow async runtime behaviour, asyncio airflow gotchas, and long-running task airflow architecture is a downstream consequence of that lifecycle.

The five phases in detail.

• Phase 1 — execute(context) on worker. The operator's regular execute() runs on a worker slot. It sets up connections, computes any pre-wait state (e.g. the S3 bucket + key it will watch), and constructs a Trigger instance holding all the state the async wait needs.
• Phase 2 — .defer(trigger, method_name). The operator calls self.defer(trigger=my_trigger, method_name="execute_complete"). This is a special Airflow control-flow call — it raises a TaskDeferred exception internally, which the worker's task-runner catches, serialises the Trigger to the metadata DB, marks the task instance as deferred, and releases the slot.
• Phase 3 — Triggerer picks up the Trigger. One of the running Triggerer instances claims the newly-created row in the trigger table, deserialises the Trigger, and starts its run() coroutine on the shared asyncio event loop. The Trigger now lives entirely inside the Triggerer's process memory.
• Phase 4 — TriggerEvent fires. Inside run(), the coroutine polls (or subscribes to a webhook, or sleeps until a wall-clock time). When the awaited condition is met, run() executes yield TriggerEvent({"payload": ...}). The yielded event is captured by the Triggerer, written to the metadata DB, and the task instance is re-queued for execution.
• Phase 5 — method_name(context, event) on worker. A worker picks up the re-queued task and calls the method named in the .defer() call — conventionally execute_complete(self, context, event). This method receives the TriggerEvent's payload as event, performs any final work (extract the key, push an XCom, mark success), and returns.

Why this contract is the whole thing.

• Worker holds the slot only during Phase 1 and Phase 5. Everything between is on the Triggerer, which is not billed per-slot.
• The metadata DB is the message bus. No direct worker-Triggerer network channel; both read/write to the trigger and task_instance tables. The scheduler orchestrates.
• method_name is the resume point. Convention is execute_complete, but any method on the operator class works. This lets you split the pre-wait and post-wait logic cleanly.
• The Trigger is pickled state. serialize() returns (classpath, kwargs); the Triggerer imports the class and calls __init__(**kwargs) to rebuild it. Anything not in kwargs is lost.

Common gotchas around the lifecycle.

• Never call blocking I/O in run(). time.sleep(), synchronous requests.get(), blocking database calls — any of these blocks the whole asyncio event loop and freezes every other Trigger. Use asyncio.sleep(), aiohttp, asyncpg, etc.
• execute() runs before the defer. Whatever execute() does is billable on the worker slot. Keep it thin — validate inputs, build the Trigger, defer. Don't do expensive setup pre-defer.
• method_name runs after resume. State from execute() is lost unless carried in the Trigger's kwargs or in the TriggerEvent payload. Design the API accordingly.
• Retries re-run execute(). If a Trigger errors out and the task retries, execute() runs from scratch. Not resume-safe from method_name.

Common interview probes on the lifecycle.

• "Walk me through the five phases of a deferrable operator." — required answer is execute → defer → Triggerer.run → TriggerEvent → method_name.
• "What happens to the worker slot during the wait?" — released the moment .defer() is called.
• "Where does the Trigger's state live during the wait?" — inside the Triggerer's async event loop memory + a row in the trigger metadata table.
• "What's the difference between execute() and execute_complete()?" — execute runs pre-wait; execute_complete runs post-wait; they can be on the same worker or different workers.

Worked example — S3KeySensorAsync deferring for 6 hours

Detailed explanation. Walk through the classic case — an S3 key sensor that waits 6 hours for an upstream export to land. Show the timeline of worker + Triggerer activity, the state transitions of the task instance in the metadata DB, and the exact CPU-seconds consumed at each phase.

• Wall-clock timeline. 06:00 execute → 06:00 defer → 12:00 file lands → TriggerEvent → 12:00 execute_complete → 12:00 success.
• Worker slot occupancy. 06:00:00 to 06:00:01 (1 second on execute) + 12:00:00 to 12:00:01 (1 second on execute_complete) = 2 seconds total.
• Triggerer activity. Continuous polling every 60 seconds from 06:00 to 12:00 = 360 poll cycles × ~30ms each = ~11 seconds of async CPU.

Question. Trace through the metadata DB state transitions and the worker + Triggerer CPU consumption for a single S3KeySensorAsync task that defers at 06:00 and resumes at 12:00.

Input.

Time	Event
06:00:00.0	Scheduler queues task
06:00:00.1	Worker picks up task; runs execute()
06:00:00.5	execute() calls .defer(...)
06:00:00.6	Task state → deferred; Trigger written to trigger table
06:00:01.0	Triggerer 1 claims trigger; starts async run()
06:00–12:00	Triggerer polls S3 every 60 seconds (360 polls)
12:00:00.0	S3 head returns 200 OK
12:00:00.1	run() yields TriggerEvent({"key": "data.parquet"})
12:00:00.2	Scheduler re-queues task
12:00:00.3	Worker picks up task; runs execute_complete(context, event)
12:00:00.5	Task state → success

Code.

# S3KeySensorAsync — the shape of the deferrable operator internals
# (simplified, pseudo-source based on the Amazon Provider)
from airflow.sensors.base import BaseSensorOperator
from airflow.triggers.base import BaseTrigger, TriggerEvent
import aiobotocore.session
import asyncio


class S3KeyTrigger(BaseTrigger):
    def __init__(self, bucket: str, key: str, aws_conn_id: str, poke_interval: int = 60):
        super().__init__()
        self.bucket = bucket
        self.key = key
        self.aws_conn_id = aws_conn_id
        self.poke_interval = poke_interval

    def serialize(self):
        return (
            "myprovider.triggers.S3KeyTrigger",
            {
                "bucket": self.bucket,
                "key": self.key,
                "aws_conn_id": self.aws_conn_id,
                "poke_interval": self.poke_interval,
            },
        )

    async def run(self):
        session = aiobotocore.session.get_session()
        async with session.create_client("s3") as s3:
            while True:
                try:
                    await s3.head_object(Bucket=self.bucket, Key=self.key)
                    yield TriggerEvent({"status": "success", "key": self.key})
                    return
                except s3.exceptions.NoSuchKey:
                    await asyncio.sleep(self.poke_interval)


class S3KeySensorAsync(BaseSensorOperator):
    def __init__(self, bucket_key: str, aws_conn_id: str = "aws_default",
                 poke_interval: int = 60, timeout: int = 60 * 60 * 24, **kw):
        super().__init__(**kw)
        self.bucket_key = bucket_key
        self.aws_conn_id = aws_conn_id
        self.poke_interval = poke_interval
        self.timeout = timeout

    def execute(self, context):
        # Runs briefly on worker, then defers
        bucket, key = self._parse(self.bucket_key)
        self.defer(
            trigger=S3KeyTrigger(
                bucket=bucket, key=key,
                aws_conn_id=self.aws_conn_id,
                poke_interval=self.poke_interval,
            ),
            method_name="execute_complete",
            timeout=self.timeout,
        )

    def execute_complete(self, context, event: dict):
        # Runs briefly on worker after TriggerEvent
        if event.get("status") != "success":
            raise RuntimeError(f"S3 wait failed: {event}")
        self.log.info("S3 key arrived: %s", event["key"])
        return event["key"]

    def _parse(self, key_uri: str) -> tuple[str, str]:
        # s3://bucket/prefix/file → (bucket, prefix/file)
        without_scheme = key_uri.replace("s3://", "", 1)
        bucket, _, key = without_scheme.partition("/")
        return bucket, key

Step-by-step explanation.

1. Phase 1 (worker, ~0.5 s): execute() parses s3://exports/data.parquet into ("exports", "data.parquet"), constructs an S3KeyTrigger(bucket="exports", key="data.parquet", ...), and calls self.defer(trigger=..., method_name="execute_complete"). Airflow raises TaskDeferred; the worker's task runner catches it, calls trigger.serialize(), writes the row to the trigger table, marks the task_instance as deferred, and returns the worker slot to the pool.
2. Phase 2 (Triggerer, ~1 second later): A Triggerer instance's scheduler-loop notices an unclaimed row in trigger, claims it by writing its triggerer_id on the row, calls S3KeyTrigger.__init__(**kwargs) from the serialised classpath, and schedules the coroutine S3KeyTrigger.run() on its asyncio event loop.
3. Phase 3 (Triggerer, 6 hours): run() enters its polling loop. Each iteration: await s3.head_object(...) (30–50 ms), on NoSuchKey catches the error, await asyncio.sleep(60). Every other Trigger on the event loop runs concurrently. Aggregate CPU across 360 polls: ~11 seconds.
4. Phase 4 (Triggerer, ~50 ms): The 361st poll returns 200 OK. run() executes yield TriggerEvent({"status": "success", "key": "data.parquet"}). The Triggerer catches the event, writes it to the trigger_event metadata, updates the task_instance state to scheduled, and the coroutine exits.
5. Phase 5 (worker, ~0.5 s): The scheduler picks up the newly-scheduled task, a worker claims it, and calls operator.execute_complete(context, event) with event = {"status": "success", "key": "data.parquet"}. The method logs, pushes the key to XCom (via return value), and returns. Task state → success.

Output.

Component	Wall-clock time	CPU seconds consumed	Slot-hours billed
Worker (execute)	06:00:00.5	0.5 s	0.5/3600 = 0.00014 h
Metadata DB write	06:00:00.6	0.01 s	(not billed)
Triggerer (run 6 h)	06:00–12:00	11 s (async)	(not billed per slot)
Metadata DB write	12:00:00.1	0.01 s	(not billed)
Worker (execute_complete)	12:00:00.3	0.5 s	0.00014 h
Total	6 hours wall	~12 s CPU	~0.0003 slot-hours

Rule of thumb. In a well-designed deferrable operator, the worker consumes ~1 second of slot-hours regardless of how long the wait is. All the wall-clock time lives in the Triggerer's async loop, where the marginal cost per Trigger is roughly zero.

Worked example — the TaskDeferred exception in the source

Detailed explanation. Understanding how .defer() actually works clarifies the whole model. .defer() doesn't return — it raises a TaskDeferred exception. The worker's task runner catches this exception in the same way it catches success or failure; the code path is symmetric. Walk through the mechanism.

• .defer() raises. Look at the Airflow source: raise TaskDeferred(trigger=trigger, method_name=method_name, timeout=timeout).
• Worker catches. The _execute_task wrapper in airflow.models.taskinstance catches TaskDeferred and handles it as a state transition, not an error.
• State transition. Task state → deferred; row written to trigger table; worker slot released.

Question. Trace through what happens if execute() calls .defer() inside a try/except Exception. What breaks?

Input.

Component	Behaviour
.defer()	raises TaskDeferred
try/except Exception	catches TaskDeferred as a generic exception
Result	task defer is silently swallowed

Code.

# Wrong — catches TaskDeferred as a generic Exception
from airflow.sensors.base import BaseSensorOperator
from airflow.exceptions import TaskDeferred

class BuggyOperator(BaseSensorOperator):
    def execute(self, context):
        try:
            self.defer(trigger=..., method_name="execute_complete")
        except Exception as e:
            # BUG: catches the TaskDeferred that .defer() raised!
            # The task never actually defers — it just returns None.
            self.log.error("defer failed: %s", e)
            return None

# Right — either don't wrap in try/except, or explicitly re-raise TaskDeferred
class CorrectOperator(BaseSensorOperator):
    def execute(self, context):
        try:
            # do pre-wait validation
            if not self._can_wait():
                raise AirflowException("nothing to wait for")
            self.defer(trigger=..., method_name="execute_complete")
        except TaskDeferred:
            # re-raise so the worker task runner catches it
            raise
        except Exception as e:
            self.log.error("pre-defer error: %s", e)
            raise

# Even better — use except (Exception, TaskDeferred) explicitly
# or narrow the catch to specific exceptions you expect
class BestOperator(BaseSensorOperator):
    def execute(self, context):
        # Validate up front
        if not self._can_wait():
            raise AirflowException("nothing to wait for")
        # Defer with no try/except — let TaskDeferred bubble
        self.defer(trigger=..., method_name="execute_complete")

Step-by-step explanation.

1. .defer() is not a normal method — it raises TaskDeferred(trigger, method_name, timeout). The exception carries the Trigger reference so the task runner can serialise and persist it.
2. The worker's _execute_task wrapper has an explicit except TaskDeferred as td: clause that handles the state transition — write to trigger table, mark instance deferred, release slot. Any other exception bubbles up as an error.
3. If your execute() wraps the defer in a generic try/except Exception, you catch the TaskDeferred too. The worker runner never sees it, the task runner returns from execute() normally, and the task is marked success with no deferral. The wait never happens.
4. The two safe patterns: (a) don't wrap .defer() in a try/except at all — let the exception propagate naturally; (b) if you need pre-defer error handling, except TaskDeferred: raise at the top of your except clause to re-raise.
5. The buggy pattern is a subtle bug because tests may still pass — a mocked S3 client returns the key immediately, the operator "works," and only in production with a real wait does the missing deferral surface as "task finishes instantly, downstream runs against a non-existent file."

Output.

Pattern	Wraps .defer in try/except?	Behaviour
BuggyOperator	yes (except Exception)	swallows TaskDeferred; wait never happens
CorrectOperator	yes (with except TaskDeferred: raise)	works
BestOperator	no try/except around .defer	works, cleanest

Rule of thumb. Never wrap self.defer(...) in a try/except Exception. TaskDeferred is a control-flow exception; catching it silently breaks the deferrable contract. If you need pre-defer error handling, put the try/except around the validation logic before .defer(), not around the .defer() call itself.

Worked example — passing state through TriggerEvent

Detailed explanation. The TriggerEvent.payload is a dict — it's the only way state flows from the async wait back to the resume method. Design the payload carefully; anything not in it is lost. Walk through a realistic example where the Trigger needs to pass the discovered S3 key, the object's ETag, and the size back to execute_complete for downstream XCom.

• What the Trigger discovers. Which of several possible keys matched, the object's ETag (for downstream idempotency), the object size (for downstream cost estimation).
• What execute_complete needs. All three, to push into XCom for the next task.
• What must NOT be in the payload. Non-JSON-serialisable objects (boto3 clients, aiohttp sessions, non-primitive Python objects). The payload is written to the metadata DB as JSON.

Question. Design the Trigger and the execute_complete method for an S3 wait that resolves any of several possible keys (glob-like), returns the matched key + ETag + size, and pushes them as XComs.

Input.

Field	Type	Purpose
status	str	"success" or "error"
matched_key	str	which key resolved
etag	str	for idempotency downstream
size_bytes	int	for cost estimation downstream

Code.

from airflow.triggers.base import BaseTrigger, TriggerEvent
import aiobotocore.session
import asyncio


class S3AnyKeyTrigger(BaseTrigger):
    def __init__(self, bucket: str, candidate_keys: list[str],
                 aws_conn_id: str, poke_interval: int = 60):
        super().__init__()
        self.bucket = bucket
        self.candidate_keys = candidate_keys
        self.aws_conn_id = aws_conn_id
        self.poke_interval = poke_interval

    def serialize(self):
        # Return everything __init__ needs to rebuild
        return (
            "myprovider.triggers.S3AnyKeyTrigger",
            {
                "bucket": self.bucket,
                "candidate_keys": self.candidate_keys,
                "aws_conn_id": self.aws_conn_id,
                "poke_interval": self.poke_interval,
            },
        )

    async def run(self):
        session = aiobotocore.session.get_session()
        async with session.create_client("s3") as s3:
            while True:
                for key in self.candidate_keys:
                    try:
                        resp = await s3.head_object(Bucket=self.bucket, Key=key)
                        # Extract the state we want to hand to execute_complete
                        yield TriggerEvent({
                            "status": "success",
                            "matched_key": key,
                            "etag": resp["ETag"].strip('"'),
                            "size_bytes": resp["ContentLength"],
                        })
                        return
                    except s3.exceptions.ClientError:
                        continue
                await asyncio.sleep(self.poke_interval)

# Operator side — execute_complete receives event dict
class S3AnyKeySensorAsync(BaseSensorOperator):
    def execute(self, context):
        self.defer(
            trigger=S3AnyKeyTrigger(
                bucket=self.bucket,
                candidate_keys=self.candidate_keys,
                aws_conn_id=self.aws_conn_id,
                poke_interval=self.poke_interval,
            ),
            method_name="execute_complete",
            timeout=self.timeout,
        )

    def execute_complete(self, context, event: dict):
        if event.get("status") != "success":
            raise RuntimeError(f"S3 wait failed: {event}")
        # Push all three pieces of state as XComs
        ti = context["ti"]
        ti.xcom_push(key="matched_key",  value=event["matched_key"])
        ti.xcom_push(key="etag",         value=event["etag"])
        ti.xcom_push(key="size_bytes",   value=event["size_bytes"])
        return event["matched_key"]     # legacy return-value XCom

Step-by-step explanation.

1. The Trigger's serialize() returns the full state the Triggerer needs to rebuild the object — bucket, candidate keys, connection id, poke interval. Anything not in this dict is unavailable to the async run() after the initial pre-defer execute().
2. run() polls each candidate key on every cycle. The first one that returns 200 wins; the Trigger extracts the ETag and size from the response and includes them in the TriggerEvent payload.
3. The payload is JSON-serialised by Airflow and written to the trigger_event metadata. All values must be JSON-compatible (primitives, lists, dicts of primitives). Non-serialisable values (datetime objects, boto3 responses) must be converted first.
4. When the task resumes on a worker, execute_complete(context, event) receives the payload as event. It pushes each piece to XCom under a stable key so downstream tasks can pull matched_key, etag, size_bytes explicitly.
5. The return value of execute_complete is also pushed as the "default" XCom (the legacy return_value key). Downstream tasks can pull either the named keys or the return value depending on preference.

Output.

Field passed via TriggerEvent	Value example	Downstream XCom key
status	"success"	(checked in method)
matched_key	"data-2026-06-22.parquet"	matched_key
etag	"abc123def456"	etag
size_bytes	104857600	size_bytes

Rule of thumb. Treat TriggerEvent.payload as a JSON message contract. Include every piece of state the resume method needs; nothing else is available. Prefer named XCom pushes over the return-value default for anything used by multiple downstream consumers.

Senior interview question on the lifecycle contract

A senior interviewer might ask: "Explain the deferrable operator lifecycle end-to-end — what runs on the worker, what runs on the Triggerer, where the state lives during the wait, and what happens if the Triggerer that owns a Trigger crashes mid-wait."

Solution Using the five-phase model + HA claim table

# The full lifecycle as Airflow implements it
# 1. Worker runs execute()
def execute(self, context):
    # brief pre-wait setup
    validated_inputs = self._validate(context)
    trigger = MyTrigger(**validated_inputs)
    # 2. .defer() raises TaskDeferred; worker slot released
    self.defer(
        trigger=trigger,
        method_name="execute_complete",
        timeout=self.timeout,
    )

# 3. Triggerer async loop hosts the run() coroutine
# (inside airflow.jobs.triggerer_job.TriggererJob)
#
# for trigger_row in claim_next_triggers():
#     trigger = deserialize(trigger_row.classpath, trigger_row.kwargs)
#     loop.create_task(_run_trigger(trigger_row.id, trigger))
#
# async def _run_trigger(trigger_id, trigger):
#     async for event in trigger.run():
#         persist_event(trigger_id, event)
#         mark_task_scheduled(trigger_id)

# 4. When a Triggerer crashes, its claims release via heartbeat timeout
# (inside airflow.jobs.triggerer_job.TriggererJobRunner)
#
# UPDATE trigger
# SET    triggerer_id = NULL
# WHERE  triggerer_id IN (
#     SELECT id FROM triggerer_job
#     WHERE  last_heartbeat < now() - interval '30 seconds'
# );
# Another Triggerer claims the freed rows on next scheduling cycle.

# 5. Worker runs method_name(context, event) on resume
def execute_complete(self, context, event: dict):
    if event["status"] != "success":
        raise RuntimeError(event)
    # brief post-wait handling
    self._finalise(context, event)
    return event["result"]

Step-by-step trace.

Phase	Component	State transition	Worker slot	Triggerer CPU
1. execute()	Worker	queued → running	held	—
2. .defer() raises	Worker	running → deferred; trigger row inserted	released	—
3. run() picked up	Triggerer	trigger.triggerer_id claimed	—	active
4a. Triggerer crash	(mid-wait)	heartbeat times out; triggerer_id cleared	—	(crashed)
4b. Recovery	Triggerer 2	triggerer_id re-claimed	—	active on Triggerer 2
4c. TriggerEvent	Triggerer 2	event persisted; task → scheduled	—	—
5. execute_complete	Worker	scheduled → running → success	held (briefly)	—

The HA story is entirely in the metadata DB — no Triggerer-to-Triggerer network channel exists. Every Triggerer heartbeats every 5 seconds; a heartbeat older than 30 seconds marks the Triggerer dead. The dead Triggerer's claimed triggers have their triggerer_id cleared; the next scheduling cycle on any live Triggerer picks them up. Wait time is extended by ~30 seconds worst case; no Trigger is lost.

Output:

Concern	Answer
Where does state live during the wait?	trigger metadata row + Triggerer process memory
Who holds the worker slot?	Nobody (released on defer)
Who runs the async coroutine?	The Triggerer's asyncio event loop
What happens on Triggerer crash?	Heartbeat times out; claims released; another Triggerer picks up
Worst-case wait extension	~30 seconds (heartbeat timeout)
Can a Trigger be lost?	No — the trigger row persists across restarts

Why this works — concept by concept:

• Metadata DB as the message bus — the worker never talks to the Triggerer directly. Both sides read/write to the trigger and task_instance tables. This is what makes HA cheap: any Triggerer with DB access can pick up any Trigger.
• Claim + heartbeat = ownership — a Trigger belongs to whichever Triggerer wrote its own triggerer_id into the row. Heartbeats keep the claim alive; missing heartbeats release the claim automatically. Standard leader-election style with the DB as the coordinator.
• TaskDeferred as control flow — .defer() raises rather than returns because the state transition (running → deferred) must be handled by the worker's task runner, not the operator itself. Same mechanism as AirflowSkipException and AirflowRescheduleException.
• TriggerEvent is a JSON contract — everything the resume method needs must be in the payload dict. No shared memory, no cached objects. This decouples the worker that ran execute() from the worker that runs execute_complete().
• Cost — the worker slot is billed for ~1 second regardless of wait duration; the Triggerer runs the wait at O(1) marginal cost per Trigger; the HA layer costs one metadata-DB row per active Trigger. The economics are unambiguous.

ETL
Topic — etl
ETL problems on deferred / async orchestration lifecycles

Practice →

SQL Topic — sql SQL problems on task-instance and trigger-table auditing

Practice →

---

3. Writing a custom Trigger class

class MyTrigger(BaseTrigger) — a three-method contract: __init__ state, serialize() classpath + kwargs, async run() yield TriggerEvent

The mental model in one line: a custom Trigger is a subclass of airflow.triggers.base.BaseTrigger that implements exactly three things — an __init__ that captures the state the wait needs, a serialize() that returns (classpath, kwargs) for reconstruction, and an async def run() that awaits the condition and yields a TriggerEvent when it fires. Every question about airflow trigger class design, asyncio airflow gotchas, custom-provider work, and internal API polling reduces to that three-part contract.

The three-part contract.

• `init(kwargs).** Capture everything the wait needs — the polling URL, credentials, the target condition, the polling cadence. Store as self.* attributes. Don't open connections or async clients here; those belong in run()`.
• serialize(self) -> tuple[str, dict]. Return (classpath, kwargs) — the classpath is the importable dotted path (e.g. "myco.triggers.MyTrigger"), the kwargs is a dict that, when splatted into __init__, rebuilds the Trigger. The Triggerer imports the class and reconstructs the Trigger on every startup.
• async def run(self) -> AsyncGenerator[TriggerEvent, None]. The async coroutine. Structured as async def with yield TriggerEvent(...) — it's an async generator. Poll, wait, yield the event when done. The Triggerer collects the first yielded event and treats it as the resume signal.

The four rules of async run().

• Only awaitable I/O. await asyncio.sleep(...), await aiohttp.ClientSession().get(...), await asyncpg.fetch(...). Never time.sleep(), never synchronous requests.get(), never blocking DB drivers.
• Yield exactly one TriggerEvent (typically). Multi-yield triggers exist but are rare. In 99% of cases, run() loops until the condition is met, yields once, and returns.
• Handle transient errors. Network blips, 5xx from the polled service, temporary auth expiries — all should be caught, logged, and re-tried with backoff. Uncaught exceptions in run() propagate as a failed TriggerEvent (with status="error") and the task fails.
• Respect cancellation. If the Trigger is cancelled (e.g. task retry, DAG paused), the asyncio event loop cancels run(). Use try/finally or try/except asyncio.CancelledError to clean up any open connections.

Common Trigger patterns.

• Polling loop. while True: check(); await asyncio.sleep(interval). The bread-and-butter — used for S3, GCS, HTTP, database queries.
• Long sleep + single check. await asyncio.sleep(target - now); yield TriggerEvent(...). Used for time-based waits (DateTimeTrigger, TimeDeltaTrigger).
• Webhook / SSE / WebSocket. async for event in ws: ...; yield TriggerEvent(...). Used for real-time notifications where polling is wasteful.
• Composite condition. while True: if a() and b(): break; await asyncio.sleep(...). Used when the resume condition depends on multiple external systems.

Common gotchas.

• serialize() must be idempotent. Called on every restart; must return the same output every time for a given Trigger state.
• serialize() must not include non-JSON. kwargs is JSON-encoded. Datetime → ISO string; complex objects → primitive form.
• run() is an async generator, not a coroutine. Must contain yield; must be defined as async def.
• Long sleeps are fine. await asyncio.sleep(86400) (one day) is a legal wait. The asyncio event loop handles it correctly.

Common interview probes.

• "Write me a Trigger that polls a REST API every 30 seconds until it returns status=200." — required implementation with aiohttp + asyncio.sleep.
• "What's the difference between serialize() and __init__?" — serialize produces the reconstruction recipe; __init__ consumes it.
• "How do you handle a transient 5xx from the polled service?" — try/except + exponential backoff inside run().
• "What happens if you use time.sleep() instead of asyncio.sleep()?" — blocks the whole event loop; freezes every Trigger.

Worked example — Trigger polling an internal REST API every 30s

Detailed explanation. A common custom-Trigger case: a data platform waits for an internal "batch ready" REST endpoint to return {"status": "ready"}. The team wants a Trigger that polls every 30 seconds, handles transient 5xx errors with exponential backoff, and times out after 6 hours. Walk through the full implementation.

• Endpoint. GET https://batch.internal/api/v1/batches/{batch_id}/status.
• Success condition. JSON body {"status": "ready"}.
• Failure. 4xx (except 429), or 6-hour timeout.
• Retryable. 429, 5xx, network errors.

Question. Implement BatchReadyTrigger with async polling, exponential backoff on retryable errors, and a BatchReadyOperator that uses it.

Input.

Parameter	Value
Endpoint	https://batch.internal/api/v1/batches/{batch_id}/status
Poll cadence	30 seconds
Backoff base	1 s
Backoff cap	60 s
Total timeout	6 hours
Auth	Bearer token from Airflow Variable

Code.

import asyncio
import aiohttp
from airflow.triggers.base import BaseTrigger, TriggerEvent
from airflow.sensors.base import BaseSensorOperator


class BatchReadyTrigger(BaseTrigger):
    def __init__(self, batch_id: str, endpoint_base: str, token: str,
                 poll_interval: int = 30):
        super().__init__()
        self.batch_id = batch_id
        self.endpoint_base = endpoint_base
        self.token = token
        self.poll_interval = poll_interval

    def serialize(self):
        return (
            "myco.triggers.batch.BatchReadyTrigger",
            {
                "batch_id": self.batch_id,
                "endpoint_base": self.endpoint_base,
                "token": self.token,
                "poll_interval": self.poll_interval,
            },
        )

    async def run(self):
        url = f"{self.endpoint_base}/batches/{self.batch_id}/status"
        headers = {"Authorization": f"Bearer {self.token}"}
        backoff = 1.0
        try:
            async with aiohttp.ClientSession() as session:
                while True:
                    try:
                        async with session.get(url, headers=headers,
                                               timeout=aiohttp.ClientTimeout(total=10)) as resp:
                            if resp.status == 200:
                                body = await resp.json()
                                if body.get("status") == "ready":
                                    yield TriggerEvent({
                                        "status": "success",
                                        "batch_id": self.batch_id,
                                        "payload": body,
                                    })
                                    return
                                # Still processing — reset backoff, poll again
                                backoff = 1.0
                                await asyncio.sleep(self.poll_interval)
                            elif resp.status in (429, 500, 502, 503, 504):
                                # Retryable — exponential backoff
                                await asyncio.sleep(min(backoff, 60))
                                backoff *= 2
                            else:
                                # 4xx (not 429) → give up
                                body = await resp.text()
                                yield TriggerEvent({
                                    "status": "error",
                                    "reason": f"http {resp.status}",
                                    "body": body[:1000],
                                })
                                return
                    except (aiohttp.ClientError, asyncio.TimeoutError) as e:
                        # Network error — retryable with backoff
                        await asyncio.sleep(min(backoff, 60))
                        backoff *= 2
        except asyncio.CancelledError:
            # Clean shutdown when task is cancelled
            raise

class BatchReadyOperator(BaseSensorOperator):
    def __init__(self, batch_id: str, endpoint_base: str,
                 token_variable: str = "batch_api_token",
                 poll_interval: int = 30, timeout: int = 6 * 3600, **kw):
        super().__init__(**kw)
        self.batch_id = batch_id
        self.endpoint_base = endpoint_base
        self.token_variable = token_variable
        self.poll_interval = poll_interval
        self.timeout = timeout

    def execute(self, context):
        from airflow.models import Variable
        token = Variable.get(self.token_variable)
        self.defer(
            trigger=BatchReadyTrigger(
                batch_id=self.batch_id,
                endpoint_base=self.endpoint_base,
                token=token,
                poll_interval=self.poll_interval,
            ),
            method_name="execute_complete",
            timeout=self.timeout,
        )

    def execute_complete(self, context, event: dict):
        if event.get("status") != "success":
            raise RuntimeError(f"Batch wait failed: {event}")
        self.log.info("Batch %s ready", event["batch_id"])
        return event["payload"]

Step-by-step explanation.

1. __init__ captures the four pieces of state — batch id, endpoint base, token, poll interval. All are primitives; all can be JSON-serialised. The token is passed in (not fetched inside the Trigger) so the Trigger doesn't need a Variable client dependency.
2. serialize() returns the classpath (importable dotted path) and the kwargs dict. When the Triggerer restarts or the Trigger is reassigned, it imports the class from myco.triggers.batch and calls BatchReadyTrigger(**kwargs) to rebuild.
3. run() opens an aiohttp session and enters the polling loop. On success (200 + ready), yields TriggerEvent and returns. On retryable errors (429, 5xx, network), sleeps with exponential backoff up to 60 seconds. On non-retryable errors (4xx except 429), yields an error TriggerEvent and returns.
4. The try: ... except asyncio.CancelledError: raise at the outer level ensures that if the task is cancelled (retry, DAG paused, Triggerer graceful shutdown), the aiohttp session is closed cleanly via the async with context manager.
5. The operator's execute() reads the token from Airflow Variable (a sync operation, fine to do pre-defer), builds the Trigger, and defers. execute_complete receives the payload dict and returns it as the XCom.

Output.

Poll cycle	Outcome	Sleep	Backoff
1	200 + still processing	30 s	reset to 1
2	200 + still processing	30 s	1
3	503 (transient)	1 s	→ 2
4	503 (transient)	2 s	→ 4
5	200 + still processing	30 s	reset to 1
…
720	200 + ready	(yields TriggerEvent)	—

Rule of thumb. A production Trigger has three loops: the outer polling loop, the inner retry-with-backoff loop for transient errors, and the timeout guard. Reset the backoff to 1 on every successful poll cycle so a transient error doesn't slow down subsequent healthy polls.

Worked example — DateTimeTrigger for time-based waits

Detailed explanation. A subtler use case: a DAG needs to wait until 03:00 UTC the next day before running. The classic TimeDeltaSensor holds a worker slot; the deferrable equivalent uses DateTimeTrigger which sleeps in the async loop until the target time. Show the implementation and the interview signal — "async sleep for hours is fine because the event loop only wakes on the specified time."

• Use case. DAG runs at 10:00; needs to wait until 03:00 next day.
• Wait duration. 17 hours.
• Classic cost. 17 hours × 1 slot × $0.08/hr = $1.36 per run, per DAG.
• Deferrable cost. ~0 (async sleep is a single-timer event on the loop).

Question. Implement DateTimeTrigger from scratch (the built-in exists, but showing the source clarifies the mechanism).

Input.

Parameter	Value
Target time	2026-06-23 03:00:00 UTC
Wait duration	17 hours
Concurrency in Triggerer	1000 such Triggers on one CPU

Code.

import asyncio
from datetime import datetime, timezone
from airflow.triggers.base import BaseTrigger, TriggerEvent
from airflow.sensors.base import BaseSensorOperator


class DateTimeTrigger(BaseTrigger):
    def __init__(self, moment: str):
        # ISO-format string for JSON-safe serialization
        super().__init__()
        self.moment = moment

    def serialize(self):
        return (
            "myco.triggers.datetime.DateTimeTrigger",
            {"moment": self.moment},
        )

    async def run(self):
        target = datetime.fromisoformat(self.moment)
        while True:
            now = datetime.now(tz=timezone.utc)
            remaining = (target - now).total_seconds()
            if remaining <= 0:
                yield TriggerEvent({"status": "success",
                                    "reached": self.moment})
                return
            # Sleep at most one hour, then re-check.
            # Long single sleeps are fine, but chunking lets the event loop
            # respond to shutdowns / cancellations more responsively.
            await asyncio.sleep(min(remaining, 3600))

class DateTimeSensorAsync(BaseSensorOperator):
    def __init__(self, target_iso: str, **kw):
        super().__init__(**kw)
        self.target_iso = target_iso

    def execute(self, context):
        self.defer(
            trigger=DateTimeTrigger(moment=self.target_iso),
            method_name="execute_complete",
        )

    def execute_complete(self, context, event):
        return event["reached"]

Step-by-step explanation.

1. __init__ stores the target moment as an ISO-format string. Datetime objects aren't JSON-serialisable directly; strings are. The Trigger converts back to datetime inside run().
2. serialize() returns the classpath and the ISO string. If the Trigger is restarted (Triggerer crash + re-claim), the same target moment is preserved.
3. run() computes the remaining seconds until the target and sleeps. Long asyncio.sleep() calls are handled by the event loop as a single timer registration — no CPU cost during the wait. The 1-hour chunk lets the loop respond to CancelledError within an hour of a task cancellation.
4. When the target is reached, run() yields TriggerEvent with the reached moment and returns. The Triggerer persists the event; the scheduler re-queues the task.
5. The critical insight: 1000 DateTimeTriggers on a single Triggerer cost roughly nothing at runtime. Each is a single timer entry in the asyncio loop. The event loop wakes when the earliest timer fires, dispatches the yield, and goes back to sleep.

Output.

Metric	Classic TimeDeltaSensor	DateTimeTrigger (deferrable)
Worker slots held during wait	1	0
CPU during wait	poll-cycle × warmup	0 (event loop dormant)
Wait cost @ 17 hours × $0.08/hr	$1.36	$0
1000 concurrent waits, cost	$1360	~$0
Triggerer memory per Trigger	—	~2 KB (timer entry + Trigger state)

Rule of thumb. Time-based waits are the lowest-cost deferrable case — a single asyncio timer per Trigger. If your DAG has any wait-until-timestamp logic, use DateTimeTrigger; the classic sensor is pure waste at any wait longer than a minute.

Worked example — a Trigger that watches a Postgres table

Detailed explanation. A common pattern in data platforms: wait for an upstream ETL to write a "batch marker" row to a Postgres control table. Rather than polling Postgres from a worker every 5 minutes, use an async Trigger with asyncpg and poll every 30 seconds. Show the implementation and the connection-pool consideration.

• Wait target. A row in etl_control.batch_ready matching a batch_id.
• Polling. Every 30 seconds via asyncpg (async Postgres driver).
• Connection. One asyncpg connection per Trigger, or a shared pool.

Question. Implement PostgresBatchTrigger with asyncpg. Discuss the connection-pool trade-off.

Input.

Parameter	Value
Postgres DSN	postgresql://airflow@db:5432/etl_control
Table	batch_ready
Key column	batch_id
Poll interval	30 s

Code.

import asyncio
import asyncpg
from airflow.triggers.base import BaseTrigger, TriggerEvent


class PostgresBatchTrigger(BaseTrigger):
    def __init__(self, dsn: str, batch_id: str, poll_interval: int = 30):
        super().__init__()
        self.dsn = dsn
        self.batch_id = batch_id
        self.poll_interval = poll_interval

    def serialize(self):
        return (
            "myco.triggers.pg.PostgresBatchTrigger",
            {
                "dsn": self.dsn,
                "batch_id": self.batch_id,
                "poll_interval": self.poll_interval,
            },
        )

    async def run(self):
        # Open one connection per Trigger — asyncpg's overhead is small
        conn = await asyncpg.connect(self.dsn)
        try:
            while True:
                row = await conn.fetchrow(
                    """
                    SELECT batch_id, row_count, completed_at
                    FROM   etl_control.batch_ready
                    WHERE  batch_id = $1
                    """,
                    self.batch_id,
                )
                if row is not None:
                    yield TriggerEvent({
                        "status": "success",
                        "batch_id": row["batch_id"],
                        "row_count": row["row_count"],
                        "completed_at": row["completed_at"].isoformat(),
                    })
                    return
                await asyncio.sleep(self.poll_interval)
        finally:
            await conn.close()

# Alternative — shared connection pool at Trigger module level
# For high-fan-out cases with hundreds of concurrent Postgres Triggers

_pool: asyncpg.Pool | None = None

async def _get_pool(dsn: str) -> asyncpg.Pool:
    global _pool
    if _pool is None:
        _pool = await asyncpg.create_pool(dsn, min_size=1, max_size=20)
    return _pool

class PostgresBatchTriggerPooled(BaseTrigger):
    # ...same __init__ and serialize...

    async def run(self):
        pool = await _get_pool(self.dsn)
        while True:
            async with pool.acquire() as conn:
                row = await conn.fetchrow(
                    "SELECT batch_id FROM etl_control.batch_ready WHERE batch_id = $1",
                    self.batch_id,
                )
                if row is not None:
                    yield TriggerEvent({"status": "success", "batch_id": row["batch_id"]})
                    return
            await asyncio.sleep(self.poll_interval)

Step-by-step explanation.

1. The simple version opens one asyncpg connection per Trigger. asyncpg is fast; opening a connection costs a few milliseconds. For a Triggerer hosting a few hundred concurrent Postgres Triggers, this is fine.
2. run() opens the connection in a try block, polls the target row every 30 seconds, and closes the connection in a finally block. finally ensures cleanup even on cancellation.
3. When the row appears, the Trigger yields the payload with the batch_id, row_count, and completed_at (converted to ISO string for JSON-safety). The Trigger's coroutine returns and asyncpg cleans up.
4. The pooled version at module level is useful when you have hundreds of concurrent Postgres Triggers — a shared pool of 20 connections handles all of them via async multiplexing. Postgres pool exhaustion happens only if all 20 connections are simultaneously mid-query.
5. The trade-off: per-Trigger connection is simpler and works for <100 concurrent Triggers; shared pool scales further but adds a module-level global. Pick per-Trigger unless benchmarks say otherwise.

Output.

Approach	Connections to Postgres	Trigger complexity	Scale ceiling
Per-Trigger connection	N (one per active Trigger)	simpler	~200 concurrent
Shared asyncpg pool	pool.max_size (e.g. 20)	needs module-global	1000+ concurrent

Rule of thumb. For fewer than 100 concurrent Postgres-polling Triggers, per-Trigger connections are simpler and safer. Switch to a shared asyncpg pool only when you have measured evidence of Postgres connection pressure or when Trigger fan-out crosses ~100 concurrent.

Senior interview question on custom Trigger design

A senior interviewer might ask: "Design a custom Trigger for a data platform that waits for an internal Kafka topic to produce a specific message. Walk me through the class design, the async polling, error handling, and the interface with the operator's execute_complete."

Solution Using aiokafka + backoff + payload envelope

import asyncio
import json
from aiokafka import AIOKafkaConsumer
from airflow.triggers.base import BaseTrigger, TriggerEvent
from airflow.sensors.base import BaseSensorOperator


class KafkaMessageTrigger(BaseTrigger):
    """Wait for a Kafka message whose payload matches `match_fn`."""

    def __init__(self, bootstrap_servers: str, topic: str,
                 group_id: str, batch_id: str):
        super().__init__()
        self.bootstrap_servers = bootstrap_servers
        self.topic = topic
        self.group_id = group_id
        self.batch_id = batch_id

    def serialize(self):
        return (
            "myco.triggers.kafka.KafkaMessageTrigger",
            {
                "bootstrap_servers": self.bootstrap_servers,
                "topic": self.topic,
                "group_id": self.group_id,
                "batch_id": self.batch_id,
            },
        )

    async def run(self):
        consumer = AIOKafkaConsumer(
            self.topic,
            bootstrap_servers=self.bootstrap_servers,
            group_id=self.group_id,
            auto_offset_reset="latest",
            enable_auto_commit=True,
        )
        await consumer.start()
        try:
            async for msg in consumer:
                try:
                    payload = json.loads(msg.value.decode("utf-8"))
                except (json.JSONDecodeError, UnicodeDecodeError):
                    continue
                # Match by batch_id in the payload
                if payload.get("batch_id") == self.batch_id:
                    yield TriggerEvent({
                        "status": "success",
                        "batch_id": self.batch_id,
                        "offset": msg.offset,
                        "partition": msg.partition,
                        "payload": payload,
                    })
                    return
        finally:
            await consumer.stop()


class KafkaMessageSensor(BaseSensorOperator):
    def __init__(self, bootstrap_servers: str, topic: str,
                 group_id: str, batch_id: str,
                 timeout: int = 6 * 3600, **kw):
        super().__init__(**kw)
        self.bootstrap_servers = bootstrap_servers
        self.topic = topic
        self.group_id = group_id
        self.batch_id = batch_id
        self.timeout = timeout

    def execute(self, context):
        self.defer(
            trigger=KafkaMessageTrigger(
                bootstrap_servers=self.bootstrap_servers,
                topic=self.topic,
                group_id=self.group_id,
                batch_id=self.batch_id,
            ),
            method_name="execute_complete",
            timeout=self.timeout,
        )

    def execute_complete(self, context, event):
        if event.get("status") != "success":
            raise RuntimeError(f"Kafka wait failed: {event}")
        return event["payload"]

Step-by-step trace.

Phase	Action	State
execute() on worker	Build KafkaMessageTrigger, .defer	trigger row inserted
Triggerer picks up	Deserialize, schedule run() on event loop	claim acquired
aiokafka consumer.start()	Connect to brokers, subscribe to topic	consumer live
async for msg	Await next Kafka batch	idle in event loop
Match found	Extract payload; yield TriggerEvent	event written
consumer.stop()	Clean shutdown in finally	resources freed
Worker execute_complete	Receive payload; return	task success

Under a burst of 500 concurrent KafkaMessageTriggers on one Triggerer, the aiokafka library multiplexes network I/O correctly; the async event loop dispatches events as they arrive. Per-Trigger cost is dominated by the network + parsing, not by Airflow overhead. The try/finally around consumer.stop() guarantees clean shutdown on task cancellation.

Output:

Concern	Answer
How does aiokafka fit the async model?	Native asyncio API; async for msg in consumer blocks the coroutine, not the loop
Where does the match logic live?	Inside run(), before the yield
What if the Kafka broker is down?	aiokafka reconnects with backoff; add outer try/except for hard failures
How is the offset preserved?	enable_auto_commit=True + consumer group id
What's the resume-side contract?	TriggerEvent payload → execute_complete → XCom

Why this works — concept by concept:

• async for over the consumer — aiokafka's async iterator hooks into the asyncio event loop, so the coroutine yields to the loop between messages. Perfect fit for the Trigger model.
• Match inside run() — the match logic runs on the Triggerer, not the worker. The worker only sees the matched payload via TriggerEvent. This is the whole point of the deferrable design.
• try/finally for cleanup — cancellation must not leak the Kafka consumer. The finally block runs on both normal exit and CancelledError.
• Consumer group id in serialize — the consumer group id is part of the Trigger's identity. If the Trigger restarts on a different Triggerer, the same group id ensures consumption resumes from the last committed offset.
• Cost — one aiokafka consumer per Trigger costs a socket + a few KB of buffer; the async loop handles I/O multiplexing. On a well-sized Triggerer, 500 concurrent Kafka Triggers is comfortable. Cost per Trigger is O(1) in wait time.

ETL
Topic — etl
ETL problems on event-driven / streaming waits

Practice →

Optimization Topic — optimization Optimization problems on async polling and backoff

Practice →

---

4. Triggerer service — deployment + HA

airflow triggerer is a first-class Airflow component alongside scheduler, webserver, and workers — HA via a shared trigger claim table, one CPU handles 100s to 1000s of concurrent Triggers

The mental model in one line: the Triggerer is a separate long-running Airflow process that runs airflow triggerer at the CLI, hosts an asyncio event loop capable of driving hundreds to thousands of concurrent Trigger run() coroutines, and achieves HA by having multiple instances share the trigger metadata table with a claim + heartbeat protocol. Every question about airflow triggerer deployment, asyncio airflow scale, and long-running task airflow operational patterns is a downstream consequence.

The Triggerer as a first-class component.

• Command. airflow triggerer starts the service. Same CLI as airflow scheduler, airflow webserver, airflow celery worker.
• Process model. Single-process, single asyncio event loop. Not multi-process. Multi-process scale is via multiple Triggerer replicas, not multiple loops per process.
• Config. [triggerer] section in airflow.cfg — default_capacity, job_heartbeat_sec, triggerer_health_check_threshold.
• Metadata DB reads. Every 5–10 seconds, the Triggerer polls the trigger table for unclaimed rows and claims some (up to default_capacity).
• Metadata DB writes. Heartbeats every job_heartbeat_sec seconds; TriggerEvent persistence on yield.

HA topology.

• Multiple instances. Two or more Triggerer pods running the same image, connected to the same metadata DB.
• Shared claim table. The trigger table has a triggerer_id column. Each Trigger belongs to whichever Triggerer wrote its id there.
• Heartbeat protocol. Each Triggerer writes its heartbeat every 5 seconds. If a Triggerer's heartbeat is older than triggerer_health_check_threshold (default 30 s), its claims are released.
• Failover time. ~30 seconds worst case — enough time for the surviving Triggerer to detect the missing heartbeat and re-claim.

Resource sizing.

• CPU. 1 CPU handles ~500–1000 concurrent Triggers for I/O-bound workloads (polling loops with asyncio.sleep). CPU-bound Triggers (JSON parsing large responses, computing hashes) reduce that ceiling.
• Memory. ~2 KB per Trigger for the async task frame + variable state per Trigger. 1000 Triggers ≈ 2 MB; a 512 MB pod handles 100k Triggers easily on this dimension.
• Network. Dominated by the async I/O to the polled services (S3, Postgres, Kafka). Sizing depends on the aggregate poll rate and the payload sizes.
• Metadata DB load. Each Triggerer polls the trigger table every ~5 seconds; heartbeats every 5 seconds. Two Triggerers add ~24 QPS to the metadata DB — negligible unless the DB is already saturated.

Deployment patterns.

• KubernetesExecutor + Helm. Set triggerer.replicas: 2 in the Airflow official chart. Add PodDisruptionBudget to prevent simultaneous restarts.
• Docker Compose. Add a triggerer service alongside scheduler/webserver/worker. Same image, different command.
• Systemd. airflow-triggerer.service unit file. Restart=always. Failure detection via the metadata DB heartbeat.
• Managed Airflow. MWAA / Composer / Astronomer provision the Triggerer for you; the replica count is usually a slider in the console.

Common interview probes.

• "What's the failure mode when a Triggerer crashes?" — heartbeat times out → claims released → surviving Triggerer picks up on next scheduling cycle → ~30 s failover.
• "Can I run one Triggerer or is HA required?" — one is fine for dev; two-plus for production. Single-Triggerer downtime pauses every deferrable task.
• "How many concurrent Triggers per CPU?" — 500–1000 for I/O-bound workloads.
• "How do I scale?" — replica count. Vertical CPU scale rarely helps (one asyncio loop = one core).

Worked example — Helm values for a 2-replica Triggerer

Detailed explanation. Walk through the Helm sub-values for a production Triggerer deployment on the Airflow official chart. Cover replica count, PodDisruptionBudget, resource requests/limits, capacity flag, and the graceful termination window.

• Workload. 300 concurrent Triggers at peak.
• Availability requirement. Zero downtime during rolling upgrades.
• Recovery window. <60 seconds for any single-pod failure.

Question. Produce the complete values.yaml sub-block for triggerer:.

Input.

Requirement	Value
Peak concurrent Triggers	300
HA replicas	2
CPU per replica	500m request, 2000m limit
Memory per replica	512Mi request, 2Gi limit
Graceful termination	60 s
PodDisruptionBudget	maxUnavailable=1

Code.

# values.yaml — Airflow official Helm chart, triggerer subvalues
triggerer:
  enabled: true
  replicas: 2

  # The capacity flag caps concurrent Triggers per instance
  args:
    - "triggerer"
  extraArgs: ["--capacity=1000"]

  # Resource sizing
  resources:
    requests:
      cpu: 500m
      memory: 512Mi
    limits:
      cpu: 2000m
      memory: 2Gi

  # Graceful termination — let claimed Triggers migrate before pod dies
  terminationGracePeriodSeconds: 60
  preStop:
    exec:
      command: ["/bin/sh", "-c", "sleep 30"]

  # Liveness / readiness probes
  livenessProbe:
    initialDelaySeconds: 30
    periodSeconds: 30
    timeoutSeconds: 10
    failureThreshold: 3

  readinessProbe:
    initialDelaySeconds: 10
    periodSeconds: 10
    timeoutSeconds: 5

  # PodDisruptionBudget — never lose both replicas at once
  podDisruptionBudget:
    enabled: true
    maxUnavailable: 1

  # Persistence — Triggerer is stateless (state in metadata DB)
  persistence:
    enabled: false

  # Airflow config overrides
  extraEnv:
    - name: AIRFLOW__TRIGGERER__DEFAULT_CAPACITY
      value: "1000"
    - name: AIRFLOW__TRIGGERER__JOB_HEARTBEAT_SEC
      value: "5"
    - name: AIRFLOW__TRIGGERER__TRIGGERER_HEALTH_CHECK_THRESHOLD
      value: "30"

# Companion: PodMonitor for Prometheus
apiVersion: monitoring.coreos.com/v1
kind: PodMonitor
metadata:
  name: airflow-triggerer
spec:
  selector:
    matchLabels:
      component: triggerer
  podMetricsEndpoints:
    - port: metrics
      interval: 15s

Step-by-step explanation.

1. replicas: 2 is the minimum for HA. Single-replica Triggerer means any pod restart pauses every deferrable task for the duration. Two replicas + PDB with maxUnavailable: 1 guarantees at least one Triggerer is always up during rolling upgrades.
2. --capacity=1000 caps the number of Triggers each Triggerer instance will claim. With 300 peak Triggers across 2 replicas, each holds ~150 in steady state and can absorb the other's load during a rolling upgrade (up to 300 on one replica).
3. Resource sizing gives 500m CPU baseline (for ~150 concurrent Triggers at ~0.3% each = 45%) and a 2000m limit for burst. Memory 512Mi baseline with 2Gi limit handles worst-case Trigger payloads without OOM.
4. terminationGracePeriodSeconds: 60 + a 30-second preStop sleep gives Kubernetes 60 seconds to send SIGTERM. The Triggerer's graceful-shutdown handler releases its claims within a few seconds; the extra sleep buys headroom for large fanouts.
5. The Prometheus PodMonitor scrapes Triggerer metrics (added in Airflow 2.7+) — active Trigger count, heartbeat age, DB query latency. Wire these into alerts for "Triggerer heartbeat > 30s" and "capacity utilisation > 80%".

Output.

Sizing dimension	Value	Reasoning
Replicas	2	HA minimum
CPU request	500m	Per-Trigger cost × steady-state concurrency
CPU limit	2000m	Burst headroom
Memory request	512Mi	2 KB × 300 Triggers × 4× safety
Memory limit	2Gi	Payload burst
Capacity	1000	2× peak per replica
PDB maxUnavailable	1	Never lose both replicas
Grace period	60 s	Claim migration + preStop sleep

Rule of thumb. Two-replica Triggerer with PDB maxUnavailable: 1 is the production baseline. Scale replicas only when a single instance's CPU regularly exceeds 70% at peak. Never run a single Triggerer in production — the deployment has no availability story.

Worked example — the heartbeat + claim protocol

Detailed explanation. Understanding exactly how the claim + heartbeat protocol works clarifies the failure model. Every Trigger row in the metadata DB has a triggerer_id column that names the current owner. Owner writes heartbeats to triggerer_job. If the heartbeat is stale, the triggerer_id is nullified by a live Triggerer, which then re-claims.

• Claim SQL. UPDATE trigger SET triggerer_id = :me WHERE id IN (SELECT id FROM trigger WHERE triggerer_id IS NULL LIMIT :capacity) RETURNING id;
• Heartbeat SQL. UPDATE triggerer_job SET latest_heartbeat = now() WHERE id = :me;
• Cleanup SQL. UPDATE trigger SET triggerer_id = NULL WHERE triggerer_id IN (SELECT id FROM triggerer_job WHERE latest_heartbeat < now() - '30 seconds');

Question. Trace through a Triggerer crash scenario: Triggerer 1 has 200 claims, Triggerer 2 has 100 claims. Triggerer 1 OOM-killed. Show exactly what happens over the next 60 seconds.

Input.

Time	Event
t=0	Triggerer 1 OOM-killed
t=0..30	Triggerer 1's last_heartbeat ages past 30s
t=30	Triggerer 2's cleanup query nullifies Triggerer 1's claims
t=30..40	Triggerer 2's claim query picks up the freed Triggers
t=40..end	All 300 Triggers run on Triggerer 2

Code.

-- Airflow triggerer job source, pseudo-SQL of the coordination protocol

-- 1. Startup: each Triggerer inserts into triggerer_job
INSERT INTO triggerer_job (hostname, start_date, latest_heartbeat, state)
VALUES (:hostname, now(), now(), 'running')
RETURNING id;

-- 2. Every job_heartbeat_sec seconds:
UPDATE triggerer_job
SET    latest_heartbeat = now()
WHERE  id = :my_triggerer_id;

-- 3. Every scheduling cycle, cleanup stale claims:
UPDATE trigger
SET    triggerer_id = NULL
WHERE  triggerer_id IN (
    SELECT id FROM triggerer_job
    WHERE  latest_heartbeat < now() - interval '30 seconds'
      AND  state = 'running'
);

-- 4. Then, claim unclaimed triggers (up to capacity):
UPDATE trigger
SET    triggerer_id = :my_triggerer_id
WHERE  id IN (
    SELECT id
    FROM   trigger
    WHERE  triggerer_id IS NULL
    ORDER  BY id
    LIMIT  :remaining_capacity
    FOR    UPDATE SKIP LOCKED
)
RETURNING id;

# Rough Python translation of what the Triggerer job runner does
import asyncio
from datetime import datetime, timezone

class TriggererJobRunner:
    def __init__(self, capacity: int = 1000):
        self.capacity = capacity
        self.active_triggers = {}   # id -> asyncio.Task

    async def loop(self):
        while True:
            # 1. Heartbeat
            await self._write_heartbeat()
            # 2. Cleanup other Triggerers' stale claims
            await self._cleanup_stale_claims()
            # 3. Claim new Triggers up to remaining capacity
            free = self.capacity - len(self.active_triggers)
            if free > 0:
                new_ids = await self._claim_triggers(limit=free)
                for tid in new_ids:
                    trigger = await self._load_and_construct(tid)
                    self.active_triggers[tid] = asyncio.create_task(
                        self._run_trigger(tid, trigger)
                    )
            # 4. Sleep briefly
            await asyncio.sleep(5)

Step-by-step explanation.

1. At t=0, Triggerer 1 is OOM-killed by Kubernetes. Its process dies; no cleanup runs. Its last recorded heartbeat is from ~t=-5 (just before death).
2. From t=0 to t=30, Triggerer 2 continues its normal 5-second scheduling loop. On each cycle it runs the cleanup query, but latest_heartbeat < now() - interval '30 seconds' doesn't match Triggerer 1 yet.
3. At around t=30, Triggerer 1's heartbeat age crosses the 30-second threshold. Triggerer 2's cleanup query updates the 200 rows previously owned by Triggerer 1 — SET triggerer_id = NULL.
4. On Triggerer 2's next iteration (~5 seconds later), the claim query finds 200 unclaimed rows. Triggerer 2 has capacity headroom (say, 400 free of 1000). It claims all 200.
5. For each claimed Trigger, Triggerer 2 deserialises the Trigger class from the row's classpath + kwargs and starts a new asyncio task. The 200 Triggers resume execution on Triggerer 2, ~35 seconds after the Triggerer 1 crash. From the DAG's perspective, the wait was extended by ~35 seconds — invisible for hour-scale waits.

Output.

Time	Triggerer 1 state	Triggerer 2 state	Trigger table state
t=-1	200 active	100 active	200 owned by T1, 100 by T2
t=0	crashed	100 active	(unchanged)
t=15	dead	100 active	(unchanged)
t=32	dead	detects stale T1	200 nulled
t=37	dead	claims 200	300 owned by T2
t=40+	dead	300 active	(steady state)

Rule of thumb. The Triggerer's HA story lives entirely in the metadata DB — the claim table + the heartbeat timeout. Failover is 30–40 seconds worst case, which is invisible for any deferrable wait longer than a minute. Multi-Triggerer is the production baseline.

Worked example — Prometheus alerts on Triggerer health

Detailed explanation. In production, a broken Triggerer is a silent outage — deferrable tasks silently stop resuming. Wire Prometheus alerts on Triggerer heartbeat age and capacity utilisation to catch the problem before it turns into an SLA breach.

• Metric 1 — heartbeat age. airflow_triggerer_heartbeat_age_seconds — how long since the Triggerer last heartbeat. Alert if > 60 s.
• Metric 2 — capacity utilisation. airflow_triggerer_capacity_used / airflow_triggerer_capacity_total. Alert if > 0.85.
• Metric 3 — event lag. airflow_trigger_event_lag_seconds — time from Trigger event fired to task re-queued. Alert if > 60 s.

Question. Write the PromQL rules for all three alerts and produce a runbook.

Input.

Alert	Threshold	For
Triggerer heartbeat age	> 60 s	1 m
Capacity utilisation	> 0.85	5 m
Event lag	> 60 s	2 m

Code.

# prometheus alert rules
groups:
  - name: airflow-triggerer
    rules:
      - alert: TriggererHeartbeatStale
        expr: |
          time() - airflow_triggerer_heartbeat_timestamp_seconds > 60
        for: 1m
        labels:
          severity: page
        annotations:
          summary: "Triggerer heartbeat stale on {{ $labels.pod }}"
          runbook: "https://runbook.internal/airflow/triggerer-heartbeat"

      - alert: TriggererCapacityHigh
        expr: |
          airflow_triggerer_capacity_used / airflow_triggerer_capacity_total > 0.85
        for: 5m
        labels:
          severity: warn
        annotations:
          summary: "Triggerer {{ $labels.pod }} at {{ $value }}% capacity"

      - alert: TriggerEventLag
        expr: |
          histogram_quantile(0.95,
            rate(airflow_trigger_event_lag_seconds_bucket[5m])
          ) > 60
        for: 2m
        labels:
          severity: page
        annotations:
          summary: "Trigger event lag p95 > 60s"

Runbook — TriggererHeartbeatStale
=================================

1. Check pod status
   kubectl get pods -l component=triggerer -n airflow
   -> if 0/2 ready: proceed to escalation
   -> if 1/2 ready: HA is working; investigate the dead pod

2. Inspect logs of the dead pod
   kubectl logs -l component=triggerer -n airflow --previous --tail=100
   -> look for OOMKilled, event loop stalls, DB connection errors

3. Verify metadata DB health
   -> Postgres CPU / connections / disk
   -> if DB is saturated, that's the root cause; unblock DB first

4. If pod is repeatedly OOM-killed
   -> increase memory limit in Helm values
   -> re-check for a leaky Trigger implementation

5. If pod is repeatedly stalled
   -> check for blocking I/O in Trigger.run() (time.sleep, sync requests)
   -> disable the offending DAG; deploy a fix

Step-by-step explanation.

1. Alert 1 is the "am I even alive?" alert. If the Triggerer heartbeat hasn't ticked in 60 s, either the process is dead or the DB is unreachable. Page immediately — this is the failure mode that silently stalls every deferrable task.
2. Alert 2 is the capacity warning. At >85% of default_capacity, the Triggerer is at risk of refusing new Triggers, which manifests as unclaimed rows in the trigger table and increasing event lag. Bump the capacity or add a replica.
3. Alert 3 is the "downstream user impact" alert. Event lag measures time from yield TriggerEvent in the Trigger to the task being re-queued for execution. High lag means the Triggerer is bottlenecked either on CPU or on metadata DB writes.
4. The runbook walks the on-call from "am I paged" to root cause in five steps — pod status, logs, DB, memory tuning, Trigger implementation review. Every deferrable-heavy platform should have this runbook checked in.
5. The three alerts together give layered coverage — availability, capacity, latency. Missing any one leaves a class of failure invisible.

Output.

Alert	Detects	Typical remediation
Heartbeat stale	Triggerer dead	Restart pod; fix crash cause
Capacity high	Triggerer at limit	Raise capacity or add replica
Event lag high	Slow throughput	Investigate DB / CPU bottleneck

Rule of thumb. All three alerts must be wired before the first deferrable DAG ships. A Triggerer failure is the most disruptive silent outage in a deferrable-heavy platform — every downstream task pauses invisibly. Alerting is not optional.

Senior interview question on Triggerer deployment

A senior interviewer might ask: "You're building a self-managed Airflow deployment on Kubernetes with 500 deferrable tasks per day. Walk me through the Triggerer deployment — replica count, sizing, HA topology, alerts, and the failure modes you'd defend against."

Solution Using 2-replica Triggerer with PDB + observability

# The full Helm values block
triggerer:
  enabled: true
  replicas: 2                         # HA baseline
  extraArgs: ["--capacity=1000"]      # 2× peak per replica

  resources:
    requests: {cpu: 500m, memory: 512Mi}
    limits:   {cpu: 2000m, memory: 2Gi}

  terminationGracePeriodSeconds: 60

  podDisruptionBudget:
    enabled: true
    maxUnavailable: 1

  livenessProbe:
    initialDelaySeconds: 30
    periodSeconds: 30

  readinessProbe:
    initialDelaySeconds: 10
    periodSeconds: 10

  extraEnv:
    - name: AIRFLOW__TRIGGERER__DEFAULT_CAPACITY
      value: "1000"
    - name: AIRFLOW__TRIGGERER__JOB_HEARTBEAT_SEC
      value: "5"
    - name: AIRFLOW__TRIGGERER__TRIGGERER_HEALTH_CHECK_THRESHOLD
      value: "30"

# Prometheus alerts (see above)
# - TriggererHeartbeatStale (page)
# - TriggererCapacityHigh (warn)
# - TriggerEventLag (page)

Step-by-step trace.

Concern	Design choice	Reasoning
HA	2 replicas + PDB maxUnavailable=1	never lose both; ~30 s failover
CPU sizing	500m request / 2000m limit	500 Triggers × ~0.3% per Trigger = ~150% total, split across 2 replicas
Memory sizing	512Mi / 2Gi	2 KB × 500 = 1 MB + 4× headroom
Capacity	1000 per replica	2× peak; free ceiling
Grace period	60 s	claim migration + preStop sleep
Alerts	3 (heartbeat, capacity, lag)	availability + capacity + latency
Metadata DB	shared with scheduler	add ~30 QPS overhead; negligible

After the rollout, the Triggerer runs 2 replicas holding ~250 concurrent Triggers each in steady state. A single pod failure triggers a ~30-second failover invisible to hour-scale waits. Prometheus alerts surface any degradation within 1–5 minutes. The DAG author sees deferrable=True as a working feature with zero operational surprises.

Output:

Concern	Value
Peak concurrent Triggers	500
Replicas	2
Per-replica load	~250 Triggers
Per-replica CPU at peak	~75% (of 500m request)
Failover time	~30 seconds
Alerts wired	heartbeat, capacity, event lag
Monthly cost	~$40 (2 × 1 vCPU × $0.05/hr × 30 d × 24 h / 2)

Why this works — concept by concept:

• Two replicas as HA minimum — one replica has no availability story; three is overkill for I/O-bound workloads at this scale. Two is the sweet spot for cost + reliability.
• PodDisruptionBudget as an invariant — Kubernetes will happily restart both replicas simultaneously during a node drain unless PDB stops it. maxUnavailable=1 is non-negotiable.
• Capacity headroom — sizing capacity at 2× peak means either replica can absorb the other's load during a rolling upgrade. Cheap insurance; no runtime cost.
• Prometheus + runbook pairing — the alert without the runbook is noise; the runbook without the alert is a static document. Ship them together or neither.
• Cost — 2 × 1 vCPU × $0.05/hr = $2.40/day = ~$72/month. Compared to $3800/month in avoided worker-idle cost from Section 1, the Triggerer is a rounding error. The ROI on the infrastructure line is >50×.

ETL
Topic — etl
ETL problems on orchestrator deployment and HA

Practice →

Optimization Topic — optimization Optimization problems on Triggerer sizing and capacity planning

Practice →

---

5. Migration + operator inventory

deferrable=True on modern operators — the codemod path from S3KeySensor to S3KeySensorAsync, the Provider inventory for 2026, and the cost-audit query that quantifies the savings

The mental model in one line: modern Airflow Providers ship a deferrable: bool = False kwarg on almost every wait-heavy operator, and migration is usually a one-flag codemod that flips the operator through the Trigger path, with the full inventory covering S3, GCS, HTTP, Snowflake, Databricks, and BigQuery in 2026. Every question about deferrable sensor migration, operator inventory, and the cost-audit story is a downstream consequence.

The 2026 operator inventory.

• Amazon Provider. S3KeySensor(deferrable=True), S3KeysUnchangedSensor(deferrable=True), AthenaOperator(deferrable=True), EmrJobFlowSensor(deferrable=True), RedshiftDataOperator(deferrable=True), EcsRunTaskOperator(deferrable=True). Fully deferrable.
• Google Provider. GCSObjectExistenceSensor(deferrable=True), BigQueryInsertJobOperator(deferrable=True), DataprocSubmitJobOperator(deferrable=True), DataflowJobStatusSensor(deferrable=True). Fully deferrable.
• HTTP Provider. HttpSensor(deferrable=True), HttpOperator(deferrable=True). Fully deferrable.
• Snowflake Provider. SnowflakeOperator(deferrable=True), SnowflakeSqlApiOperator(deferrable=True). Fully deferrable.
• Databricks Provider. DatabricksSubmitRunOperator(deferrable=True), DatabricksRunNowOperator(deferrable=True), DatabricksSqlOperator(deferrable=True). Fully deferrable.
• Core. TimeDeltaSensor(deferrable=True), DateTimeSensor(deferrable=True), ExternalTaskSensor(deferrable=True). Fully deferrable.
• Global default. [operators] default_deferrable = True in airflow.cfg flips every operator's default to deferrable=True without per-operator changes.

The three migration paths.

• Flag flip. MyOperator(..., deferrable=True). Zero code change beyond the kwarg. The vast majority of operators support this.
• Class rename. MyOperatorAsync for older Provider versions that ship separate async classes. Prefer flag-flip when both exist; the flag is the future.
• Custom Trigger. For operators without a deferrable version yet, write a custom BaseTrigger (see Section 3) and either subclass or replace the operator.

The global default.

• Config. [operators] default_deferrable = True in airflow.cfg (or AIRFLOW__OPERATORS__DEFAULT_DEFERRABLE=True env var).
• Effect. Every operator whose __init__ reads deferrable from conf.getboolean('operators', 'default_deferrable', fallback=False) defaults to True.
• Caveat. Not every operator wires this — check per-provider release notes. The safe default is to set the flag explicitly per operator.

The cost-audit query.

• Purpose. Quantify slot-hours consumed by classic-mode operators before migration.
• Source. Airflow metadata DB task_instance table.
• Output. Ranked list of DAG/task pairs by total slot-hours.
• Follow-up. After migration, re-run to compare.

Common interview probes on migration.

• "Walk me through migrating an S3KeySensor to deferrable." — one-flag flip.
• "Which built-in operators support deferrable in 2026?" — inventory above.
• "What if my custom operator doesn't have deferrable?" — write a custom Trigger + operator.
• "How do you quantify the savings?" — metadata-DB audit query on task_instance.duration.

Worked example — one-flag migration from S3KeySensor to deferrable

Detailed explanation. The canonical migration case. A DAG uses S3KeySensor with mode="reschedule". Flipping deferrable=True moves the wait to the Trigger path with zero downstream code changes. Show the diff.

• Before. S3KeySensor(mode="reschedule", poke_interval=300, timeout=6*3600).
• After. S3KeySensor(deferrable=True, poke_interval=60, timeout=6*3600).
• Downstream tasks. No changes required.

Question. Show the full DAG before and after, and walk through the semantic differences.

Input.

Field	Before	After
mode	reschedule	(dropped)
deferrable	(default False)	True
poke_interval	300 s	60 s (cheap now)
timeout	6 * 3600 s	6 * 3600 s
Slot cost per wait	~50 s warmup × 72 pokes	~1 s execute + 1 s execute_complete

Code.

# Before — classic reschedule-mode sensor
from airflow.decorators import dag, task
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from airflow.operators.python import PythonOperator
from datetime import datetime

@dag(
    dag_id="daily_export_v1",
    start_date=datetime(2026, 1, 1),
    schedule="0 6 * * *",
    catchup=False,
)
def daily_export():
    wait_for_export = S3KeySensor(
        task_id="wait_for_export",
        bucket_key="s3://exports/dt={{ ds }}/data.parquet",
        aws_conn_id="aws_default",
        mode="reschedule",       # frees worker between pokes, but pays warmup
        poke_interval=300,
        timeout=6 * 3600,
    )

    @task
    def load_to_warehouse():
        # ... regular ETL ...
        pass

    wait_for_export >> load_to_warehouse()

daily_export()

# After — one flag change; downstream code unchanged
from airflow.decorators import dag, task
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from datetime import datetime

@dag(
    dag_id="daily_export_v2",
    start_date=datetime(2026, 1, 1),
    schedule="0 6 * * *",
    catchup=False,
)
def daily_export():
    wait_for_export = S3KeySensor(
        task_id="wait_for_export",
        bucket_key="s3://exports/dt={{ ds }}/data.parquet",
        aws_conn_id="aws_default",
        deferrable=True,         # THE change
        poke_interval=60,        # cheap now; runs in the async loop
        timeout=6 * 3600,
    )

    @task
    def load_to_warehouse():
        # ... regular ETL ...
        pass

    wait_for_export >> load_to_warehouse()

daily_export()

Step-by-step explanation.

1. The mode="reschedule" kwarg is dropped — it's redundant when deferrable=True. The operator routes through the Trigger path regardless.
2. deferrable=True flips the operator's execute() to build a Trigger and call .defer(). The classic poke loop is replaced by the Trigger's async run().
3. poke_interval is tightened from 300 to 60 seconds. Because polling is now cheap (no warmup, no DAG re-parse), a tighter cadence means faster response to the file arrival without cost implications.
4. The downstream load_to_warehouse task doesn't change. It waits on the sensor's success signal just as before; the sensor's internal implementation is invisible to it.
5. Deployment: merge the flag flip, wait for one DAG run, verify the sensor task transitions running → deferred → running → success (visible in the Grid view). Look for the deferred state and the shorter total worker slot time.

Output.

DAG version	Sensor task states	Worker slot-hours	Cost per run
v1 (reschedule)	queued → running (poke) × 72 → success	72 × 0.083 min = ~1 slot-hour × 6 hrs of waits	~$0.48
v2 (deferrable)	queued → running → deferred → running → success	~2 seconds	~$0.0001

Rule of thumb. For the modern Amazon Provider, S3KeySensor(deferrable=True) is the drop-in replacement for every reschedule-mode use. No downstream changes; no operator swap. Ship the codemod as one PR per DAG batch.

Worked example — the deferrable BigQueryInsertJobOperator

Detailed explanation. Not just sensors — expensive-query operators also benefit from deferrable. A BigQueryInsertJobOperator running a 90-minute analytical query classically holds the worker slot for those 90 minutes. Deferrable moves the wait to the Trigger, releases the worker after submitting the job, and re-hydrates only when the query finishes.

• Query cost. 90-minute BigQuery job.
• Classic operator cost. 90 minutes × 1 slot × $0.08 = $0.12 per run.
• Deferrable operator cost. ~1 second execute + 1 second execute_complete = negligible.

Question. Show the deferrable BigQuery operator config and the Trigger's role.

Input.

Parameter	Value
Query	SELECT ... FROM tenant_events WHERE dt=... GROUP BY ...
Estimated duration	90 minutes
Concurrent jobs at peak	20

Code.

from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator

# Deferrable BigQuery operator — submits job then defers
run_daily_agg = BigQueryInsertJobOperator(
    task_id="run_daily_agg",
    configuration={
        "query": {
            "query": "SELECT * FROM `myproj.analytics.aggregation_v1`",
            "useLegacySql": False,
            "destinationTable": {
                "projectId": "myproj",
                "datasetId": "analytics",
                "tableId": "daily_agg_{{ ds_nodash }}",
            },
            "writeDisposition": "WRITE_TRUNCATE",
        },
        "jobTimeoutMs": "10800000",   # 3 hours max
    },
    location="US",
    gcp_conn_id="google_default",
    deferrable=True,                  # THE flag
)

# Internal — what happens on execute()
#   1. Submit the BigQuery job via the BQ REST API
#   2. Get back a job_id
#   3. Build a BigQueryInsertJobTrigger(job_id=..., project=..., location=...)
#   4. self.defer(trigger=trigger, method_name="execute_complete")
#
# Inside the Trigger's run():
#   while True:
#       state = await bq.jobs.get(job_id).state
#       if state in ("DONE",):
#           yield TriggerEvent({"status": "success", "job_id": job_id})
#           return
#       elif state in ("FAILED", "ERROR"):
#           yield TriggerEvent({"status": "error", "job_id": job_id, "errors": ...})
#           return
#       await asyncio.sleep(30)

Step-by-step explanation.

1. execute() uses the sync BQ client to submit the job (a brief REST call), then constructs a BigQueryInsertJobTrigger with the returned job_id and calls .defer().
2. The worker slot is released within a couple of seconds of the DAG task starting. The BQ job runs in Google's infrastructure — Airflow's worker doesn't need to be involved.
3. The Trigger's async run() polls bq.jobs.get(job_id) every 30 seconds. Each poll is a REST call taking ~50 ms; hundreds of concurrent BQ Triggers run comfortably on one Triggerer.
4. When the job's state transitions to DONE, the Trigger yields TriggerEvent with the job_id and any result metadata (row count, bytes processed). The task re-hydrates on a worker for execute_complete.
5. Cost: 20 concurrent 90-minute BQ jobs × 90 min × $0.08/hr / 60 = $2.40/hr classic. Deferrable brings that to ~$0. Multiply by the number of daily BQ jobs and the savings are substantial.

Output.

Metric	Classic	Deferrable
Slot-hours per job	1.5 (90 min)	0.0006 (2 sec)
20 concurrent, cost/hr	$2.40	$0.001
20 concurrent, daily cost	$57.60	$0.02
Monthly cost	$1728	$0.60

Rule of thumb. Long-running query operators (BigQuery, Snowflake, Redshift, Athena) are just as important as sensors for the deferrable migration. Any operator with a wait longer than 60 seconds is a candidate.

Worked example — the cost-audit query and monthly savings report

Detailed explanation. Before and after migration, run a metadata-DB query to quantify the savings. Ship it as a monthly dashboard so the FinOps + platform teams see the ongoing win. Walk through the query and the interpretation.

• Table. task_instance.
• Metric. EXTRACT(EPOCH FROM (end_date - start_date)) = wall-clock duration held; duration = actual worker-executed seconds.
• Ratio. wall/duration is the "held-idle" ratio for classic sensors (high) vs deferrable (~1).

Question. Produce the SQL query, the interpretation, and the monthly savings report format.

Input.

Field	Value
Airflow metadata DB	Postgres
Look-back window	30 days
Sensor / operator types	S3KeySensor, HttpSensor, BigQueryInsertJobOperator, DatabricksSubmitRunOperator

Code.

-- Cost audit query — held-idle vs actual worker seconds
WITH sensor_runs AS (
  SELECT
      dag_id,
      task_id,
      operator,
      execution_date::date                                                    AS run_date,
      EXTRACT(EPOCH FROM (end_date - start_date))                             AS wall_seconds,
      duration                                                                AS worker_seconds,
      state
  FROM   task_instance
  WHERE  execution_date >= now() - interval '30 days'
    AND  state = 'success'
    AND  (operator LIKE '%Sensor%'
       OR operator IN ('BigQueryInsertJobOperator',
                       'SnowflakeOperator',
                       'DatabricksSubmitRunOperator',
                       'HttpOperator'))
)
SELECT
    dag_id,
    task_id,
    operator,
    COUNT(*)                                                    AS runs,
    ROUND(SUM(wall_seconds)   / 3600.0, 2)                       AS total_wall_hours,
    ROUND(SUM(worker_seconds) / 3600.0, 2)                       AS total_worker_hours,
    ROUND(SUM(wall_seconds - COALESCE(worker_seconds, wall_seconds)) / 3600.0, 2)
                                                                 AS held_idle_hours,
    ROUND(0.08 * SUM(wall_seconds - COALESCE(worker_seconds, wall_seconds)) / 3600.0, 2)
                                                                 AS held_idle_cost_usd
FROM   sensor_runs
GROUP  BY dag_id, task_id, operator
ORDER  BY held_idle_cost_usd DESC
LIMIT  50;

Interpretation — held_idle_hours is the target metric
======================================================

Classic sensor (mode=poke or mode=reschedule):
  wall_seconds   ≈ full wait time (e.g. 21600 s = 6 hours)
  worker_seconds ≈ same (poke) or ~poke_count × warmup (reschedule)
  held_idle       ≈ wall_seconds - worker_seconds ≈ 5+ hours

Deferrable sensor:
  wall_seconds   ≈ full wait time (unchanged)
  worker_seconds ≈ 2 seconds (execute + execute_complete)
  held_idle       ≈ ~0 (the wait was on the Triggerer, not billed as worker)

The savings signal: post-migration, worker_seconds for the same DAG task
plummets from ~wall_seconds to ~2. Multiply by task_instance count and
worker_slot_cost per second for the FinOps report.

# Optional — automated monthly report as a DAG
from airflow.decorators import dag, task
from airflow.providers.postgres.hooks.postgres import PostgresHook
from datetime import datetime

@dag(
    dag_id="airflow_cost_audit",
    schedule="@monthly",
    start_date=datetime(2026, 1, 1),
    catchup=False,
)
def cost_audit():
    @task
    def run_report():
        pg = PostgresHook(postgres_conn_id="airflow_meta")
        rows = pg.get_records("""
            /* the query above */
        """)
        # push to Slack, email, or a metrics table
        for r in rows[:20]:
            print(r)

    run_report()

cost_audit()

Step-by-step explanation.

1. The CTE sensor_runs filters task_instance for the last 30 days of successful sensor + long-running operator runs. wall_seconds is derived from end_date - start_date; worker_seconds is the duration column that Airflow writes on completion.
2. The outer query aggregates by (dag_id, task_id, operator). held_idle_hours = wall_seconds - worker_seconds — the time held that wasn't productive work.
3. The cost column multiplies held-idle hours by an assumed $0.08 per slot-hour. Adjust for your Kubernetes / MWAA / Astronomer billing rate.
4. Sorting by held_idle_cost_usd DESC surfaces the top 50 (dag, task) pairs to migrate. This becomes the migration backlog for the platform team.
5. The optional monthly DAG runs the query on a schedule and pushes the report to Slack / email. Ongoing visibility keeps the deferrable win in front of leadership; the report also catches regression when a new DAG author accidentally uses classic mode.

Output.

dag_id	task_id	operator	runs	wall_hours	worker_hours	held_idle_hours	cost_usd
daily_export	wait_for_export	S3KeySensor	30	120	120	0 (already migrated)	$0
hourly_ingest	poll_api	HttpSensor	720	480	480	0	$0
legacy_pipeline	wait_for_batch	S3KeySensor	30	180	180	0	$14.40 (pre-mig)

Rule of thumb. Run the cost-audit query before starting migration to build the backlog, and monthly after migration to catch regressions. The dashboard becomes the canonical FinOps artefact for the Airflow platform.

Senior interview question on migration strategy

A senior interviewer might ask: "You inherit an Airflow deployment with 400 sensors per day, all classic reschedule mode. Walk me through the migration plan — how you'd audit, prioritise, execute, monitor, and defend against regressions."

Solution Using a four-week migration plan + audit dashboard

Migration plan — classic → deferrable in 4 weeks
================================================

Week 1 — Audit + prep
  - Run the cost-audit query → rank top 50 offenders
  - Verify Airflow version supports deferrable (2.4+; ideally 2.7+)
  - Deploy Triggerer service (2 replicas, HA)
  - Wire Prometheus alerts on Triggerer heartbeat / capacity / event lag
  - Verify Provider versions support deferrable on the target operators

Week 2 — Batch 1 (safest 25% by risk)
  - Migrate top 10 highest-cost DAGs (usually simple S3KeySensor cases)
  - One PR per DAG batch; deferrable=True flag flip
  - Deploy in shadow mode: run new DAG alongside old for 1 week
  - Compare wall-clock, worker-hours, cost in the audit dashboard

Week 3 — Batch 2 (next 50% by risk)
  - Migrate BigQuery / Snowflake / Databricks long-runners
  - HTTP sensors with custom retry logic → verify backoff still fires
  - Continue shadow mode

Week 4 — Batch 3 (custom + edge cases)
  - Custom sensor / operator code — write custom Trigger classes
  - LISTEN-like session-scope waits — carefully validate that the deferrable path preserves semantics
  - Final rollout; delete old DAG versions

Ongoing — Regression defence
  - Cost-audit dashboard monthly
  - Global default: [operators] default_deferrable = True
  - PR template requirement: "does this add a classic-mode sensor?"
  - Airflow lint rule: warn on Sensor without deferrable

Step-by-step trace.

Week	Focus	Coverage	Cumulative savings
1	Infra + audit	0%	$0 (baseline)
2	Top 10 DAGs	25% of volume	$1000/mo
3	Long-runners	75% of volume	$2500/mo
4	Custom + edge	100%	$3800/mo
Ongoing	Regressions	maintain	preserved

After week 4, the platform runs entirely on deferrable operators. The Triggerer runs stable; the cost-audit dashboard shows near-zero held-idle-hours across the top 50 DAGs. The monthly savings capture the delta between the pre-migration and post-migration cost columns. Regression defence (global default + lint rule + PR template) prevents backsliding.

Output:

Milestone	Cost	Savings	Risk
Week 0 baseline	$3840/mo	—	high (no HA)
Week 1 (Triggerer up)	$3840/mo	$0	medium (unblocked)
Week 2 (25% migrated)	$2900/mo	$940/mo	medium
Week 3 (75% migrated)	$1400/mo	$2440/mo	low
Week 4 (100% migrated)	$55/mo	$3785/mo	low (ongoing defence)

Why this works — concept by concept:

• Audit before action — the cost-audit query converts "we should probably migrate" into a ranked backlog with dollar values. Prioritisation is data-driven.
• Shadow mode during batching — running the new DAG alongside the old for one week catches semantic regressions (missing state, wrong TriggerEvent payload) before production reliance.
• Global default + lint — makes the deferrable path the pit of success. Any new DAG defaults to deferrable; classic mode requires explicit opt-in with review.
• Cost-audit dashboard as a permanent artefact — the monthly report captures the ongoing win and surfaces regressions early. It's the canonical FinOps view for the Airflow platform.
• Cost — 4 weeks × 1 senior engineer = ~$16k in migration labor; monthly savings $3785 = 4-month payback. After year 1 the deferrable migration has paid for itself 3× and continues to save every month.

ETL
Topic — etl
ETL problems on pipeline migration and cost optimisation

Practice →

Optimization
Topic — optimization
Optimization problems on slot-hour reduction and audit queries

Practice →

---

Cheat sheet — deferrable recipes

• Deferrable operator flag toggle. deferrable=True on modern operators (S3KeySensor, HttpSensor, BigQueryInsertJobOperator, SnowflakeOperator, DatabricksSubmitRunOperator, etc). Set globally via [operators] default_deferrable = True in airflow.cfg. Prefer explicit per-operator flags for clarity; use the global default as backstop.
• Custom BaseTrigger skeleton. class MyTrigger(BaseTrigger) with __init__(**kwargs) capturing state, serialize() returning (classpath, kwargs) for reconstruction, and async def run() yielding TriggerEvent({...}) when the awaited condition fires. Never call blocking I/O in run() — only await asyncio.sleep(...), await aiohttp.get(...), await asyncpg.fetch(...), etc.
• Custom operator skeleton. In execute(context), build the Trigger and call self.defer(trigger=..., method_name="execute_complete", timeout=...). In execute_complete(context, event), receive the TriggerEvent payload dict, verify event["status"] == "success", extract state, return XCom values. Never wrap .defer() in a generic try/except Exception.
• Triggerer Helm-values snippet. triggerer.replicas: 2 with extraArgs: ["--capacity=1000"], resources.requests.cpu: 500m / limits.cpu: 2000m, terminationGracePeriodSeconds: 60, and podDisruptionBudget.enabled: true with maxUnavailable: 1. Two replicas is the HA baseline; scale up only when a single instance's CPU stays above 70% at peak.
• Cost audit query. Query the metadata DB task_instance table over the last 30 days; compute wall_seconds = EXTRACT(EPOCH FROM (end_date - start_date)) vs worker_seconds = duration; the delta is held_idle_hours. Multiply by your slot-hour cost. Sort DESC to build the migration backlog. Re-run monthly to catch regressions.
• Poke-to-deferrable migration recipe. (1) Verify Airflow ≥ 2.4 and Provider version supports deferrable kwarg; (2) deploy Triggerer service (2 replicas, HA) before migrating any DAG; (3) flip deferrable=True on the operator, drop mode="reschedule", tighten poke_interval since it's now cheap; (4) shadow-mode the DAG for one week; (5) monitor the cost-audit dashboard.
• TriggerEvent payload contract. Dict of primitives only (str, int, float, bool, list, dict of same). No datetime objects (convert to ISO string), no boto3 responses (extract fields), no non-JSON-serialisable state. The payload is persisted to the metadata DB as JSON and delivered to execute_complete unchanged.
• .defer() mechanics. Raises TaskDeferred (a control-flow exception), which the worker task runner catches to write the trigger row and release the slot. Never catch TaskDeferred in a generic try/except Exception; if you need error handling around .defer(), add except TaskDeferred: raise explicitly or put the try/except around the validation logic before the defer.
• Triggerer HA protocol. Multiple airflow triggerer instances share the trigger metadata table. Each Trigger has a triggerer_id column claimed by the current owner. Owners heartbeat every 5s to triggerer_job; stale heartbeats (>30s) release their claims. Failover time is ~30 seconds — invisible for hour-scale waits.
• Async I/O only in run(). Use asyncio.sleep, aiohttp, asyncpg, aiokafka, aiobotocore. Never use time.sleep, requests, blocking DB drivers, or synchronous SDKs — they block the event loop and freeze every other Trigger on the Triggerer.
• Backoff inside run(). Exponential backoff min(base * 2**n, cap) on transient errors (5xx, network, timeouts). Reset the counter on every successful poll. Cap at 60 seconds. Never retry more than the timeout budget allows.
• Prometheus alerts. Wire three alerts before shipping deferrable DAGs: (1) TriggererHeartbeatStale on heartbeat age > 60s for 1m (page); (2) TriggererCapacityHigh on capacity used / total > 0.85 for 5m (warn); (3) TriggerEventLag on p95 event lag > 60s for 2m (page). Missing any one leaves a failure class invisible.
• Global default flag. [operators] default_deferrable = True in airflow.cfg (or AIRFLOW__OPERATORS__DEFAULT_DEFERRABLE=True). Flips the default for every operator whose __init__ reads the config. Prevents accidental classic-mode sensors in new DAGs. Combine with a PR-template check and an Airflow lint rule for defence in depth.
• The 60-second break-even. Any wait longer than 60 seconds is a deferrable candidate. Waits under 60 seconds don't save meaningful cost after Trigger overhead. Above 60 seconds the deferrable path wins by 10× to 1000× depending on wait duration.

Frequently asked questions

What is an Airflow deferrable operator and why does it save 90% of worker cost?

An airflow deferrable operator is an operator that separates its wait logic from its worker-side execution. Instead of holding a worker slot idle for the duration of a wait (as classic mode="poke" and mode="reschedule" sensors do), a deferrable operator's execute() runs briefly on a worker, calls self.defer(trigger=..., method_name=...), and releases the worker slot back to the pool. The wait itself runs as an async coroutine inside the airflow triggerer service — a separate Airflow process that hosts an asyncio event loop capable of concurrently driving hundreds to thousands of Trigger.run() coroutines on a single CPU. When the awaited condition fires (an S3 key lands, an HTTP endpoint returns 200, a query finishes), the Trigger yields a TriggerEvent; the task re-hydrates on a worker to run method_name(context, event) for the final result handling. The 90% cost saving comes from the slot-hour math — a 6-hour classic sensor holds 6 slot-hours idle at $0.08/hour = $0.48 per wait; a deferrable equivalent holds ~2 seconds of worker time (execute + execute_complete) and adds a shared Triggerer overhead of ~$0.001/wait, for a total of ~$0.001 per wait. Multiplied across 400 sensor tasks per day, the monthly savings routinely hit $3000–$4000 per platform.

Do I still need Airflow workers if every task is deferrable?

Yes — workers still run execute() and execute_complete() on every deferrable task, plus every non-deferrable task in your DAGs (Python transforms, SQL runs, custom code, etc). What changes is the amount of worker time per task. A deferrable sensor consumes ~2 seconds of worker time regardless of how long the wait is; a deferrable BigQuery operator consumes ~2 seconds per query, regardless of query duration. So you still size worker capacity, just for the actual productive work (transforms, IO to warehouses, custom Python) rather than for the idle waits. In practice, a deferrable-heavy Airflow deployment can shrink its worker fleet by 60–90% while keeping the same DAG concurrency. The Triggerer replaces the slot-hours previously spent on waits; workers handle the seconds of real work at the start and end of each task.

Can I write my own Trigger class for a wait Airflow doesn't ship?

Absolutely — writing a custom BaseTrigger is a first-class Airflow extensibility point. Subclass airflow.triggers.base.BaseTrigger, implement __init__(**kwargs) to capture state, serialize(self) to return (classpath, kwargs) for reconstruction on Triggerer restart, and async def run(self) to host the async wait loop that finally yields a TriggerEvent({...}). Common custom Trigger patterns include polling internal REST APIs (aiohttp), watching Postgres tables (asyncpg), consuming Kafka topics (aiokafka), waiting on Redis events (aioredis), and sleeping until wall-clock timestamps (asyncio.sleep with a chunked outer loop). The rules: never use blocking I/O in run() (no time.sleep, no requests, no sync DB drivers) — they freeze the whole event loop and stall every other Trigger. Handle transient errors with exponential backoff inside run(). Include everything the wait needs in the serialize() output — anything not captured is unavailable to the async coroutine.

How many Triggerer instances do I need, and how does the HA topology work?

Two Triggerer replicas is the production baseline — enough for HA, cheap enough that scaling further is rarely justified. A single 1-CPU Triggerer handles 500–1000 concurrent Triggers for I/O-bound workloads (polling loops with asyncio.sleep), so two replicas comfortably cover most self-managed deployments. The HA protocol is entirely in the metadata DB: each Trigger row has a triggerer_id column claiming its current owner, each Triggerer writes a heartbeat every 5 seconds to triggerer_job, and stale heartbeats (default triggerer_health_check_threshold = 30 s) mark the Triggerer dead. Live Triggerers run a cleanup query that UPDATEs triggerer_id = NULL for dead-Triggerer claims; the next scheduling cycle re-claims them. Failover is ~30–40 seconds worst case — invisible for hour-scale waits. Deploy on Kubernetes with triggerer.replicas: 2, a PodDisruptionBudget with maxUnavailable: 1, terminationGracePeriodSeconds: 60, and Prometheus alerts on heartbeat age, capacity utilisation, and event lag.

Do deferrable operators work with all sensors and long-running operators?

In 2026, effectively yes for the modern Provider catalogue. The Amazon Provider ships deferrable=True on S3KeySensor, AthenaOperator, EmrJobFlowSensor, RedshiftDataOperator, EcsRunTaskOperator, and more. The Google Provider covers GCSObjectExistenceSensor, BigQueryInsertJobOperator, DataprocSubmitJobOperator, DataflowJobStatusSensor. The HTTP Provider covers HttpSensor and HttpOperator. The Snowflake Provider covers SnowflakeOperator and SnowflakeSqlApiOperator. The Databricks Provider covers DatabricksSubmitRunOperator, DatabricksRunNowOperator, and DatabricksSqlOperator. Core sensors include TimeDeltaSensor, DateTimeSensor, and ExternalTaskSensor. For any operator without a shipped deferrable version, you can write a custom Trigger (10–30 lines of async code) and either subclass the operator or wrap it. The global default [operators] default_deferrable = True in airflow.cfg flips every operator that reads the config to default-on, making deferrable the pit-of-success rather than the opt-in.

How do I test a deferrable operator locally?

Three testing layers. Unit tests for the Trigger — instantiate the Trigger, drive run() under pytest-asyncio with async for event in trigger.run(), mock the external service (aiohttp response mocker, asyncpg fixtures, S3 stubbers), assert the yielded TriggerEvent payload matches expectations. Integration tests for the operator — use Airflow's TaskDeferred assertion helper: run operator.execute(context) and assert it raises TaskDeferred with the correct trigger and method_name. Then separately test execute_complete(context, event) with a mock event dict. End-to-end tests in a docker-compose Airflow — spin up scheduler + webserver + worker + triggerer, run a real DAG with the deferrable operator against a mock external service (a localstack S3 or a mockserver HTTP endpoint), verify the task transitions running → deferred → running → success in the Grid view. For CI, the unit + integration layers are usually enough; end-to-end is manual or nightly. The airflow.triggers.testing module (2.7+) provides a TriggerRunner fixture that drives Triggers in a test event loop without a real Triggerer service.

Practice on PipeCode

• Drill the ETL practice library → for the sensor-driven orchestration, pipeline-cost-audit, and event-driven-wait problems senior interviewers love.
• Rehearse on the SQL practice library → for the metadata-DB cost-audit query patterns and the batch-marker table-based wait designs.
• Sharpen the tuning axis with the optimization practice library → for the slot-hour reduction, Triggerer sizing, and async backoff problems.
• Stack the prerequisites against PipeCode's broader 450+ data-engineering catalogue to anchor the deferrable + Trigger + Triggerer intuition against real graded inputs.

Lock in deferrable-operator muscle memory

Airflow docs describe the API. PipeCode drills describe the decision — when to write a custom Trigger vs use `deferrable=True`, how to size the Triggerer for peak concurrency, why the cost-audit query catches regressions before FinOps does. Pipecode.ai is Leetcode for Data Engineering — pattern-first practice tuned for the production trade-offs senior data engineers actually face.

Practice ETL problems →
Practice optimization problems →