avolant commented on code in PR #70307:
URL: https://github.com/apache/airflow/pull/70307#discussion_r4315117295
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -968,21 +980,37 @@ def _select_task_instances_to_queue(
pool_stats["open"] = open_slots
- is_done = executable_tis or len(task_instances_to_examine) <
max_tis
+ if len(executable_tis) >= max_tis:
+ break
+
# Check this to avoid accidental infinite loops
found_new_filters = (
len(starved_pools) > num_starved_pools
- or len(starved_dags) > num_starved_dags
+ or len(starved_dag_runs) > num_starved_dag_runs
or len(starved_tasks) > num_starved_tasks
or len(starved_tasks_task_dagrun_concurrency) >
num_starved_tasks_task_dagrun_concurrency
)
- if is_done or not found_new_filters:
+ if len(executable_tis) >= max_tis or not found_new_filters:
break
+ if executable_tis:
Review Comment:
Good catch, thanks. Added
`test_select_task_instances_to_queue_does_not_cap_queries_before_first_selection`
in 5c083773ed, using the scenario you described.
`MAX_TI_REFILL_QUERIES_PER_LOOP` + 2 tasks with `max_active_tis_per_dag=1` are
running in one run and scheduled in the next, plus one lower-priority runnable
task. With `max_tis=1`, each query returns one row, so five blocked queries
come before the runnable one. The test expects only runnable to be queued. The
blocked count is tied to the constant, so the test still means something if the
cap changes.
I changed if `executable_tis`: to `if True: locally` to check it. The new
test fails with an empty batch, and the rest of `test_scheduler_job.py` still
passes.
--
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]
