You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by ka...@apache.org on 2025/09/15 19:15:42 UTC
(airflow) 10/13: Fix handling of `priority_weight` for Dag Processor callbacks (#55436)
This is an automated email from the ASF dual-hosted git repository.
kaxilnaik pushed a commit to branch v3-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 114dba649cf484e3db8e475e01e78028ec38979a
Author: Ramit Kataria <ra...@amazon.com>
AuthorDate: Mon Sep 15 09:47:36 2025 -0700
Fix handling of `priority_weight` for Dag Processor callbacks (#55436)
This reverses the order in the DB query to fetch callbacks for Dag
Processor from ascending to descending. As far as I can tell, currently
the `priority_weight` for Dag Processor is always 10. See:
https://github.com/apache/airflow/blob/811d1e91c5a387a208eb3570b43363ecc137f4ba/airflow-core/src/airflow/callbacks/database_callback_sink.py#L38
So this is not going to have a change in user experience but if we
wanted to allow the user to specify it in the future, this would cause
confusion because it was prioritizing lower weights over higher weights,
which is the opposite of how it works for tasks.
* Change default priority_weight from 10 to 1
(cherry picked from commit 6dd02f9533fba4f7b63a1b20832821f0138dc2b9)
---
airflow-core/src/airflow/callbacks/database_callback_sink.py | 2 +-
airflow-core/src/airflow/dag_processing/manager.py | 4 +++-
airflow-core/tests/unit/dag_processing/test_manager.py | 9 ++++++++-
3 files changed, 12 insertions(+), 3 deletions(-)
diff --git a/airflow-core/src/airflow/callbacks/database_callback_sink.py b/airflow-core/src/airflow/callbacks/database_callback_sink.py
index 390f7620022..e7e27cf86b5 100644
--- a/airflow-core/src/airflow/callbacks/database_callback_sink.py
+++ b/airflow-core/src/airflow/callbacks/database_callback_sink.py
@@ -35,5 +35,5 @@ class DatabaseCallbackSink(BaseCallbackSink):
@provide_session
def send(self, callback: CallbackRequest, session: Session = NEW_SESSION) -> None:
"""Send callback for execution."""
- db_callback = DbCallbackRequest(callback=callback, priority_weight=10)
+ db_callback = DbCallbackRequest(callback=callback, priority_weight=1)
session.add(db_callback)
diff --git a/airflow-core/src/airflow/dag_processing/manager.py b/airflow-core/src/airflow/dag_processing/manager.py
index cf21c28f85e..725e354610e 100644
--- a/airflow-core/src/airflow/dag_processing/manager.py
+++ b/airflow-core/src/airflow/dag_processing/manager.py
@@ -451,7 +451,9 @@ class DagFileProcessorManager(LoggingMixin):
callback_queue: list[CallbackRequest] = []
with prohibit_commit(session) as guard:
query = select(DbCallbackRequest)
- query = query.order_by(DbCallbackRequest.priority_weight.asc()).limit(self.max_callbacks_per_loop)
+ query = query.order_by(DbCallbackRequest.priority_weight.desc()).limit(
+ self.max_callbacks_per_loop
+ )
query = with_row_locks(query, of=DbCallbackRequest, session=session, skip_locked=True)
callbacks = session.scalars(query)
for callback in callbacks:
diff --git a/airflow-core/tests/unit/dag_processing/test_manager.py b/airflow-core/tests/unit/dag_processing/test_manager.py
index 6b46a3dc0b9..85bb362785d 100644
--- a/airflow-core/tests/unit/dag_processing/test_manager.py
+++ b/airflow-core/tests/unit/dag_processing/test_manager.py
@@ -740,6 +740,8 @@ class TestDagFileProcessorManager:
@conf_vars({("core", "load_examples"): "False"})
def test_fetch_callbacks_from_database(self, configure_testing_dag_bundle):
+ """Test _fetch_callbacks returns callbacks ordered by priority_weight desc."""
+
dag_filepath = TEST_DAG_FOLDER / "test_on_failure_callback_dag.py"
callback1 = DagCallbackRequest(
@@ -767,7 +769,12 @@ class TestDagFileProcessorManager:
manager = DagFileProcessorManager(max_runs=1)
with create_session() as session:
- manager.run()
+ callbacks = manager._fetch_callbacks(session=session)
+
+ # Should return callbacks ordered by priority_weight desc (highest first)
+ assert callbacks[0].run_id == "123"
+ assert callbacks[1].run_id == "456"
+
assert session.query(DbCallbackRequest).count() == 0
@conf_vars(