This is an automated email from the ASF dual-hosted git repository.
pierrejeambrun 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 316dc6d82e7 Dispatch the highest priority tasks first in the executor
(#70942)
316dc6d82e7 is described below
commit 316dc6d82e7809b096028ee77bc3bdc42538aaff
Author: Ignacio Paricio <[email protected]>
AuthorDate: Mon Aug 17 15:28:40 2026 +0200
Dispatch the highest priority tasks first in the executor (#70942)
---
.../src/airflow/executors/base_executor.py | 7 +++++--
.../tests/unit/executors/test_base_executor.py | 23 ++++++++++++++++++++++
2 files changed, 28 insertions(+), 2 deletions(-)
diff --git a/airflow-core/src/airflow/executors/base_executor.py
b/airflow-core/src/airflow/executors/base_executor.py
index 8030e9c1445..ef5fafded8b 100644
--- a/airflow-core/src/airflow/executors/base_executor.py
+++ b/airflow-core/src/airflow/executors/base_executor.py
@@ -428,7 +428,10 @@ class BaseExecutor(LoggingMixin):
def order_queued_tasks_by_priority(self) -> list[tuple[TaskInstanceKey,
workloads.ExecuteTask]]:
"""
- Orders the queued tasks by priority.
+ Orders the queued tasks by priority, highest ``priority_weight`` first.
+
+ Consumers take workloads from the front of this list, so the highest
priority
+ tasks must come first for them to be scheduled before the rest.
:return: List of workloads from the queued_tasks according to the
priority.
"""
@@ -439,7 +442,7 @@ class BaseExecutor(LoggingMixin):
return sorted(
self.queued_tasks.items(),
key=lambda x: x[1].ti.priority_weight,
- reverse=False,
+ reverse=True,
)
def trigger_tasks(self, open_slots: int) -> None:
diff --git a/airflow-core/tests/unit/executors/test_base_executor.py
b/airflow-core/tests/unit/executors/test_base_executor.py
index 8858eac59f3..1d8a2dfa329 100644
--- a/airflow-core/tests/unit/executors/test_base_executor.py
+++ b/airflow-core/tests/unit/executors/test_base_executor.py
@@ -360,6 +360,29 @@ def test_trigger_running_tasks(dag_maker):
executor._process_workloads.assert_called_once()
[email protected]_test
+def test_trigger_tasks_schedules_highest_priority_first(dag_maker):
+ """When there are fewer open slots than queued tasks, the lowest priority
ones wait."""
+ date = timezone.utcnow()
+
+ with dag_maker("test_trigger_tasks_priority_order"):
+ BaseOperator(task_id="low", priority_weight=1)
+ BaseOperator(task_id="medium", priority_weight=5)
+ BaseOperator(task_id="high", priority_weight=10)
+
+ dagrun = dag_maker.create_dagrun(logical_date=date)
+
+ executor = BaseExecutor()
+ executor._process_workloads = mock.Mock(spec=lambda workloads: None)
+ for task_instance in dagrun.task_instances:
+ executor.queued_tasks[task_instance.key] =
workloads.ExecuteTask.make(task_instance)
+
+ executor.trigger_tasks(open_slots=2)
+
+ scheduled = [workload.ti.task_id for workload in
executor._process_workloads.call_args[0][0]]
+ assert scheduled == ["high", "medium"]
+
+
def test_debug_dump(caplog):
executor = BaseExecutor()
with caplog.at_level(logging.INFO):
