Re: [PR] Include the max_active_tasks limit in the query fetching TIs to be queued [airflow]

["xBis7 (via GitHub)" <gi...@apache.org>]

xBis7 commented on code in PR #54103:
URL: https://github.com/apache/airflow/pull/54103#discussion_r2483790152


##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -408,10 +433,63 @@ def _executable_task_instances_to_queued(self, max_tis: int, session: Session) -
                 .where(~DM.is_paused)
                 .where(TI.state == TaskInstanceState.SCHEDULED)
                 .where(DM.bundle_name.is_not(None))
+                .join(
+                    dr_task_concurrency_subquery,
+                    TI.run_id == dr_task_concurrency_subquery.c.run_id,
+                    isouter=True,
+                )
+                .where(func.coalesce(dr_task_concurrency_subquery.c.dr_count, 0) < DM.max_active_tasks)

Review Comment:
   I did some testing and here is what I found.
   
   Let's assume that we have 3 dags with the same `run_id` and all the checks use only the `run_id`.
   
   * the query will return `max_active_tasks` from the 1st dag and ignore the other two because they have the same `run_id`
   * every consecutive iteration until the tasks from the 1st dag finish, won't pick any new tasks for examining and queueing
   * once the tasks finish, then it will pick `max_active_tasks` but it will again concentrate on 1 dag
   
   It would group all tasks from all dags as if they all belonged to the same dag_run. But the `dag_id, run_id` pair is unique and now the tasks are properly separated.
   
   This was a serious logical bug. I don't know how common it will be for multiple `dag_runs` to have the same `run_id` but it's good to have already accounted for that edge case. Thanks @kaxil, great catch! With the suggested changes it works as it should.



-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org