This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new a7b60ec14ef Fix backfill DagRuns being permanently starved from
scheduling (#72730)
a7b60ec14ef is described below
commit a7b60ec14ef6d98169d84ddce37c933c3016fb3d
Author: Sean Muth <[email protected]>
AuthorDate: Wed Sep 16 00:42:22 2026 -0500
Fix backfill DagRuns being permanently starved from scheduling (#72730)
---
airflow-core/docs/core-concepts/backfill.rst | 6 +++
airflow-core/src/airflow/models/dagrun.py | 3 --
airflow-core/tests/unit/models/test_dagrun.py | 63 +++++++++++++++++++++++++++
docs/spelling_wordlist.txt | 1 +
4 files changed, 70 insertions(+), 3 deletions(-)
diff --git a/airflow-core/docs/core-concepts/backfill.rst
b/airflow-core/docs/core-concepts/backfill.rst
index 298a7a32e7e..e8ecf675e1a 100644
--- a/airflow-core/docs/core-concepts/backfill.rst
+++ b/airflow-core/docs/core-concepts/backfill.rst
@@ -42,6 +42,12 @@ You can set ``max_active_runs`` on a backfill and it will
control how many Dag r
the backfill can run concurrently. Backfill ``max_active_runs`` is applied
independently
the Dag ``max_active_runs`` setting.
+Once a backfill's Dag runs are actually running, the scheduler does not
otherwise
+deprioritize their task scheduling relative to other running Dag runs.
``max_active_runs``
+is therefore the primary lever for controlling how much of your scheduling
capacity a
+backfill consumes at once — lower it to leave more room for time-sensitive
scheduled work
+while the backfill is in progress.
+
Run ordering
------------
diff --git a/airflow-core/src/airflow/models/dagrun.py
b/airflow-core/src/airflow/models/dagrun.py
index 2941005a15b..c22fbc4a75a 100644
--- a/airflow-core/src/airflow/models/dagrun.py
+++ b/airflow-core/src/airflow/models/dagrun.py
@@ -756,7 +756,6 @@ class DagRun(Base, LoggingMixin):
:meta private:
"""
- from airflow.models.backfill import BackfillDagRun
from airflow.models.dag import DagModel
query = (
@@ -764,13 +763,11 @@ class DagRun(Base, LoggingMixin):
.with_hint(cls, "USE INDEX (idx_dag_run_running_dags)",
dialect_name="mysql")
.where(cls.state == DagRunState.RUNNING)
.join(DagModel, DagModel.dag_id == cls.dag_id)
- .join(BackfillDagRun, BackfillDagRun.dag_run_id == DagRun.id,
isouter=True)
.where(
DagModel.is_paused == false(),
DagModel.is_stale == false(),
)
.order_by(
- nulls_first(cast("ColumnElement[Any]",
BackfillDagRun.sort_ordinal), session=session),
nulls_first(cast("ColumnElement[Any]",
cls.last_scheduling_decision), session=session),
cls.run_after,
)
diff --git a/airflow-core/tests/unit/models/test_dagrun.py
b/airflow-core/tests/unit/models/test_dagrun.py
index a4ce3ccc44e..ffa22350803 100644
--- a/airflow-core/tests/unit/models/test_dagrun.py
+++ b/airflow-core/tests/unit/models/test_dagrun.py
@@ -4747,6 +4747,69 @@ def
test_get_running_dag_runs_to_examine_eager_loads_dag_tags(dag_maker, session
assert dr.stats_tags == {"dag_id": "eager_tag_dag", "run_type":
dr.run_type, "env": "prod"}
+def
test_get_running_dag_runs_to_examine_does_not_starve_backfill_runs(dag_maker,
session, monkeypatch):
+ """A running backfill DagRun with no scheduling decision yet must not be
starved out of the
+ per-loop examine batch by ordinary running DagRuns, regardless of how far
back in its backfill's
+ own ordering it sits.
+
+ Regression test for the scheduler ordering backfill DagRuns strictly
behind every non-backfill
+ DagRun (via `BackfillDagRun.sort_ordinal` nulls-first), which meant a
backfill run could sit
+ RUNNING indefinitely without ever having its task instances scheduled once
the number of
+ concurrently running non-backfill DagRuns met or exceeded
`max_dagruns_per_loop_to_schedule`.
+ """
+ from airflow.models.backfill import Backfill, BackfillDagRun,
ReprocessBehavior
+
+ # Cap the per-loop examine batch well below the number of ordinary running
DagRuns below, so the
+ # old behavior (backfill always sorts last) would exclude the backfill
DagRun from every batch.
+ examine_limit = 5
+ monkeypatch.setattr(DagRun, "DEFAULT_DAGRUNS_TO_EXAMINE", examine_limit)
+
+ # Plenty of ordinary running DagRuns, each with a real
last_scheduling_decision so none of them
+ # compete on the nulls-first tier with the backfill run below.
+ for i in range(examine_limit * 2):
+ with dag_maker(f"busy_dag_{i}", schedule="@daily", session=session):
+ pass
+ dr = dag_maker.create_dagrun(state=DagRunState.RUNNING)
+ dr.last_scheduling_decision = pendulum.now("UTC")
+ session.commit()
+
+ # One backfill DagRun, freshly promoted to RUNNING: no scheduling decision
has happened yet, and
+ # its BackfillDagRun.sort_ordinal is deliberately large (deep in its own
backfill's own ordering).
+ with dag_maker("starved_backfill_dag", schedule="@daily", session=session)
as dag:
+ pass
+ backfill = Backfill(
+ dag_id=dag.dag_id,
+ from_date=pendulum.parse("2021-01-01"),
+ to_date=pendulum.parse("2021-01-10"),
+ max_active_runs=10,
+ dag_run_conf={},
+ reprocess_behavior=ReprocessBehavior.NONE,
+ )
+ session.add(backfill)
+ session.flush()
+ backfill_dr = dag_maker.create_dagrun(
+ run_id="backfill__2021-01-10T00:00:00+00:00",
+ run_type=DagRunType.BACKFILL_JOB,
+ state=DagRunState.RUNNING,
+ backfill_id=backfill.id,
+ )
+ backfill_dr.last_scheduling_decision = None
+ session.add(
+ BackfillDagRun(
+ backfill_id=backfill.id,
+ dag_run_id=backfill_dr.id,
+ logical_date=backfill_dr.logical_date,
+ sort_ordinal=999,
+ )
+ )
+ session.commit()
+
+ examined_dag_ids = {
+ r.dag_id for r in
DagRun.get_running_dag_runs_to_examine(session=session,
eagerly_load_dag_tags=False)
+ }
+ assert "starved_backfill_dag" in examined_dag_ids
+
+
class TestClearPartitionRuns:
"""Direct unit tests for the clear_partition_runs model-layer function."""
diff --git a/docs/spelling_wordlist.txt b/docs/spelling_wordlist.txt
index 74bdc1864ed..bfc7fc8a43b 100644
--- a/docs/spelling_wordlist.txt
+++ b/docs/spelling_wordlist.txt
@@ -474,6 +474,7 @@ dependant
DependencyMixin
Deprecations
deprecations
+deprioritize
deps
deques
deregister
