Don't ignore legacy concurrency dag parameter

[zbstof]

Currently, even if legacy concurrency dag parameter is specified, it is always ignored because max_active_tasks is always initialized from core.max_active_tasks_per_dag.
In our case this caused unexpected throttling of the task concurrency on production and performance issues.

concurrency parameter should always be used, if provided, as this preserves backward compatibility for users performing the migration.

Tested locally:

DAG(
    dag_id='xxx',
...
    concurrency=1,
)

Before fix:

{scheduler_job.py:413} INFO - DAG xxx has 1/16 running and queued tasks

After fix:

{scheduler_job.py:413} INFO - DAG xxx has 1/1 running and queued tasks

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.

[kaxil]

This will be included in Airflow 2.2.0 which should be released next week.

[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.

[kaxil]

@zbstof While the fix is correct the description was not, I just updated it

If you check the diff in this PR, it is not max_active_tasks that has been removed, which has not been released in an Airflow version. The description said max_active_runs_per_dag which is unrelated and affects DagRuns not number of running tasks.

[kaxil]

Check the following line for instance from 2.1.4 -- so I am still not sure why this affected your production system if you were running 2.1.4 -- Are you sure you are not running from 2.2.0b2? :

https://github.com/apache/airflow/blob/2.1.4/airflow/models/dag.py#L281

airflow/airflow/models/dag.py

Line 281 in 614858f

self._concurrency = concurrency

[zbstof]

While the fix is correct the description was not, I just updated it

Thank you, my mistake. I copied the wrong line when compiling the description

so I am still not sure why this affected your production system if you were running 2.1.4

You're completely right, this has nothing to do with 2.1 branch, edited the description. We're using v2.2.0.dev0. I didn't check where the issue was introduced and assumed it was in 2.1 branch.

Thank you for the corrections!

[kaxil]

Thank you, my mistake. I copied the wrong line when compiling the description

No worries, thanks for the PR :)

[ashb]

Retriggering CI.

[kaxil]

(rebased on main to fix issues with dependency -- fixed by #18695)

[zbstof]

Not sure if I should (could?) do something to fix Build Images / Build CI images 3.7 🤔

[kaxil]

Not sure if I should (could?) do something to fix Build Images / Build CI images 3.7 🤔

No, don't worry about that, we will take care of it.