Resumable Tasks

Added in version 3.3.0.

Many data engineering workflows involve submitting work to an external system and waiting for it to complete. A Spark job, a BigQuery query, a Kubernetes batch pod, an EMR step: these are all tasks where the real work happens outside Airflow, and the operator’s job is mostly to submit, poll, and collect the result.

These tasks share a common failure mode. In classic operator cases, the worker slot is held for the entire polling duration, and if the worker process is restarted or the host is preempted, the task retries from scratch, losing all the progress made. Depending on the operator, that means the external job is submitted again, creating a duplicate run in context of the external system.

Durable execution

Surviving that failure mode is what durable execution means: a task outlives the loss of the process running it and can continue / re-attach from where it stopped, rather than repeating work already done or submitting a duplicate job to an external system.

The task state store is the mechanism Airflow provides to achieve it. It is the only per-task-instance storage that outlives a worker crash and is still readable by the next attempt, which is what lets a retry recover a checkpoint or an external job identifier written by the attempt before it. Durable execution is the outcome; the task state store is how you get it.

Operator documentation across the provider ecosystem uses the same term for the operator-level feature built on this mechanism, normally exposed as a durable parameter. Spark, Kubernetes, Databricks, Snowflake, BigQuery, Redshift and Glue operators each have a “Durable execution” section describing what they persist and how they reconnect on retry.

Airflow recommends three approaches for achieving this with long-running external work. Understanding the trade-offs between them helps you choose the right one for your situation.

Deferrable Operators

A deferrable operator pauses itself at the point where it would otherwise start polling, hands the polling work to the Triggerer component, and releases its worker slot. When the external condition is met, the Triggerer wakes the task and the worker resumes from where the operator left off.

This is the most resource-efficient option. A single Triggerer process can concurrently watch thousands of conditions, so the rest of the worker pool stays free for other tasks.

The trade-offs are:

  • A Triggerer component must be running. Deployments that do not include a Triggerer cannot use this pattern.

  • Writing a custom deferrable operator requires implementing a dedicated Trigger class in addition to the operator itself.

  • The polling logic runs inside the Triggerer’s async event loop. Blocking calls inside a Trigger stall the entire Triggerer process.

If a deferrable operator already exists for your use case, or your team is comfortable implementing one, this is the recommended path considering its resource efficiency.

For more details, see Deferrable Operators & Triggers.

Resumable Tasks

A resumable task uses the task state store to persist a checkpoint before it would otherwise lose progress. On retry, the task reads that checkpoint and continues from where it left off rather than starting over.

The worker slot is held for the full duration of the task, the same as a standard synchronous operator. The benefit is crash safety and continuity, not resource efficiency.

Resumable tasks are useful when:

  • No deferrable operator exists for your external system and writing one is not practical.

  • You want crash recovery without operating a Triggerer.

  • The task processes work incrementally (for example, reading files from a list or paginating through API results) and should be able to resume from its last completed batch.

General pattern

The task reads a checkpoint from task_state_store at the start, does some work, writes an updated checkpoint, and either continues or finishes. On the next run (whether due to a retry after a crash or a deliberate reschedule), it reads the checkpoint again and picks up from there.

from airflow.sdk import dag, task


@dag(schedule=None)
def process_files_dag():

    @task(retries=5)
    def process_files(context=None):
        task_state_store = context["task_state_store"]
        files = ["a.csv", "b.csv", "c.csv", "d.csv"]

        last_processed = task_state_store.get("last_processed")
        start_index = 0
        if last_processed is not None:
            start_index = files.index(last_processed) + 1

        for file in files[start_index:]:
            # ... process the file ...
            task_state_store.set("last_processed", file)

    process_files()


process_files_dag()

This pattern works without any additional work, relying only on context. The state store is just a key-value store scoped to the task instance, and what you checkpoint is up to you.

Resumable operators for external jobs

When the task submits a job to an external system and then polls for completion, there is an additional problem: on retry, the task would submit a second job even though the first one may still be running. Instead of having to handle this manually, the ResumableJobMixin solves this by persisting the external job identifier before polling starts, and reconnecting to the existing job on retry instead of submitting a new one.

For more details and a working example, see ResumableJobMixin.

Retries resume, clearing starts over

A retry keeps the task’s task_state_store entries, which is what makes crash recovery work: the next attempt reads the checkpoint or the external job id written by the attempt before it.

Clearing a task discards them. Clearing means “run this again”, and a checkpoint records how far a task got, not what it got there with. If you fixed the code or the upstream data and cleared the task, resuming would leave the work done before the fix in place and silently mix it with the corrected work. So by default a cleared task starts from the beginning.

To resume from the checkpoint instead, set keep_task_state when clearing, or tick the corresponding box in the clear dialog. That is the right choice when nothing about the inputs or the code changed and you only want the task to carry on where it stopped.

This applies to clearing individual task instances through the REST API, the UI, or airflowctl dags clear (which clears every task instance in the matched Dag run(s) through this same endpoint). Clearing an entire Dag run through the Clear Run dialog/API, marking a task as failed or success (which clears downstream tasks as a side effect), and the core CLI’s airflow tasks clear / airflow dags clear (which call clear_task_instances() directly and never reach the discard endpoint) all still keep task state unconditionally today; see #72929.

Clearing a task that submitted an external job

For an operator with durable execution the stored value is an external job id, so discarding it has a different consequence: the next attempt submits a new job rather than reconnecting to the existing one.

Whether that matters depends on what happened to the job:

  • Most operators cancel the external job in on_kill, so clearing a running task stops the job and there is nothing left to reconnect to. Submitting a fresh one is the only option anyway.

  • An operator configured, or defaulting, to leave the job running on kill keeps it alive, so a fresh submission runs alongside it. Check the operator’s own docs — for example GlueJobOperator defaults stop_job_run_on_kill to False and so leaves the job running unless you turn it on.

  • Clearing a failed task never runs on_kill at all, so an external job that outlived the worker is still running.

  • A deferred task has no worker process to run on_kill on. Instead, the Triggerer cancels the orphaned trigger and runs the trigger’s on_kill, bounded by [triggerer] on_kill_timeout. Most triggers cancel the external job there too. GlueJobCompleteTrigger and LivyTrigger don’t implement on_kill, but neither writes a job id to the state store when deferred either, so keep_task_state will not help for them. For a deferred GlueJobOperator, durable defaults to True, so it reattaches to a still-running Glue run by default and clearing it does not start over; stop the Glue run before clearing, or set durable=False, to force a fresh one. For a deferred LivyOperator, cancel the batch yourself before clearing.

In the cases where the job is left running (the second bullet or a failed task), pass keep_task_state so the next attempt reconnects to the job already in flight instead of paying for a second one.

Note that [state_store] clear_on_success is a separate control: it discards a task’s entries as soon as it reaches SUCCESS, so nothing is left for a later clear to find either way (see Task and Asset State Store Configuration).

Asynchronous Tasks

Note

Async task support applies to Python tasks only: @task decorated async def functions and class-based operators that subclass BaseAsyncOperator. It is not available for other operator types.

Python tasks support async/await syntax. When the decorated callable is an async function, Airflow runs it inside an event loop, which lets you fan out many concurrent I/O operations (HTTP requests, database queries, file reads) within a single task execution without blocking the event loop while waiting for each one.

The worker slot is held for the full duration of the task. Async tasks are not designed for long external waits; they are designed for high-throughput I/O work that completes within a single execution. To survive a worker crash, an async task can checkpoint its progress to task state store with aget/aset and resume from the checkpoint on retry.

For guidance on when to use async tasks versus deferrable operators, see Deferred vs Async Operators.

Comparison

Characteristic

Deferrable operator

Resumable task

Async task

Worker slot during external wait

Freed

Held

Held

Requires Triggerer

Yes

No

No

Handles crash recovery

Yes (via Triggerer)

Yes (via task state store checkpoint)

No

Prevents duplicate job submission

Depends on operator implementation

Yes (with ResumableJobMixin)

Not applicable

Suitable for concurrent I/O fan-out

No

No

Yes

Available from

Airflow 2.2

Airflow 3.3

Airflow 3.2

Retry policies and durable execution

Long-running tasks typically need two separate things together: a retry policy, and durable execution as defined above. Each is configurable per task, and together they let a workflow span days without losing progress.

A retry policy decides whether a failed attempt gets another try, and how long to wait before it. This is RetryPolicy and RetryDecision (see Retry Policies), or the LLM-driven LLMRetryPolicy, which uses a model to read the error and make that call. That policy adds its own fallback_rules, applied when the classification call itself fails, so the decision does not depend on the model being reachable (see Retry policies).

Durable execution is what that next attempt resumes from. Backed by the task state store described above, it is what lets a task recover a checkpoint written by the attempt before it, rather than starting over.

Long-running tasks, or LLM-driven tasks like agentic workflows, benefit most from both: the retry policy decides whether the error is worth retrying at all, and the task state store is what the retry resumes from.

Consider using both when a task:

  • Runs long enough that a worker crash mid-run is a real risk, and

  • Needs a retry decision more nuanced than “always retry” or “never retry”, for example retrying rate limits but failing outright on bad credentials.

The two operate independently. A task with retries=5 and a checkpoint still stops for good after six failed attempts (the initial attempt plus five retries), but each of those retries picks up from the last checkpoint instead of reprocessing files it already finished, or resubmitting a job that is still running.

Was this entry helpful?