<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.2.2</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">I'm running two DAGs composed by 50 independent tasks in each DAG. Lets call DAG <strong>A</strong> and DAG <strong>B</strong>. DAG <strong>A</strong> has <code class="notranslate">max_active_tasks</code> set to 1 which means that only one task can run each time in this DAG.</p>
<p dir="auto">When I run only DAG <strong>B</strong> its 50 tasks get executed properly, respecting the <code class="notranslate">AIRFLOW__CORE__MAX_ACTIVE_TASKS_PER_DAG </code> parameter, which limits the number of running tasks at the same time, defaulted to 16.</p>
<p dir="auto">But when I run DAG <strong>A</strong> and <strong>B</strong> at the same time all tasks of DAG <strong>B</strong> get stucked in <code class="notranslate">scheduled</code> state. On the other hand DAG <strong>A</strong> still executing properly, running tasks one by one, respecting the <code class="notranslate">max_active_tasks</code> parameter.</p>
<p dir="auto">It seems that the <strong>max_active_tasks</strong> used together with a <strong>high number of scheduled tasks</strong> is causing a side effect on any other DAG.</p>
<p dir="auto">Only after a part of DAG <strong>A</strong> tasks gets executed Airflow starts to execute DAG <strong>B</strong> tasks.</p>
<p dir="auto">One way to overcome this bug is raising <code class="notranslate">AIRFLOW__CORE__PARALLELISM</code> variable. Nevertheless it seems a logical bug in the scheduler.</p>
<h3 dir="auto">What you expected to happen</h3>
<p dir="auto">Considering that <code class="notranslate">max_active_tasks</code> was set only in DAG <strong>A</strong>, DAG <strong>B</strong> tasks should start running anytime since they were scheduled.</p>
<h3 dir="auto">How to reproduce</h3>
<ul dir="auto">
<li>
<p dir="auto">Airflow 2.2.2 with <code class="notranslate">LocalExecutor</code> on Python <code class="notranslate">3.9</code></p>
</li>
<li>
<p dir="auto">DA <strong>A</strong></p>
</li>
</ul>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="from datetime import datetime
import time

from airflow.decorators import dag, task

TOTAL_TASKS = 50

@dag(schedule_interval=None,
    start_date=datetime(2022, 1, 15),
    default_args={'owner': 'nitai'},
    max_active_tasks=1)
def DAG_A():

   @task()
   def sleepinp_task():
       time.sleep(10)

   for _ in range(TOTAL_TASKS):
       sleepinp_task()

dag = DAG_A()"><pre class="notranslate"><code class="notranslate">from datetime import datetime
import time

from airflow.decorators import dag, task

TOTAL_TASKS = 50

@dag(schedule_interval=None,
    start_date=datetime(2022, 1, 15),
    default_args={'owner': 'nitai'},
    max_active_tasks=1)
def DAG_A():

   @task()
   def sleepinp_task():
       time.sleep(10)

   for _ in range(TOTAL_TASKS):
       sleepinp_task()

dag = DAG_A()
</code></pre></div>
<ul dir="auto">
<li>DA <strong>B</strong></li>
</ul>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="from datetime import datetime
import time

from airflow.decorators import dag, task

TOTAL_TASKS = 50

@dag(schedule_interval=None,
     start_date=datetime(2022, 1, 15),
     default_args={'owner': 'nitai'})
def DAG_B():

    @task()
    def sleepinp_task():
        time.sleep(10)

    for _ in range(TOTAL_TASKS):
        sleepinp_task()

dag = DAG_B()"><pre class="notranslate"><code class="notranslate">from datetime import datetime
import time

from airflow.decorators import dag, task

TOTAL_TASKS = 50

@dag(schedule_interval=None,
     start_date=datetime(2022, 1, 15),
     default_args={'owner': 'nitai'})
def DAG_B():

    @task()
    def sleepinp_task():
        time.sleep(10)

    for _ in range(TOTAL_TASKS):
        sleepinp_task()

dag = DAG_B()
</code></pre></div>
<p dir="auto">Start DAG <strong>A</strong> and after a few seconds start DAG <strong>B</strong>.</p>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Debian GNU/Linux 10 (buster)</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto">apache-airflow-providers-amazon==2.4.0<br>
apache-airflow-providers-celery==2.1.0<br>
apache-airflow-providers-cncf-kubernetes==2.1.0<br>
apache-airflow-providers-docker==2.3.0<br>
apache-airflow-providers-elasticsearch==2.1.0<br>
apache-airflow-providers-ftp==2.0.1<br>
apache-airflow-providers-google==6.1.0<br>
apache-airflow-providers-grpc==2.0.1<br>
apache-airflow-providers-hashicorp==2.1.1<br>
apache-airflow-providers-http==2.0.1<br>
apache-airflow-providers-imap==2.0.1<br>
apache-airflow-providers-microsoft-azure==3.3.0<br>
apache-airflow-providers-mongo==2.3.0<br>
apache-airflow-providers-mysql==2.1.1<br>
apache-airflow-providers-odbc==2.0.1<br>
apache-airflow-providers-postgres==2.3.0<br>
apache-airflow-providers-redis==2.0.1<br>
apache-airflow-providers-sendgrid==2.0.1<br>
apache-airflow-providers-sftp==2.2.0<br>
apache-airflow-providers-slack==4.1.0<br>
apache-airflow-providers-sqlite==2.0.1<br>
apache-airflow-providers-ssh==2.3.0</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">As I mentioned before, only after a part of DAG <strong>A</strong> tasks get executed Airflow starts to execute DAG <strong>B</strong> tasks. I defined the number of tasks as 50 to make sure the bug keeps occurring for a while. The lowest number of tasks in DAG <strong>A</strong> to activate this bug is 33. It means that at the beginning 1 task will start running and 32 will be kept in scheduled state. After that first task ended running (and one another DAG <strong>A</strong> task starts running) the DAG <strong>B</strong> acquires space to start running. But only one task at this moment. As the number of DAG <strong>A</strong> tasks in scheduled state decreases, the number of DAG <strong>B</strong> tasks running at the same time increases.</p>
<p dir="auto">From this experiment I can conclude that in some point the scheduler is limiting to 32 the number of tasks checked to transition from <code class="notranslate">scheduled</code> to <code class="notranslate">queued</code> state and ordering this query prioritizing the DAG <strong>A</strong> scheduled tasks that are forbidden to run due to the <code class="notranslate">max_active_tasks</code> usage.</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" checked=""> 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.6.1</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">This <a href="https://github.com/apache/airflow/pull/28187" data-hovercard-type="pull_request" data-hovercard-url="/apache/airflow/pull/28187/hovercard">PR</a> introduced the get_iam_token method in <code class="notranslate">redshift_sql.py</code>. This is the breaking change as  introduces the check for <code class="notranslate">iam</code> in extras, and it's set to False by default.</p>
<p dir="auto">Error log:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="
self = &lt;airflow.providers.amazon.aws.hooks.redshift_sql.RedshiftSQLHook object at 0x7f29f7c208e0&gt;
conn = redshift_default

    def get_iam_token(self, conn: Connection) -&gt; tuple[str, str, int]:
        &quot;&quot;&quot;
        Uses AWSHook to retrieve a temporary ***word to connect to Redshift.
        Port is required. If none is provided, default is used for each service
        &quot;&quot;&quot;
        port = conn.port or 5439
        # Pull the custer-identifier from the beginning of the Redshift URL
        # ex. my-cluster.ccdre4hpd39h.us-east-1.redshift.amazonaws.com returns my-cluster
&gt;       cluster_identifier = conn.extra_dejson.get(&quot;cluster_identifier&quot;, conn.host.split(&quot;.&quot;)[0])
E       AttributeError: 'NoneType' object has no attribute 'split'

.nox/test-3-8-airflow-2-6-0/lib/python3.8/site-packages/airflow/providers/amazon/aws/hooks/redshift_sql.py:107: AttributeError"><pre class="notranslate"><code class="notranslate">
self = &lt;airflow.providers.amazon.aws.hooks.redshift_sql.RedshiftSQLHook object at 0x7f29f7c208e0&gt;
conn = redshift_default

    def get_iam_token(self, conn: Connection) -&gt; tuple[str, str, int]:
        """
        Uses AWSHook to retrieve a temporary ***word to connect to Redshift.
        Port is required. If none is provided, default is used for each service
        """
        port = conn.port or 5439
        # Pull the custer-identifier from the beginning of the Redshift URL
        # ex. my-cluster.ccdre4hpd39h.us-east-1.redshift.amazonaws.com returns my-cluster
&gt;       cluster_identifier = conn.extra_dejson.get("cluster_identifier", conn.host.split(".")[0])
E       AttributeError: 'NoneType' object has no attribute 'split'

.nox/test-3-8-airflow-2-6-0/lib/python3.8/site-packages/airflow/providers/amazon/aws/hooks/redshift_sql.py:107: AttributeError
</code></pre></div>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">It should have backward compatibility</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">Run an example DAG for redshift with the AWS IAM profile given at hook initialization to retrieve a temporary password to connect to Amazon Redshift.</p>
<h3 dir="auto">Operating System</h3>
<p dir="auto">mac-os</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">Astronomer</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" checked=""> 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>