Add task-level circuit breaker to pause failing tasks

Status

Motivation

My bachelor's in Computer Science involves a "final project" which consists of contributing towards eligible open source projects- the two main deliverables of this course are attempting to solve a bug with an open issue (the now merged PR #64235) and attempting to contribute a new feature for the repo (be it associated to an existing ticket or otherwise).

Considerations

What change do you propose to make?

Add a task-level circuit breaker to Apache Airflow. Three optional parameters are added to BaseOperator: circuit_breaker_max_failures, circuit_breaker_window, and circuit_breaker_reset_delay. A new TaskCircuitBreaker model persists circuit state per (dag_id, task_id). When a task exceeds its failure budget within the configured rolling window, the circuit opens and subsequent scheduled instances are skipped until the circuit resets — either automatically after the cooldown period or manually via three new REST API endpoints.

What problem does it solve?

Airflow already has max_consecutive_failed_dag_runs as a DAG-level safeguard, but no equivalent exists at the task level. When a single task fails repeatedly (e.g. a partner API goes down, credentials expire, or an upstream table disappears), every scheduled DAG run continues to attempt and fail that task indefinitely. Per-run retries (BaseOperator.retries) are scoped to a single run with no cross-run failure counter, so there is no built-in way to suppress a persistently failing task without pausing the entire DAG.

Why is it needed?

A full DAG pause is too coarse a response to an isolated task failure. Consider a @hourly DAG with a task that retries 3 times — a 48-hour outage produces 96 failed task instances per day with no suppression and no clear signal beyond FAILED. A task-level circuit breaker limits noise, reduces unnecessary scheduler and executor load, and surfaces the failure state explicitly (via SKIPPED status with a "Circuit open: …" note), making incidents easier to detect and respond to.

Are there any downsides to this change?

- Adds a new database table (task_circuit_breaker) and a new column on TaskInstance, requiring an Alembic migration.
-Operators that set circuit_breaker_max_failures will have tasks silently skipped when the circuit is open, which could be surprising if downstream tasks depend on them — users should be aware that downstream dependencies will also be skipped or marked upstream-failed as usual.
- The feature is opt-in (all three parameters default to None), so there is no impact on existing DAGs.

Which users are affected by the change?

DAG authors who opt in by setting circuit_breaker_max_failures on an operator. All other users are unaffected. Airflow administrators gain three new REST endpoints for inspecting and resetting circuit state.

How are users affected by the change? (e.g. DB upgrade required?)

A DB upgrade (Alembic migration) is required to add the task_circuit_breaker table and the circuit_breaker_max_failures column on task_instance. This follows the same pattern as existing migrations and is handled automatically on airflow db migrate. No DAG-level changes are needed for users who do not use the feature.

What is the level of migration effort (manual and automated) needed for the users to adapt to the breaking changes? (especially in context of Airflow 3)

No breaking changes are introduced. The new parameters are optional and default to None, making this fully backwards-compatible. No DAG author needs to change anything. Users who wish to adopt the feature add the new parameters to relevant operators. No migration utilities are needed.

Other considerations?

The scheduler integration uses time_machine-based tests to verify auto-reset behaviour, and the feature follows established Airflow patterns throughout: SessionDep/AirflowRouter for REST endpoints, the db_cleanup.py batching pattern for bulk resets, and the max_tries/retries mirror pattern for the new TaskInstance column.

What defines this AIP as "done"?

• TaskCircuitBreaker model and Alembic migration merged and passing CI

• BaseOperator updated with three new optional parameters.

• Scheduler logic updated to skip open-circuit TIs and auto-reset on heartbeat.

• Three REST endpoints (GET/POST on circuitBreaker, GET on circuitBreakers) implemented and documented.

• Unit tests (6 cases) and scheduler integration tests (2 cases) passing.

• Smoke test confirmed: a DAG with circuit_breaker_max_failures=2 skips the third TI with a "Circuit open" note.
  Progress tracked at: https://github.com/apache/airflow/pull/67724