<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.2.3 (latest released)</p>
<h3 dir="auto">What happened</h3>
<h2 dir="auto">Setup</h2>
<p dir="auto">Have 3 DAGs:</p>
<ol dir="auto">
<li>~240 Tasks, executes every hour, runs for about 45-50 minutes total</li>
<li>~600 Tasks, executes every 5 days, runs for days</li>
<li>~10 Tasks, executes on trigger, runs for 10-50 minutes</li>
</ol>
<p dir="auto">DAGs run on the <code class="notranslate">default_pool</code> with <code class="notranslate">max_active_tasks</code> set to 2.</p>
<p dir="auto">My AirFlow config file has the following:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="parallelism = 32
default_pool_task_slot_count = 6
executor = LocalExecutor
default_task_weight_rule = absolute"><pre class="notranslate"><code class="notranslate">parallelism = 32
default_pool_task_slot_count = 6
executor = LocalExecutor
default_task_weight_rule = absolute
</code></pre></div>
<p dir="auto"><code class="notranslate">parallelism</code> was set initially like that.<br>
<code class="notranslate">default_pool_task_slot_count</code> was set to 6, because it seemed rational that if all of my 3 dags are executing at the same time, the maximum amount of tasks that can be executed is 3*2=6.</p>
<h2 dir="auto">The problem:</h2>
<p dir="auto">Almost every time any one of the DAGs is executed, no tasks from other DAGs will start until all the tasks from the first one finish. That is, if slow DAG_2 with 600 tasks starts, DAG_1 will have to wait for days to start even a single Task.</p>
<p dir="auto">The Logs for the Scheduler look like this:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="scheduler_1  | 
scheduler_1  | [2022-02-09 10:02:27,116] {scheduler_job.py:288} INFO - 4 tasks up for execution:
scheduler_1  |  &lt;TaskInstance: Slow_DAG.files_processing.group_7.command_0 scheduled__2022-02-04T09:57:13.142966+00:00 [scheduled]&gt;
scheduler_1  |  &lt;TaskInstance: Slow_DAG.files_processing.group_119.command_0 scheduled__2022-02-04T09:57:13.142966+00:00 [scheduled]&gt;
scheduler_1  |  &lt;TaskInstance: Slow_DAG.files_processing.group_118.command_0 scheduled__2022-02-04T09:57:13.142966+00:00 [scheduled]&gt;
scheduler_1  |  &lt;TaskInstance: Slow_DAG.files_processing.group_117.command_0 scheduled__2022-02-04T09:57:13.142966+00:00 [scheduled]&gt;
scheduler_1  | [2022-02-09 10:02:27,119] {scheduler_job.py:322} INFO - Figuring out tasks to run in Pool(name=default_pool) with 4 open slots and 4 task instances ready to be queued
scheduler_1  | [2022-02-09 10:02:27,119] {scheduler_job.py:349} INFO - DAG Slow_DAG has 2/2 running and queued tasks
scheduler_1  | [2022-02-09 10:02:27,119] {scheduler_job.py:357} INFO - Not executing &lt;TaskInstance: Slow_DAG.files_processing.group_7.command_0 scheduled__2022-02-04T09:57:13.142966+00:00 [scheduled]&gt; since the number of tasks running or queued from DAG Slow_DAG is &gt;= to the DAG's max_active_tasks limit of 2
scheduler_1  | [2022-02-09 10:02:27,119] {scheduler_job.py:349} INFO - DAG Slow_DAG has 2/2 running and queued tasks
scheduler_1  | [2022-02-09 10:02:27,119] {scheduler_job.py:357} INFO - Not executing &lt;TaskInstance: Slow_DAG.files_processing.group_119.command_0 scheduled__2022-02-04T09:57:13.142966+00:00 [scheduled]&gt; since the number of tasks running or queued from DAG Slow_DAG is &gt;= to the DAG's max_active_tasks limit of 2
scheduler_1  | [2022-02-09 10:02:27,119] {scheduler_job.py:349} INFO - DAG Slow_DAG has 2/2 running and queued tasks
scheduler_1  | [2022-02-09 10:02:27,120] {scheduler_job.py:357} INFO - Not executing &lt;TaskInstance: Slow_DAG.files_processing.group_118.command_0 scheduled__2022-02-04T09:57:13.142966+00:00 [scheduled]&gt; since the number of tasks running or queued from DAG Slow_DAG is &gt;= to the DAG's max_active_tasks limit of 2
scheduler_1  | [2022-02-09 10:02:27,120] {scheduler_job.py:349} INFO - DAG Slow_DAG has 2/2 running and queued tasks
scheduler_1  | [2022-02-09 10:02:27,120] {scheduler_job.py:357} INFO - Not executing &lt;TaskInstance: Slow_DAG.files_processing.group_117.command_0 scheduled__2022-02-04T09:57:13.142966+00:00 [scheduled]&gt; since the number of tasks running or queued from DAG Slow_DAG is &gt;= to the DAG's max_active_tasks limit of 2
scheduler_1  | [2022-02-09 10:02:27,120] {scheduler_job.py:410} INFO - Setting the following tasks to queued state:
scheduler_1  | 
"><pre class="notranslate"><code class="notranslate">scheduler_1  | 
scheduler_1  | [2022-02-09 10:02:27,116] {scheduler_job.py:288} INFO - 4 tasks up for execution:
scheduler_1  |  &lt;TaskInstance: Slow_DAG.files_processing.group_7.command_0 scheduled__2022-02-04T09:57:13.142966+00:00 [scheduled]&gt;
scheduler_1  |  &lt;TaskInstance: Slow_DAG.files_processing.group_119.command_0 scheduled__2022-02-04T09:57:13.142966+00:00 [scheduled]&gt;
scheduler_1  |  &lt;TaskInstance: Slow_DAG.files_processing.group_118.command_0 scheduled__2022-02-04T09:57:13.142966+00:00 [scheduled]&gt;
scheduler_1  |  &lt;TaskInstance: Slow_DAG.files_processing.group_117.command_0 scheduled__2022-02-04T09:57:13.142966+00:00 [scheduled]&gt;
scheduler_1  | [2022-02-09 10:02:27,119] {scheduler_job.py:322} INFO - Figuring out tasks to run in Pool(name=default_pool) with 4 open slots and 4 task instances ready to be queued
scheduler_1  | [2022-02-09 10:02:27,119] {scheduler_job.py:349} INFO - DAG Slow_DAG has 2/2 running and queued tasks
scheduler_1  | [2022-02-09 10:02:27,119] {scheduler_job.py:357} INFO - Not executing &lt;TaskInstance: Slow_DAG.files_processing.group_7.command_0 scheduled__2022-02-04T09:57:13.142966+00:00 [scheduled]&gt; since the number of tasks running or queued from DAG Slow_DAG is &gt;= to the DAG's max_active_tasks limit of 2
scheduler_1  | [2022-02-09 10:02:27,119] {scheduler_job.py:349} INFO - DAG Slow_DAG has 2/2 running and queued tasks
scheduler_1  | [2022-02-09 10:02:27,119] {scheduler_job.py:357} INFO - Not executing &lt;TaskInstance: Slow_DAG.files_processing.group_119.command_0 scheduled__2022-02-04T09:57:13.142966+00:00 [scheduled]&gt; since the number of tasks running or queued from DAG Slow_DAG is &gt;= to the DAG's max_active_tasks limit of 2
scheduler_1  | [2022-02-09 10:02:27,119] {scheduler_job.py:349} INFO - DAG Slow_DAG has 2/2 running and queued tasks
scheduler_1  | [2022-02-09 10:02:27,120] {scheduler_job.py:357} INFO - Not executing &lt;TaskInstance: Slow_DAG.files_processing.group_118.command_0 scheduled__2022-02-04T09:57:13.142966+00:00 [scheduled]&gt; since the number of tasks running or queued from DAG Slow_DAG is &gt;= to the DAG's max_active_tasks limit of 2
scheduler_1  | [2022-02-09 10:02:27,120] {scheduler_job.py:349} INFO - DAG Slow_DAG has 2/2 running and queued tasks
scheduler_1  | [2022-02-09 10:02:27,120] {scheduler_job.py:357} INFO - Not executing &lt;TaskInstance: Slow_DAG.files_processing.group_117.command_0 scheduled__2022-02-04T09:57:13.142966+00:00 [scheduled]&gt; since the number of tasks running or queued from DAG Slow_DAG is &gt;= to the DAG's max_active_tasks limit of 2
scheduler_1  | [2022-02-09 10:02:27,120] {scheduler_job.py:410} INFO - Setting the following tasks to queued state:
scheduler_1  | 

</code></pre></div>
<p dir="auto">I have fixed this by setting</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="parallelism = 1000
default_pool_task_slot_count = 999"><pre class="notranslate"><code class="notranslate">parallelism = 1000
default_pool_task_slot_count = 999
</code></pre></div>
<h2 dir="auto">The reason why this occurs and why the solution works</h2>
<p dir="auto">We need to look at this method of the Scheduler:<br>
<a href="https://github.com/apache/airflow/blob/2.2.3/airflow/jobs/scheduler_job.py#L229">https://github.com/apache/airflow/blob/2.2.3/airflow/jobs/scheduler_job.py#L229</a><br>
In it, the overall logic is the following:</p>
<ol dir="auto">
<li>Receive <code class="notranslate">max_tis</code>(== <code class="notranslate">parallelism - active running tasks</code> from the config) in the arguments. In my case that would be 30</li>
<li>Calculate how many free slots in <strong>all</strong> all Pools we have. In my case that would be 4.</li>
<li>Select the minimum and update the <code class="notranslate">max_tis</code> variable: <code class="notranslate">max_tis = min(4, 30) = 4</code></li>
<li>Query the DB for tasks that are Scheduled in an unpaused DAGs that are running normally and order them by DAG execution date.</li>
<li>Limit the query by <code class="notranslate">max_tis</code>.</li>
<li>Loop over each returned task and check if we can run it. Run, if possible.</li>
</ol>
<p dir="auto">Now, since the tasks are ordered by <code class="notranslate">execution date</code>, this will return us tasks for a DAG that started first, which lets say was DAG_2.<br>
It has over 600 tasks. The LIMIT operation will return 4 of them.<br>
Each off these tasks cannot be run since there are already 2 tasks running and the <code class="notranslate">max_active_tasks</code> of the DAG is 2.<br>
Thus, we just wait till one of these tasks finish and start another task from the same DAG.</p>
<h3 dir="auto">What you expected to happen</h3>
<p dir="auto">All DAGs are executed in parallel, up to 6 tasks in total.</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">Take a look at <a href="https://github.com/EvGe22/airflow_bug_reproduce">https://github.com/EvGe22/airflow_bug_reproduce</a></p>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Ubuntu 18.04.5 LTS</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Docker-Compose</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Anything else</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Are you willing to submit PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>
<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.3.3 (latest released)</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">As the title states, if you have dynamically mapped tasks inside of a <code class="notranslate">TaskGroup</code>, those tasks do not get the <code class="notranslate">group_id</code> prepended to their respective <code class="notranslate">task_id</code>s.  This causes at least a couple of undesirable side effects:</p>
<ol dir="auto">
<li>Task names are truncated in Grid/Graph* View.  The tasks below are named <code class="notranslate">plus_one</code> and <code class="notranslate">plus_two</code>:</li>
</ol>
<p dir="auto"><a target="_blank" rel="noopener noreferrer nofollow" href="https://user-images.githubusercontent.com/7269927/179826453-a4293c14-2a83-4739-acf2-8b378e4e85e9.png"><img src="https://user-images.githubusercontent.com/7269927/179826453-a4293c14-2a83-4739-acf2-8b378e4e85e9.png" alt="Screenshot from 2022-07-19 13-29-05" style="max-width: 100%;"></a><br>
<a target="_blank" rel="noopener noreferrer nofollow" href="https://user-images.githubusercontent.com/7269927/179826442-b9e3d24d-52ff-49fc-a8cc-fe1cb5143bcb.png"><img src="https://user-images.githubusercontent.com/7269927/179826442-b9e3d24d-52ff-49fc-a8cc-fe1cb5143bcb.png" alt="Screenshot from 2022-07-19 13-47-47" style="max-width: 100%;"></a></p>
<p dir="auto">Presumably this is because the UI normally strips off the <code class="notranslate">group_id</code> prefix.</p>
<p dir="auto">* Graph View was very inconsistent in my experience.  Sometimes the names are truncated, and sometimes they render correctly.  I haven't figured out the pattern behind this behavior.</p>
<ol start="2" dir="auto">
<li>Duplicate <code class="notranslate">task_id</code>s between groups result in a <code class="notranslate">airflow.exceptions.DuplicateTaskIdFound</code>, even if the <code class="notranslate">group_id</code> would normally disambiguate them.</li>
</ol>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">These dynamic tasks inside of a group should have the <code class="notranslate">group_id</code> prepended for consistent behavior.</p>
<h3 dir="auto">How to reproduce</h3>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="#!/usr/bin/env python3
import datetime

from airflow.decorators import dag, task
from airflow.utils.task_group import TaskGroup


@dag(
    start_date=datetime.datetime(2022, 7, 19),
    schedule_interval=None,
)
def test_dag():
    with TaskGroup(group_id='group'):
        @task
        def plus_one(x: int):
            return x + 1

        plus_one.expand(x=[1, 2, 3])

    with TaskGroup(group_id='ggg'):
        @task
        def plus_two(x: int):
            return x + 2

        plus_two.expand(x=[1, 2, 3])


dag = test_dag()


if __name__ == '__main__':
    dag.cli()"><pre class="notranslate"><code class="notranslate">#!/usr/bin/env python3
import datetime

from airflow.decorators import dag, task
from airflow.utils.task_group import TaskGroup


@dag(
    start_date=datetime.datetime(2022, 7, 19),
    schedule_interval=None,
)
def test_dag():
    with TaskGroup(group_id='group'):
        @task
        def plus_one(x: int):
            return x + 1

        plus_one.expand(x=[1, 2, 3])

    with TaskGroup(group_id='ggg'):
        @task
        def plus_two(x: int):
            return x + 2

        plus_two.expand(x=[1, 2, 3])


dag = test_dag()


if __name__ == '__main__':
    dag.cli()
</code></pre></div>
<h3 dir="auto">Operating System</h3>
<p dir="auto">CentOS Stream 8</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto">N/A</p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Other</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto">Standalone</p>
<h3 dir="auto">Anything else</h3>
<p dir="auto">Possibly related: <a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="741611909" data-permission-text="Title is private" data-url="https://github.com/apache/airflow/issues/12309" data-hovercard-type="issue" data-hovercard-url="/apache/airflow/issues/12309/hovercard" href="https://github.com/apache/airflow/issues/12309">#12309</a></p>
<h3 dir="auto">Are you willing to submit PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>