Issue 1:
Apache Airflow version

2.2.2

What happened

I'm running two DAGs composed by 50 independent tasks in each DAG. Lets call DAG A and DAG B. DAG A has max_active_tasks set to 1 which means that only one task can run each time in this DAG.

When I run only DAG B its 50 tasks get executed properly, respecting the AIRFLOW__CORE__MAX_ACTIVE_TASKS_PER_DAG parameter, which limits the number of running tasks at the same time, defaulted to 16.

But when I run DAG A and B at the same time all tasks of DAG B get stucked in scheduled state. On the other hand DAG A still executing properly, running tasks one by one, respecting the max_active_tasks parameter.

It seems that the max_active_tasks used together with a high number of scheduled tasks is causing a side effect on any other DAG.

Only after a part of DAG A tasks gets executed Airflow starts to execute DAG B tasks.

One way to overcome this bug is raising AIRFLOW__CORE__PARALLELISM variable. Nevertheless it seems a logical bug in the scheduler.

What you expected to happen

Considering that max_active_tasks was set only in DAG A, DAG B tasks should start running anytime since they were scheduled.

How to reproduce

Airflow 2.2.2 with LocalExecutor on Python 3.9

DA A

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()

DA B

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()

Start DAG A and after a few seconds start DAG B.

Operating System

Debian GNU/Linux 10 (buster)

Versions of Apache Airflow Providers

apache-airflow-providers-amazon==2.4.0

apache-airflow-providers-celery==2.1.0

apache-airflow-providers-cncf-kubernetes==2.1.0

apache-airflow-providers-docker==2.3.0

apache-airflow-providers-elasticsearch==2.1.0

apache-airflow-providers-ftp==2.0.1

apache-airflow-providers-google==6.1.0

apache-airflow-providers-grpc==2.0.1

apache-airflow-providers-hashicorp==2.1.1

apache-airflow-providers-http==2.0.1

apache-airflow-providers-imap==2.0.1

apache-airflow-providers-microsoft-azure==3.3.0

apache-airflow-providers-mongo==2.3.0

apache-airflow-providers-mysql==2.1.1

apache-airflow-providers-odbc==2.0.1

apache-airflow-providers-postgres==2.3.0

apache-airflow-providers-redis==2.0.1

apache-airflow-providers-sendgrid==2.0.1

apache-airflow-providers-sftp==2.2.0

apache-airflow-providers-slack==4.1.0

apache-airflow-providers-sqlite==2.0.1

apache-airflow-providers-ssh==2.3.0

Deployment

Docker-Compose

Deployment details

No response

Anything else

As I mentioned before, only after a part of DAG A tasks get executed Airflow starts to execute DAG B 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 A 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 A task starts running) the DAG B acquires space to start running. But only one task at this moment. As the number of DAG A tasks in scheduled state decreases, the number of DAG B tasks running at the same time increases.

From this experiment I can conclude that in some point the scheduler is limiting to 32 the number of tasks checked to transition from scheduled to queued state and ordering this query prioritizing the DAG A scheduled tasks that are forbidden to run due to the max_active_tasks usage.

Are you willing to submit PR?

Yes I am willing to submit a PR!

Code of Conduct

I agree to follow this project's Code of Conduct

Issue 2:
Apache Airflow version

2.6.1

What happened

This PR introduced the get_iam_token method in redshift_sql.py. This is the breaking change as introduces the check for iam in extras, and it's set to False by default.

Error log:

self = <airflow.providers.amazon.aws.hooks.redshift_sql.RedshiftSQLHook object at 0x7f29f7c208e0>
conn = redshift_default

def get_iam_token(self, conn: Connection) -> 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
> 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

What you think should happen instead

It should have backward compatibility

How to reproduce

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.

Operating System

mac-os

Versions of Apache Airflow Providers

No response

Deployment

Astronomer

Deployment details

No response

Anything else

No response

Are you willing to submit PR?

Yes I am willing to submit a PR!

Code of Conduct

I agree to follow this project's Code of Conduct