This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new a3a1c9a5a8f [v3-3-test] Dispatch the highest priority tasks first in
the executor (#70942) (#71715)
a3a1c9a5a8f is described below
commit a3a1c9a5a8fc937626980d6769ad0410ec8e5275
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Mon Aug 17 21:55:13 2026 +0200
[v3-3-test] Dispatch the highest priority tasks first in the executor
(#70942) (#71715)
(cherry picked from commit 316dc6d82e7809b096028ee77bc3bdc42538aaff)
Co-authored-by: Ignacio Paricio
<[email protected]>
---
.../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 ef577233110..72dab61ca8f 100644
--- a/airflow-core/src/airflow/executors/base_executor.py
+++ b/airflow-core/src/airflow/executors/base_executor.py
@@ -427,7 +427,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.
"""
@@ -438,7 +441,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 cb646f7ce8d..7d4642b2d1d 100644
--- a/airflow-core/tests/unit/executors/test_base_executor.py
+++ b/airflow-core/tests/unit/executors/test_base_executor.py
@@ -352,6 +352,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):
