Fix Tasks getting stuck in scheduled state

[tanelk]

closes #19622

The scheduler_job can get stuck in a state, where it is not able to queue new tasks. It will get out of this state on its own, but the time taken depends on the runtime of current tasks - this could be several hours or even days.

If the scheduler can't queue any tasks because of different concurrency limits (per pool, dag or task), then on next iterations of the scheduler loop it will try to queue the same tasks. Meanwhile there could be some scheduled tasks with lower priority_weight that could be queued, but they will remain waiting.

The proposed solution is to keep track of dag and task ids, that are concurrecy limited and then repeat the query with these dags and tasks filtered out.

---

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code change, Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in UPDATING.md.

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst)
Here are some useful points:

• Pay attention to the quality of your code (flake8, mypy and type annotations). Our pre-commits will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it’s a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[tanelk]

unrelated, but corrected a variable name, that caused some confusion

[ashb]

This offset approach won't work well with multiple schedulers running

[vapiravfif]

Apologies if this comment won't help the discussion, but in my opinion the better approach could may be something being done around the "limit" clause -
query = query.limit(max_tis) (origin)
and
max_tis = min(self.max_tis_per_query, self.executor.slots_available) (origin)

Because it seems that the assumption that it is a better to limit query of tasks waiting to be queued by potentially available slots might not be the best practice in some cases.

[tanelk]

Apologies if this comment won't help the discussion, but in my opinion the better approach could may be something being done around the "limit" clause - query = query.limit(max_tis) (origin) and max_tis = min(self.max_tis_per_query, self.executor.slots_available) (origin)

Because it seems that the assumption that it is a better to limit query of tasks waiting to be queued by potentially available slots might not be the best practice in some cases.

Yes reducing the value of max_tis will make this situation worse, but removing the max_tis = min(...) does not guarantee that the issue will get solved.

As long as we are using any limit value, there is a chance that none of the selected task instancess can be queued but there are some task instances beyond the limit that could have been scheduled.

I can think of 3 possible solutions, that will never cause the scheduler to get stuck:

1. Remove the limit - could risk memory issues on a large airflow installation (many dags and tasks)
2. Build all the concurrency filters into the SQL query - the task instance limit seems to be impossible/very difficult in the current data model.
3. Some sort of iterative approach to look at the task instances further down - this is proposed in this PR.

[tanelk]

This offset approach won't work well with multiple schedulers running

@ashb I now reworked the method

[ashb]

A lot more work this way, but I think this looks good. I'll need to take a look at this again with fresher eyes next week.

How much/where have you tested this change?

[kaxil]

Just rebased on latest main branch to fix merge conflicts

[ephraimbuddy]

@tanelk please rebase. I guess the failing tests have been resolved in main

[jedcunningham]

We definitely need another committer to take a look at this as well.

[github-actions]

The PR most likely needs to run full matrix of tests because it modifies parts of the core of Airflow. However, committers might decide to merge it quickly and take the risk. If they don't merge it quickly - please rebase it to the latest main at your convenience, or amend the last commit of the PR, and push it with --force-with-lease.

[boring-cyborg]

Awesome work, congrats on your first merged pull request!

[jedcunningham]

Thanks @tanelk! Congrats on your first commit 🎉