<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.1.4</p>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Amazon Linux 2 AMI</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">Other</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto">Cloudformation template based on <a href="https://github.com/villasv/aws-airflow-stack/blob/v2/aws/cloud-formation-template.yml">https://github.com/villasv/aws-airflow-stack/blob/v2/aws/cloud-formation-template.yml</a></p>
<h3 dir="auto">What happened</h3>
<p dir="auto">DAG can have multiple active runs despite of <code class="notranslate">max_active_runs</code> set to 1 when it is triggered externally from TriggerDagRunOperator</p>
<h3 dir="auto">What you expected to happen</h3>
<p dir="auto">If DAG has max_active_runs = 1 it should never be started more than once</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">DAG <a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="69689814" data-permission-text="Title is private" data-url="https://github.com/apache/airflow/issues/1" data-hovercard-type="pull_request" data-hovercard-url="/apache/airflow/pull/1/hovercard" href="https://github.com/apache/airflow/pull/1">#1</a></p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="with DAG('dag_1',
                default_args=default_args,
                max_active_runs=1,
                start_date=datetime(2021, 11, 1),
                schedule_interval=&quot;* * * * *&quot;,
                dagrun_timeout=timedelta(minutes=1),
                catchup=False) as dag:

    TriggerDagRunOperator(task_id='trigger_dag_2',
                                                          trigger_dag_id='dag_2')"><pre class="notranslate"><code class="notranslate">with DAG('dag_1',
                default_args=default_args,
                max_active_runs=1,
                start_date=datetime(2021, 11, 1),
                schedule_interval="* * * * *",
                dagrun_timeout=timedelta(minutes=1),
                catchup=False) as dag:

    TriggerDagRunOperator(task_id='trigger_dag_2',
                                                          trigger_dag_id='dag_2')
</code></pre></div>
<p dir="auto">DAG <a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="69690133" data-permission-text="Title is private" data-url="https://github.com/apache/airflow/issues/2" data-hovercard-type="pull_request" data-hovercard-url="/apache/airflow/pull/2/hovercard" href="https://github.com/apache/airflow/pull/2">#2</a></p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="from airflow.operators.bash import BashOperator

with DAG('dag_2',
                default_args=default_args,
                max_active_runs=1,
                start_date=datetime(2021, 11, 1),
                schedule_interval=None,
                dagrun_timeout=timedelta(minutes=10),
                catchup=False) as dag:

    BashOperator(
        task_id=&quot;dag_keeper&quot;,
        bash_command=&quot;sleep 5m&quot;
    )"><pre class="notranslate"><code class="notranslate">from airflow.operators.bash import BashOperator

with DAG('dag_2',
                default_args=default_args,
                max_active_runs=1,
                start_date=datetime(2021, 11, 1),
                schedule_interval=None,
                dagrun_timeout=timedelta(minutes=10),
                catchup=False) as dag:

    BashOperator(
        task_id="dag_keeper",
        bash_command="sleep 5m"
    )
</code></pre></div>
<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.1.4 (latest released)</p>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Ubuntu 20.04</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">Virtualenv installation</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">What happened</h3>
<p dir="auto"><code class="notranslate">max_active_runs</code> isn't respected when dag is triggered externally by TriggerDagRunOperator</p>
<h3 dir="auto">What you expected to happen</h3>
<p dir="auto">Limit number of concurrent runs specified by <code class="notranslate">max_active_runs</code> ,subsequent dag runs should be allocated on new queue state</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">Modify example_dags\example_trigger_target_dag.py to limit <code class="notranslate">max_active_runs = 2 </code> and add time.sleep for test convenience :</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago


def run_this_func(**context):
    &quot;&quot;&quot;
    Print the payload &quot;message&quot; passed to the DagRun conf attribute.

    :param context: The execution context
    :type context: dict
    &quot;&quot;&quot;
    import time
    time.sleep(10)
    print(f&quot;Remotely received value of {context['dag_run'].conf['message']} for key=message&quot;)



with DAG(
    dag_id=&quot;example_trigger_target_dag&quot;,
    default_args={&quot;owner&quot;: &quot;airflow&quot;},
    start_date=days_ago(2),
    schedule_interval=None,
    max_active_runs=2,
    tags=['example'],
) as dag:

    run_this = PythonOperator(task_id=&quot;run_this&quot;, python_callable=run_this_func)

    bash_task = BashOperator(
        task_id=&quot;bash_task&quot;,
        bash_command='echo &quot;Here is the message: $message&quot;',
        env={'message': '{{ dag_run.conf[&quot;message&quot;] if dag_run else &quot;&quot; }}'},
    )
"><pre class="notranslate"><code class="notranslate">
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago


def run_this_func(**context):
    """
    Print the payload "message" passed to the DagRun conf attribute.

    :param context: The execution context
    :type context: dict
    """
    import time
    time.sleep(10)
    print(f"Remotely received value of {context['dag_run'].conf['message']} for key=message")



with DAG(
    dag_id="example_trigger_target_dag",
    default_args={"owner": "airflow"},
    start_date=days_ago(2),
    schedule_interval=None,
    max_active_runs=2,
    tags=['example'],
) as dag:

    run_this = PythonOperator(task_id="run_this", python_callable=run_this_func)

    bash_task = BashOperator(
        task_id="bash_task",
        bash_command='echo "Here is the message: $message"',
        env={'message': '{{ dag_run.conf["message"] if dag_run else "" }}'},
    )

</code></pre></div>
<p dir="auto">And manually trigger multiple runs of example_trigger_controller_dag  :</p>
<p dir="auto"><a target="_blank" rel="noopener noreferrer nofollow" href="https://user-images.githubusercontent.com/10963531/135154981-304f145b-7e96-43d6-9f06-e7a565922206.PNG"><img src="https://user-images.githubusercontent.com/10963531/135154981-304f145b-7e96-43d6-9f06-e7a565922206.PNG" alt="max_active_run_ignored" style="max-width: 100%;"></a></p>
<h3 dir="auto">Anything else</h3>
<p dir="auto">Same behavior if triggered with cli:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content=" for i in {1..10}; do  airflow dags trigger example_trigger_target_dag -c '{&quot;message&quot;: &quot;Hello World&quot;}'; done"><pre class="notranslate"><code class="notranslate"> for i in {1..10}; do  airflow dags trigger example_trigger_target_dag -c '{"message": "Hello World"}'; done
</code></pre></div>
<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>