This is an automated email from the ASF dual-hosted git repository.
kaxil 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 d5c31e94892 Stop the scheduler creating Dag runs for paused
partitioned-asset Dags (#73151)
d5c31e94892 is described below
commit d5c31e94892c3acc998a5537d2edcf7288d1cd5b
Author: Kaxil Naik <[email protected]>
AuthorDate: Tue Sep 15 16:02:22 2026 +0100
Stop the scheduler creating Dag runs for paused partitioned-asset Dags
(#73151)
The partitioned-asset creation loop was the only scheduler path that did not
check is_paused. Every other path goes through
DagModel.dags_needing_dagruns,
which has filtered it since the method existed.
The sharpest way in is a completed drain. _finalize_draining_dags converges
a
draining Dag to paused in one step, clearing is_draining and setting
is_paused
together, so every pending AssetPartitionDagRun that the draining filter had
been holding became eligible the instant the drain finished -- and the loop
created a run for exactly the Dag the operator had just drained.
A plain pause reaches it through a narrower window, since AssetManager skips
inactive Dags when recording partition keys and no APDR can advance while
the
Dag is paused. What remains is an APDR already satisfiable at the moment of
pausing: the pause lands between the satisfying asset event and the next
scheduler tick, and the run is created anyway. That includes the automatic
pause after max_consecutive_failed_dag_runs, which the user never initiated.
Either way the run cannot start -- get_queued_dag_runs_to_set_running
filters
on is_paused -- so it sits in QUEUED, listed in the UI and the API but
unable
to progress, until someone reactivates the Dag, at which point it starts
alongside everything else.
The filter goes in the WHERE clause rather than a post-query skip so a
paused
Dag's APDRs do not consume the per-tick FIFO budget ahead of Dags that can
actually run.
---
.../src/airflow/jobs/scheduler_job_runner.py | 17 +++++
airflow-core/tests/unit/jobs/test_scheduler_job.py | 81 ++++++++++++++++++++--
2 files changed, 94 insertions(+), 4 deletions(-)
diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
index 9be6eaec550..fe1e09a9ee5 100644
--- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
+++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
@@ -2300,6 +2300,22 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
asset, firing on stale history would conflict with the declared
topology,
so the APDR waits. Reactivating the asset resumes evaluation
automatically.
This matches the UI's progress view (``_fetch_active_assets_per_dag``).
+
+ Pausing and draining freeze pending APDRs, mirroring the ``is_paused``
/
+ ``is_draining`` half of
:meth:`~airflow.models.dag.DagModel.dags_needing_dagruns`.
+ ``has_import_errors`` needs no predicate of its own: it is only ever
set together
+ with ``is_stale`` (``_update_import_errors``) and cleared together
with it
+ (``DagModelOperation.update_dags``), so the ``is_stale`` filter above
already
+ excludes those Dags. ``exceeds_max_non_backfill`` is the one genuine
divergence --
+ an APDR for a Dag already at ``max_active_runs`` still creates its
run, which then
+ waits at the QUEUED->RUNNING gate rather than being held back here.
+
+ Nothing accrues while a Dag is inactive --
``AssetManager.register_asset_change``
+ drops paused and draining Dags before any ``PartitionedAssetKeyLog``
row is
+ written -- so an event produced during the pause is never recorded and
a partially
+ satisfied APDR cannot advance past it. On reactivation the APDR
resumes from the
+ keys logged before it went inactive, unless the rollup definition
changed in the
+ meantime, in which case the stale-fingerprint cleanup below drops it
instead.
"""
# Cap per-tick work so the scheduler transaction stays bounded and
other
# scheduling work isn't starved. Remaining APDRs drain across
subsequent ticks.
@@ -2320,6 +2336,7 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
.join(DagModel, DagModel.dag_id ==
AssetPartitionDagRun.target_dag_id)
.where(
AssetPartitionDagRun.created_dag_run_id.is_(None),
+ DagModel.is_paused.is_(False),
DagModel.is_draining.is_(False),
DagModel.is_stale.is_(False),
)
diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py
b/airflow-core/tests/unit/jobs/test_scheduler_job.py
index de21286adf7..ca91ef5f76f 100644
--- a/airflow-core/tests/unit/jobs/test_scheduler_job.py
+++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py
@@ -11634,9 +11634,21 @@ def _produce_and_register_asset_event(
@pytest.mark.need_serialized_dag
@pytest.mark.usefixtures("clear_asset_partition_rows")
-def test_partitioned_asset_dag_run_is_not_created_while_draining(dag_maker:
DagMaker, session: Session):
[email protected](
+ "scheduling_state",
+ [DagSchedulingState.DRAINING, DagSchedulingState.PAUSED],
+)
+def test_partitioned_asset_dag_run_waits_while_not_active_and_fires_on_resume(
+ dag_maker: DagMaker, session: Session, scheduling_state: DagSchedulingState
+):
+ """
+ A pending APDR is frozen while its Dag is paused or draining, not consumed.
+
+ The run it would have created is deferred rather than dropped, so
reactivating the
+ Dag fires it on the next tick.
+ """
asset = Asset(name="asset")
- consumer_dag_id = "draining-asset-event-consumer"
+ consumer_dag_id = "inactive-asset-event-consumer"
with dag_maker(
dag_id=consumer_dag_id,
schedule=PartitionedAssetTimetable(assets=asset),
@@ -11645,8 +11657,12 @@ def
test_partitioned_asset_dag_run_is_not_created_while_draining(dag_maker: DagM
EmptyOperator(task_id="consumer")
session.commit()
+ # Ordering is load-bearing: the asset event must be produced while the Dag
is still
+ # active, because AssetManager.register_asset_change skips inactive Dags
entirely.
+ # Pausing first would leave no APDR at all and the assertions below would
pass
+ # vacuously.
apdr = _produce_and_register_asset_event(
- dag_id="draining-asset-event-producer",
+ dag_id="inactive-asset-event-producer",
asset=asset,
partition_key="partition",
session=session,
@@ -11654,7 +11670,7 @@ def
test_partitioned_asset_dag_run_is_not_created_while_draining(dag_maker: DagM
)
dag_model = session.get(DagModel, consumer_dag_id)
assert dag_model is not None
- dag_model.set_scheduling_state(DagSchedulingState.DRAINING)
+ dag_model.set_scheduling_state(scheduling_state)
session.commit()
runner = SchedulerJobRunner(
@@ -11666,6 +11682,63 @@ def
test_partitioned_asset_dag_run_is_not_created_while_draining(dag_maker: DagM
assert partition_dags == set()
assert apdr.created_dag_run_id is None
+ # The APDR is deferred, not dropped: a rollup that was already satisfiable
before the
+ # pause fires as soon as the Dag is active again.
+ dag_model.set_scheduling_state(DagSchedulingState.ACTIVE)
+ session.commit()
+
+ assert runner._create_dagruns_for_partitioned_asset_dags(session=session)
== {consumer_dag_id}
+ session.refresh(apdr)
+ assert apdr.created_dag_run_id is not None
+
+
[email protected]_serialized_dag
[email protected]("clear_asset_partition_rows")
+def test_partitioned_asset_dag_run_is_not_created_after_a_drain_completes(
+ dag_maker: DagMaker, session: Session
+):
+ """
+ Completing a drain must not hand the drained Dag a fresh partition-driven
run.
+
+ ``_finalize_draining_dags`` converges draining to paused in one step, so a
pending
+ APDR that the draining filter had been holding would otherwise become
eligible the
+ instant the drain finished -- against the very Dag the operator just
drained.
+ """
+ asset = Asset(name="asset")
+ consumer_dag_id = "drained-asset-event-consumer"
+ with dag_maker(
+ dag_id=consumer_dag_id,
+ schedule=PartitionedAssetTimetable(assets=asset),
+ session=session,
+ ):
+ EmptyOperator(task_id="consumer")
+ session.commit()
+
+ apdr = _produce_and_register_asset_event(
+ dag_id="drained-asset-event-producer",
+ asset=asset,
+ partition_key="partition",
+ session=session,
+ dag_maker=dag_maker,
+ )
+ dag_model = session.get(DagModel, consumer_dag_id)
+ assert dag_model is not None
+ dag_model.set_scheduling_state(DagSchedulingState.DRAINING)
+ session.commit()
+
+ runner = SchedulerJobRunner(
+ job=Job(job_type=SchedulerJobRunner.job_type),
executors=[MockExecutor(do_update=False)]
+ )
+ assert runner._create_dagruns_for_partitioned_asset_dags(session=session)
== set()
+
+ # The Dag has no unfinished runs, so the drain converges to paused on this
tick.
+ runner._finalize_draining_dags(session=session)
+ assert dag_model.scheduling_state == DagSchedulingState.PAUSED
+ session.flush()
+
+ assert runner._create_dagruns_for_partitioned_asset_dags(session=session)
== set()
+ assert apdr.created_dag_run_id is None
+
@pytest.mark.need_serialized_dag
@pytest.mark.usefixtures("clear_asset_partition_rows")
