Rerun Airflow Dags and tasks

You can set when to run Airflow Dags using a wide variety of scheduling options. Some use cases where you might want tasks or Dags to run outside of their regular schedule include:

You want one or more tasks to automatically run again if they fail.
You need to manually rerun a failed task for one or multiple Dag runs.
You want to deploy a Dag with a start date of one year ago and trigger all Dag runs that would have been scheduled in the past year.
You have a running Dag and realize you need it to process data for two months prior to the Dag’s start date.

In this guide, you’ll learn how to configure automatic retries, rerun tasks or Dags, trigger historical Dag runs, and review the Airflow concepts of catchup and backfill.

To learn more, watch the Best practices for pipeline failure recovery webinar.

Assumed knowledge

To get the most out of this guide, you should have an understanding of:

Dag scheduling. See Schedule Dags in Airflow

Automatically retry tasks

In Airflow, you can configure tasks to retry automatically when they fail. You can set a fixed number of retries with a delay between them using retry parameters, and in Airflow 3.3+ you can use retry policies to decide how to handle a failure based on the exception that caused it.

A retry runs the task from the beginning. To keep work that already completed, combine retries with a durable execution pattern: save progress with the task state store so a retry resumes from the last checkpoint.

Retry parameters

The following task parameters control the retry behavior of a task. Some of them have an Airflow configuration that sets the default for all tasks in your Airflow environment.

Task parameter	Airflow configuration	Default	Description
`retries`	`AIRFLOW__CORE__DEFAULT_TASK_RETRIES`	`0`	The number of times a task instance retries before it fails. You can think of this as the number of additional attempts your task instance gets.
`retry_delay`	`AIRFLOW__CORE__DEFAULT_TASK_RETRY_DELAY`	300 seconds	The time between retries.
`retry_exponential_backoff`	-	`False`	Set to `True` to double the delay with each retry, up to a maximum of `max_retry_delay`. In Airflow 3.2+, you can set a float to define the factor the delay is multiplied by between retries, for example `3.0`.
`max_retry_delay`	`AIRFLOW__CORE__MAX_TASK_RETRY_DELAY`	24 hours	The maximum time between retries when you use `retry_exponential_backoff`.
`retry_policy`	-	`None`	A retry policy that decides how to handle a failure based on the exception. Airflow 3.3+.

It’s common practice to set the retry parameters for all tasks in a Dag with default_args and override them for specific tasks as needed. The following Dag contains four tasks that always fail. Each of the tasks uses a different retry parameter configuration.

from airflow.sdk import dag
from airflow.providers.standard.operators.bash import BashOperator
from pendulum import datetime, duration


@dag(
    start_date=datetime(2023, 4, 1),
    schedule="@daily",
    default_args={
        "retries": 3,
        "retry_delay": duration(seconds=2),
        "retry_exponential_backoff": True,
        "max_retry_delay": duration(hours=2),
    },
)
def retry_example():
    t1 = BashOperator(task_id="t1", bash_command="echo I get 3 retries! && False")

    t2 = BashOperator(
        task_id="t2",
        bash_command="echo I get 6 retries and never wait long! && False",
        retries=6,
        max_retry_delay=duration(seconds=10),
    )

    t3 = BashOperator(
        task_id="t3",
        bash_command="echo I wait exactly 20 seconds between each of my 4 retries! && False",
        retries=4,
        retry_delay=duration(seconds=20),
        retry_exponential_backoff=False,
    )

    t4 = BashOperator(
        task_id="t4",
        bash_command="echo I have to get it right the first time! && False",
        retries=0,
    )


retry_example()

Retry policies

In Airflow 3.3+, you can attach a retry policy to a task with the retry_policy parameter. When a task instance fails, the retry policy evaluates the exception and decides what happens next. For example, you can retry a task instance that failed because of an error that is likely transient, such as a 503 error, and fail a task instance immediately that failed because of an error that requires manual intervention, such as invalid credentials. The following example attaches an ExceptionRetryPolicy to a task. The policy contains one RetryRule, which maps the PermissionError exception to the RetryAction.FAIL action:

from airflow.sdk import ExceptionRetryPolicy, RetryAction, RetryRule, task

fail_on_permission_error = ExceptionRetryPolicy(
    rules=[
        RetryRule(exception=PermissionError, action=RetryAction.FAIL),
    ]
)


@task(retries=3, retry_policy=fail_on_permission_error)
def load_orders():
    raise PermissionError("Insufficient privileges to operate on table 'ORDERS'")

When load_orders raises a PermissionError, the task instance fails immediately instead of using its three retries. Any other exception uses the retry behavior defined on the task. Internally, every retry policy returns a RetryDecision object that contains the action, and optionally a delay and a reason. You only use RetryDecision in your code when you write a custom retry policy. Each decision is written to the task logs, for example:

INFO - Retry policy decision action=fail reason=Matched rule for PermissionError

AirflowFailException and AirflowSensorTimeout always fail the task instance immediately. Airflow doesn’t evaluate the retry policy for these exceptions.

You can use the following retry policies:

Exception retry policy: Maps exception types to retry actions.
AI-based retry policies: LLMRetryPolicy and ClassifierRetryPolicy use a model to classify the failure. Requires the Common AI provider.
Chain retry policy: Combines several retry policies in order.
Custom retry policy: Defines your own logic in a subclass of RetryPolicy.

Retry actions

There are four possible retry actions:

RetryAction.DEFAULT: Use the retry behavior defined by the retry parameters on the task itself.
RetryAction.FAIL: Fail the task instance immediately without using its remaining retries.
RetryAction.RETRY: Retry the task instance, optionally with a different delay. The maximum total number of retries a task instance can get is limited by the retries parameter.
RetryAction.RESCHEDULE: Airflow 3.4+. Run the task instance again later without using a retry. Airflow sets the state of the task instance to up_for_reschedule. When you use this action, you need to set max_reschedules, reschedule_timeout, or both on the RetryRule or RetryDecision.
from datetime import timedelta from airflow.sdk import RetryAction, RetryRule RetryRule( exception=ClusterBusyError, action=RetryAction.RESCHEDULE, retry_delay=timedelta(seconds=30), max_reschedules=3, reschedule_timeout=timedelta(minutes=10), reason="Cluster has no free capacity yet", )
max_reschedules determines how often the task instance can be rescheduled. reschedule_timeout determines how much time can pass after the first reschedule before Airflow stops rescheduling the task instance. After that, Airflow uses the retry behavior defined on the task, including its retry_delay instead of the retry_delay of the rule. Both limits apply per try: when the task instance is retried, for example because a different rule returns RetryAction.RETRY, the count and the time start over.

The following table compares RetryAction.RETRY and RetryAction.RESCHEDULE:

	`RetryAction.RETRY`	`RetryAction.RESCHEDULE`
Use when	The task instance failed while doing its work, and you want a new attempt.	The task instance can’t do its work yet because a resource it needs isn’t available.
Example	A network connection times out while the task instance loads data.	A compute cluster has no free capacity yet.
Uses one of the `retries`	Yes	No
Try number	Increases by one	Stays the same
Logs	New logs for the new try	Added to the logs of the same try
`on_retry_callback`	Runs	Doesn’t run
Limited by	`retries`	`max_reschedules`, `reschedule_timeout`, or both. After that, Airflow uses the retry behavior defined on the task.

Exception retry policy

ExceptionRetryPolicy maps exception types to actions with a list of RetryRule objects. Each rule defines the exception, the action, and optionally a custom retry delay and a reason. Airflow evaluates the rules in order and the first rule that matches the exception wins. If no rule matches, the task instance uses the retry behavior defined on the task. You can define the exception of a rule as an exception class, as a dotted import path string, or as a list of either. By default, a rule also matches subclasses of the exception. Set match_subclasses=False to only match the exact exception type.

from datetime import timedelta

from airflow.sdk import ExceptionRetryPolicy, RetryAction, RetryRule, task

FILE_RETRY_POLICY = ExceptionRetryPolicy(
    rules=[
        RetryRule(
            exception=TimeoutError,
            action=RetryAction.RETRY,
            retry_delay=timedelta(seconds=30),
            reason="Request timed out",
        ),
        RetryRule(
            exception="builtins.OSError",
            match_subclasses=False,
            action=RetryAction.RETRY,
            retry_delay=timedelta(minutes=2),
            reason="Transient I/O error",
        ),
        RetryRule(
            exception=[FileNotFoundError, "builtins.PermissionError"],
            action=RetryAction.FAIL,
            reason="Missing file or permissions, a person needs to fix it",
        ),
    ],
)


@task(retries=4, retry_delay=timedelta(seconds=30), retry_policy=FILE_RETRY_POLICY)
def process_files():
    ...

In this example:

A TimeoutError retries after 30 seconds. The rule uses the exception class.
A plain OSError, such as a generic input/output error, retries after two minutes. The rule uses a dotted import path and match_subclasses=False, so it doesn’t match subclasses of OSError such as FileNotFoundError and PermissionError. Without match_subclasses=False, this rule would retry those errors too. TimeoutError is also a subclass of OSError, but the first rule already matches it.
A FileNotFoundError or PermissionError fails the task instance immediately. The rule uses a list that combines an exception class and a dotted import path.
Any other exception, such as a ValueError, uses the retry behavior defined on the task.

For more information, see the Airflow documentation.

AI-based retry policies

The Common AI provider includes two retry policies that use a model to classify the failure. Both send the exception message to the model provider of your connection.

LLMRetryPolicy: A text model classifies the failure and decides whether to retry and how long to wait, based on your instructions.
ClassifierRetryPolicy: A classifier model sorts the failure into one of the categories you define, and the matching ErrorCategory decides whether the task instance retries and after how long.

The following table compares the two policies:

	`LLMRetryPolicy`	`ClassifierRetryPolicy`
Model	Text model, such as `anthropic:claude-sonnet-5-5` or `openai:gpt-6-luna`	Designed for decision models, which report a confidence for their answer, such as `openai-decisions:gpt-6-luna` or `typesafe:jev-1.13.0`.
You provide	`instructions` that describe how to classify failures and when to retry	`categories`, with an `ErrorCategory` for each that defines whether to retry and the delay
The model returns	A category, whether to retry, how long to wait, and a reason	One of the categories you define, and the decision confidence
Confidence threshold	Not available	`min_confidence` on the policy or on an `ErrorCategory`
If the model call fails	`fallback_rules`, then the retry behavior defined on the task	`fallback_policy`, then `fallback_rules`, then the retry behavior defined on the task. The same applies if the decision confidence is below `min_confidence`.

By default, both policies mask values that Airflow registered as secrets, such as connection passwords, but not other sensitive information in the exception message. Use the redactor parameter to define your own masking. See Custom redactors in the Common AI provider documentation.

To use these retry policies, install the Common AI provider and define a PydanticAI connection. You can use the model_id parameter of the policy to override the model of the connection. The following example uses an LLMRetryPolicy with a text model. If the model call fails, the policy uses its fallback_rules:

from datetime import timedelta

from airflow.providers.common.ai.policies.retry import LLMRetryPolicy
from airflow.sdk import RetryAction, RetryRule, task

llm_policy = LLMRetryPolicy(
    llm_conn_id="pydanticai_default",
    model_id="openai:gpt-6-luna",
    fallback_rules=[
        RetryRule(exception=ConnectionError, action=RetryAction.RETRY, retry_delay=timedelta(seconds=10)),
    ],
)


@task(retries=3, retry_policy=llm_policy)
def call_external_api():
    raise RuntimeError("429 Too Many Requests: Rate limit exceeded. Retry after 60 seconds.")

The following example uses a ClassifierRetryPolicy with a decision model and three categories:

from datetime import timedelta

from airflow.providers.common.ai.policies.retry import ClassifierRetryPolicy, ErrorCategory
from airflow.sdk import task

classifier_policy = ClassifierRetryPolicy(
    llm_conn_id="pydanticai_default",
    model_id="openai-decisions:gpt-6-luna",
    min_confidence=0.8,
    categories={
        "rate_limited": ErrorCategory(
            "The API rejected the request because of a rate limit.",
            retry=True,
            delay=timedelta(minutes=1),
        ),
        "invalid_credentials": ErrorCategory(
            "The API rejected the credentials.",
            retry=False,
        ),
        "other": ErrorCategory("Any other failure.", retry=True),
    },
)


@task(retries=3, retry_policy=classifier_policy)
def call_external_api():
    raise RuntimeError("401 Unauthorized: Invalid API key provided.")

Astronomer recommends that you define an other category. The model always picks one of the categories you define. Without an other category, the model can sort an unrelated failure into one of your specific categories, sometimes even with high confidence. Use the retry parameter of the other category to decide whether failures that fit none of your categories retry or fail.

Chain retry policy

In Airflow 3.4+, ChainRetryPolicy combines several retry policies. Airflow evaluates the policies in order, and the first decision that isn’t RetryAction.DEFAULT wins.

If a policy raises an error, for example because of a bug in the policy code, Airflow evaluates the next policy.

The following example first checks for a known failure without calling a model, and then asks a decision model to classify the failure with a ClassifierRetryPolicy.

from datetime import timedelta

from airflow.providers.common.ai.policies.retry import ClassifierRetryPolicy, ErrorCategory
from airflow.sdk import ChainRetryPolicy, ExceptionRetryPolicy, RetryAction, RetryRule, task

known_failures = ExceptionRetryPolicy(
    rules=[
        RetryRule(exception=PermissionError, action=RetryAction.FAIL),
    ]
)

decision_model = ClassifierRetryPolicy(
    llm_conn_id="pydanticai_default",
    model_id="openai-decisions:gpt-6-luna",
    min_confidence=0.8,
    categories={
        "warehouse_suspended": ErrorCategory(
            "The warehouse is suspended and will auto-resume.",
            retry=True,
            delay=timedelta(seconds=30),
        ),
        "other": ErrorCategory("Any other failure.", retry=True),
    },
)

chain_policy = ChainRetryPolicy([known_failures, decision_model])


@task(retries=3, retry_delay=timedelta(minutes=1), retry_policy=chain_policy)
def query_warehouse():
    ...

In this example:

If the failure is a PermissionError, known_failures fails the task instance immediately without calling the model.
If the failure isn’t a PermissionError, the decision model classifies it. If the model classifies the failure as warehouse_suspended with a confidence of at least 0.8, the task instance retries after 30 seconds.
If the failure matches neither, the decision model classifies it as other. This category doesn’t define a delay, so the task instance retries with the retry_delay defined on the task, after one minute.

Custom retry policies

You can also implement a custom retry policy by subclassing RetryPolicy and implementing its evaluate() method. This lets you inspect exception attributes or use context that ExceptionRetryPolicy doesn’t have access to. Some errors can only be distinguished by an attribute on the exception rather than by the exception class itself. For example, Snowflake errors carry an error code in their errno attribute. A custom policy can implement retry logic on those attributes:

from datetime import timedelta

from airflow.sdk import RetryDecision, RetryPolicy


class SnowflakeErrorCodeRetryPolicy(RetryPolicy):
    """Route Snowflake errors by error code rather than exception class."""

    RESOURCE_PRESSURE_CODES = {605, 606}     # statement canceled, resource shortage
    SCHEMA_ERROR_CODES = {2003, 2043, 3001}  # missing object, cannot drop, invalid identifier

    def evaluate(self, exception, try_number, max_tries, context=None):
        errno = getattr(exception, "errno", None)
        if errno in self.RESOURCE_PRESSURE_CODES:
            return RetryDecision.retry(delay=timedelta(minutes=5))
        if errno in self.SCHEMA_ERROR_CODES:
            return RetryDecision.fail(reason=f"Snowflake error {errno}, retries won't fix it")
        return RetryDecision.default()

Another common example would be customizing retry behavior based on how much time has passed since the Dag started running. If your Dag produces time-sensitive data products, you may want the task instance to fail sooner so the team is notified, rather than waiting on multiple retries.

from datetime import timedelta

from airflow.sdk import RetryDecision, RetryPolicy, timezone


class StaleRunRetryPolicy(RetryPolicy):
    """Stop retrying once the run is too stale to be useful."""

    STALENESS_THRESHOLD = timedelta(hours=6)

    def evaluate(self, exception, try_number, max_tries, context=None):
        if context is None:
            return RetryDecision.default()
        run_after = context["dag_run"].run_after
        if timezone.utcnow() - run_after > self.STALENESS_THRESHOLD:
            return RetryDecision.fail(
                reason="Run is past the staleness threshold, escalating"
            )
        return RetryDecision.default()

Automatically pause a failing Dag

You can configure Airflow to automatically pause a Dag after a certain number of failed Dag runs, preventing a failing Dag from continuing to run and potentially causing more issues. To set the maximum number of consecutive failed Dag runs for all your Dags in your Airflow environment, set the core.max_consecutive_failed_dag_runs_per_dag config. For example, to automatically pause all your Dags after they had 5 failed Dag runs in a row, set:

AIRFLOW__CORE__MAX_CONSECUTIVE_FAILED_DAG_RUNS_PER_DAG=5

You can override this setting for a specific Dag by setting the max_consecutive_failed_dag_runs parameter in the Dag instantiation. For example, to pause a specific Dag after 3 failed Dag runs in a row, set:

TaskFlow
Traditional

# from airflow.sdk import dag
# from pendulum import datetime

@dag(
    start_date=datetime(2024, 4, 1),
    schedule="@daily",
    max_consecutive_failed_dag_runs=3,
    catchup=False,
)
def my_dag():
    # Define your tasks here

my_dag()

# from airflow.sdk import DAG
# from pendulum import datetime

with DAG(
    dag_id="my_dag",
    start_date=datetime(2024, 4, 1),
    schedule="@daily",
    max_consecutive_failed_dag_runs=3,
    catchup=False,
):
    # Define your tasks here

The max_consecutive_failed_dag_runs config and Dag-level parameter is currently experimental and might be subject to breaking changes in future releases.

Manually rerun tasks or Dags

Rerunning tasks or full Dags in Airflow is a common workflow. To rerun a task in Airflow, you clear the task instance. Clearing sets the state of the task instance to None, so the scheduler runs it again. A cleared task instance gets its full number of retries again, no matter how many tries it used before. To clear a task instance, go to your Dag in the Airflow UI, select the task instance you want to rerun, and click Clear Task Instance.

A popup window appears, giving you the following options to clear and rerun additional task instances related to the selected task:

Past: Clears any instances of the task in Dag runs with a logical date before the selected task instance.
Future: Clears any instances of the task in Dag runs with a logical date after the selected task instance.
Upstream: Clears any tasks in the current Dag run which are upstream from the selected task instance.
Downstream: Clears any tasks in the current Dag run which are downstream from the selected task instance.
Clear only failed tasks: Clears only failed instances of any task instances selected based on the previous options.
Keep task state and resume: Keeps the task state store entries of the cleared task instances. By default, clearing a task instance discards them. See Clearing behavior.
Prevent rerun if task is running: Skips task instances that are currently running.
Run with latest bundle version: Runs the task instances with the latest version of the Dag code. This option only appears for Dags in a versioned Dag bundle when the Dag changed since the Dag run. See Run on latest version.

The window shows which task instances will be cleared with the current settings. Click Confirm and the task(s) will be cleared and rescheduled for another run. Clearing a task instance discards its task state store entries by default. To keep them, select Keep task state and resume. See Clearing behavior.

You can also clear task instances programmatically with the Airflow CLI or with the POST /api/v2/dags/{dag_id}/clearTaskInstances endpoint of the Airflow REST API. To clear a full Dag run, click the Dag run, and then click Clear Dag Run as shown in the following image.

Don’t clear or change task statuses directly in the Airflow metastore. This can cause unexpected behavior in Airflow.

Add notes to cleared tasks and Dags

You can add notes to task instances and Dag runs in the Airflow UI. This feature is useful for tracking manual changes to task instances, such as reruns or task status changes. Astronomer recommends leaving a note on a task or Dag whenever you manually update a task instance through the Airflow UI. To add a note to a task instance or Dag run:

Go to your Dag in the Airflow UI.
Select a task instance or Dag run.
Click Add a note.
Write a note and click Confirm.

Catchup

You can use the built-in catchup Dag argument to process data for logical dates between the set start_date of a Dag and the current date. When the catchup parameter for a Dag is set to True, at the time the Dag is turned on in Airflow, the scheduler starts a Dag run for every data interval that hasn’t been run between the Dag’s start_date and the current data interval. For example, if your Dag is scheduled to run daily and has a start_date of 1/1/2025, and you deploy that Dag and turn it on 2/1/2025, Airflow will schedule and start all of the daily Dag runs for January. Catchup is also triggered when you turn a Dag off for a period and then turn it on again. Catchup can be controlled by setting the parameter in your Dag’s arguments. By default, catchup is set to False. This example Dag has catchup turned on:

@dag(
    dag_id="example_dag",
    start_date=datetime(2025, 4, 23),
    max_active_runs=1,
    schedule="@daily",
    default_args={
        "retries": 1,
        "retry_delay": timedelta(minutes=3),
    },
    catchup=True
)

Catchup is a powerful feature, but it should be used with caution. For example, if you deploy a Dag that runs every 5 minutes with a start date of 1 year ago and set catchup to True, Airflow will schedule numerous Dag runs all at once. When using catchup, keep in mind what resources Airflow has available and how many Dag runs you can support at one time. To avoid overloading your scheduler or external systems, you can use the following parameters in conjunction with catchup:

max_active_runs: Set at the Dag level and limits the number of Dag runs that Airflow will execute for that particular Dag at any given time. For example, if you set this value to 3 and the Dag had 15 catchup runs to complete, they would be executed in 5 chunks of 3 runs.
depends_on_past: Set at the task level or as a default_arg for all tasks at the Dag level. When set to True, the task instance must wait for the same task in the most recent Dag run to be successful. This ensures sequential data loads and allows only one Dag run to be executed at a time in most cases.
wait_for_downstream: Set at the task level or as a default_arg for all tasks at the Dag level. When set to True, the task instance waits for the same task in the previous Dag run and the tasks immediately downstream of it to succeed or be skipped. Setting wait_for_downstream to True also sets depends_on_past to True.

If you want to set catchup to True by default for all Dags in your Airflow environment, for example for migration purposes, you can set the Airflow config AIRFLOW__SCHEDULER__CATCHUP_BY_DEFAULT to True. If you want to deploy your Dag with catchup enabled but there are some tasks you don’t want to run during the catchup, you can use the LatestOnlyOperator in your Dag. This operator only runs during the Dag’s most recent scheduled interval. In every other Dag run it is ignored, along with any tasks downstream of it.

Backfill

Backfilling is the concept of running a Dag for a specified period in the past to re-process historical or missed data. Unlike catchup, which triggers missed Dag runs from the Dag’s start_date through the current data interval, backfill periods can be specified explicitly and can include periods prior to the Dag’s start_date. In Airflow 3, backfills are managed by the scheduler and can be triggered through the UI, API, or CLI. To run a backfill from the UI, click the blue Trigger button and select Backfill. Choose the date range you want to backfill for, and which runs you want to reprocess. You also have the option to select the number of max active runs for the backfill, whether you want to run backwards or forwards, specify run parameters and select one of three reprocessing behaviors:

Missing Runs: Creates and runs only the Dag runs that don’t already exist for the selected period.
Missing and Errored Runs: Creates any missing runs and also re-runs any existing Dag runs that previously failed.
All Runs: Clears and re-runs all existing Dag runs within the date range, in addition to creating any that are missing.

Execute the backfill by clicking Run Backfill. The UI tells you how many runs will be triggered and the backfill uses the latest Dag version that is available for your Dag.

Once the backfill has started, you can pause or cancel it at any time in the UI. Backfilled Dag runs are indicated in your grid by the u-turn arrow. To see an example of backfilling using the CLI, see the Airflow docs. For information on how to backfill using the Airflow REST API see the Airflow REST API docs. When using backfill, make sure to consider your available resources. If your backfill will trigger many Dag runs, and/or you have many other Dags running at the same time, you should set a max active runs that won’t overload your scheduler.

​Assumed knowledge

​Automatically retry tasks

​Retry parameters

​Retry policies

​Retry actions

​Exception retry policy

​AI-based retry policies

​Chain retry policy

​Custom retry policies

​Automatically pause a failing Dag

​Manually rerun tasks or Dags

​Add notes to cleared tasks and Dags

​Catchup

​Backfill

Assumed knowledge

Automatically retry tasks

Retry parameters

Retry policies

Retry actions

Exception retry policy

AI-based retry policies

Chain retry policy

Custom retry policies

Automatically pause a failing Dag

Manually rerun tasks or Dags

Add notes to cleared tasks and Dags

Catchup

Backfill