xBis7 commented on PR #54284:
URL: https://github.com/apache/airflow/pull/54284#issuecomment-3172437703
After some more testing, if we remove the `last_queueing_decision` from the
`order_by` and let the query work exactly as before, then this won't break
priorities and we can sort the results with python.
We can do something like this
```diff
diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
index 34aec98fee..23734e3dac 100644
--- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
+++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
@@ -353,7 +356,6 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
.where(DM.bundle_name.is_not(None))
.options(selectinload(TI.dag_model))
.order_by(
- nulls_first(TI.last_queueing_decision.desc(),
session=session),
-TI.priority_weight,
DR.logical_date,
TI.map_index,
@@ -394,7 +398,14 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
assert executor.name
executor_slots_available[executor.name] =
executor.slots_available
+ task_instances_to_examine = sorted(
+ task_instances_to_examine,
+ key=lambda t: (t.last_queueing_decision is not None,
t.last_queueing_decision),
+ reverse=True # desc.
+ )
+
for task_instance in task_instances_to_examine:
+ task_instance.last_queueing_decision = timezone.utcnow()
pool_name = task_instance.pool
pool_stats = pools.get(pool_name)
if not pool_stats:
@@ -543,9 +554,367 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
Stats.gauge("scheduler.tasks.executable", len(executable_tis))
- non_scheduled_tis = [x for x in task_instances_to_examine if x not
in executable_tis]
- for ti in non_scheduled_tis:
- ti.last_queueing_decision = timezone.utcnow()
+ # non_scheduled_tis = [x for x in task_instances_to_examine if x
not in executable_tis]
+ # for ti in non_scheduled_tis:
+ # ti.last_queueing_decision = timezone.utcnow()
```
Unfortunately, this still doesn't give us the desired behavior because we
will end up constantly queueing the tasks from the bottom of the list and
eventually the ones on the top will starve.
Here is an idea, we can restore the logic of keeping track of starved tasks
and check that if a task has been starving and it's in the examining list, then
we will do the sorting.
--
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]
