Enhance max_active_runs log to include blocking Run IDs and Task IDs #72387

Description

Currently, when a DAG reaches its max_active_runs limit, the scheduler logs the following generic message:

DAG <dag_id> is at (or above) max_active_runs (<X> of <Y>), not creating any more runs

While this indicates that the limit has been reached, it does not provide any context about what is occupying those active slots. To figure out why the DAG is stuck, users currently have to manually navigate through the Airflow UI or web server logs.

This contribution enhances the log message in scheduler_job_runner.py. When the max_active_runs limit is hit, the scheduler will now perform targeted queries to fetch the specific run_ids and task_ids (along with their current states) that are holding the active slots.

DAG test_dag is at (or above) max_active_runs (1 of 1), not creating any more runs. Active Runs: scheduled__2023-10-18T09:55:00+00:00 | Active Tasks: [sleep_task in scheduled__2023-10-18T09:55:00+00:00 (running)]

Implementation Note:
To avoid unnecessary database load, the queries to fetch the active runs and tasks are placed strictly inside the if total_active_runs >= dag.max_active_runs: condition. This ensures there is zero performance overhead during normal, healthy scheduling loops.

Use case/motivation

By printing the exact Run IDs and Task IDs directly into the scheduler logs, platform administrators and data engineers can instantly identify the bottleneck without leaving their logging environment. This drastically reduces Time To Resolution for stuck DAGs.

Related issues

No response

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct