ignacioparicio opened a new pull request, #70942:
URL: https://github.com/apache/airflow/pull/70942
`BaseExecutor` sorts its queued tasks by `priority_weight` ascending, then
dispatches from the front of that list. So when more tasks are queued than
there are open slots, the **lowest priority tasks are the ones sent to
workers**, and the highest priority ones are held back.
For example, with weights 1, 5 and 10 and one slot freeing up per loop, they
run in the order 1, 5, 10; the most important task runs last.
It also does not even out over time, because the queue is re-sorted on every
call.
<details>
<summary> Where it happens </summary>
In `airflow-core/src/airflow/executors/base_executor.py`:
```python
# order_queued_tasks_by_priority(): lowest weight first
return sorted(self.queued_tasks.items(), key=lambda x:
x[1].ti.priority_weight, reverse=False)
# _get_workloads_to_schedule(): reads from the front, stops at open_slots
for task_key, task_workload in self.order_queued_tasks_by_priority():
if len(workloads_to_schedule) >= open_slots:
break
workloads_to_schedule.append((task_key, task_workload))
```
</details>
<details>
<summary> How it got here </summary>
The sort and the way it is consumed used to match. They drifted apart in two
steps:
- #61376 (`fd4fd67c0f`) switched the sort from descending to ascending and,
in the same commit, changed `sorted_queue.pop(0)` to `sorted_queue.pop()`. An
ascending list popped from the back still gives the highest priority first, so
this was correct, and O(1) instead of O(n).
- #61153 (`9fa13c82ff`) refactored `trigger_tasks` to support callback
workloads and replaced that `pop()` with the front-to-back loop above. The sort
was left ascending.
</details>
This only affects an executor that has more queued tasks than open slots, so
deployments that stay under their `parallelism` ceiling, or that never set
`priority_weight`, see no difference either way.
What this PR does:
- `base_executor.py`: sorts descending again, and notes on the method that
consumers read from the front.
- `test_base_executor.py`: adds a test with three weights and fewer open
slots than tasks. The existing tests all run with more slots than tasks, so
none of them reach the truncating branch.
- Adds a bugfix newsfragment.
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes, Claude Code (Opus 5)
Used to trace the ordering back through the history of the two commits
above, and to check that nothing else in the repo overrides the sort or depends
on it being ascending.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
