max_active_runs = 1 can still create multiple active execution runs

[match-gabeflores]

Edit: There is a separate issue affecting max_active_runs in 1.10.14. That regression is fixed in 1.10.15.

Edit2: Version v2.1.3 contains some fixes but also contains bad regressions involving max_active_runs. Use v2.14 for the complete fixes to this issue

Edit3: Version 2.2.0 contains a fix for max_active_runs using dags trigger command or TriggerDagRunOperator. #18583

--

Apache Airflow version: 1.10.11, localExecutor

What happened:

I have max_active_runs = 1 in my dag file (which consists of multiple tasks) and I manually triggered a dag. While it was running, a second execution began under its scheduled time while the first execution was running.

I should note that the second execution is initially queued. It's only when the dag's 1st execution moves to the next task that the second execution actually starts.

My dag definition. The dag just contains tasks using pythonOperator.

dag = DAG(
    'dag1',
    default_args=default_args,
    description='xyz',
    schedule_interval=timedelta(hours=1),
    catchup=False,
    max_active_runs=1
)

What you expected to happen:

Only one execution should run. A second execution should be queued but not begin executing.

How to reproduce it:
In my scenario:

1. Manually trigger dag with multiple tasks.. have task1 take longer than the beginning of the next scheduled execution. (Dag Execution1). As an example, if the scheduled interval is 1 hour, have task1 take longer than 1 hour so as to queue up the second execution (Execution2).
2. When task1 of Execution1 finishes and just before starting task2, the second execution (Execution2, which is already queued) begins running.

Anything else we need to know:
I think the second execution begins in between the task1 and task2 of execution1. I think there's a few second delay there and maybe that's when Airflow thinks there's no dag execution? That's just a guess.

Btw, this can have potentially disastrous effects (errors, incomplete data without errors, etc)

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template!

[mik-laj]

The problem is, we don't have a state that describes DAG Run that are saved but not running. All DAG Run have running state initially. If we want to fix this bug we have to add a new dag state.

[dinesh-procore]

I am running into the exact same issue.

[alechheb]

The same issue here

[ashb]

Would someone be able to test if this specific case still happens on Airflow 2.0.0alpha1? (A few things about how we created DagRuns changed so this might have been fixed, but I didn't specifically set out to fix this.

[ashb]

Read the reproduction steps, and this bit sounds bang on:

I think the second execution begins in between the task1 and task2 of execution1. I think there's a few second delay there and maybe that's when Airflow thinks there's no dag execution? That's just a guess.

Yes, looking at the code that sounds right, and also hasn't changed in 2.0.0alpha1, the same logic is used.

[natejenkins21]

Same issue here. Causing a lot of issues for my backfills...

[kaxil]

@natejenkins21 Can you provide reproduction steps please?

[nathadfield]

For what it's worth, this doesn't seem to be an issue on master but it certainly is on 2.0 installed from pip.

I created this DAG, switched it on and allowed it to catchup and then cleared the status of all tasks.

from airflow import models
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2020, 12, 10),
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

dag_name = 'test_dag'

with models.DAG(dag_name,
                default_args=default_args,
                schedule_interval='0 0 * * *',
                catchup=True,
                max_active_runs=1
                ) as dag:

    start = BashOperator(
        task_id=f'start',
        bash_command='echo "Starting"; echo "Sleeping for 30 seconds"; sleep 30; echo "Finished"'
    )

As you can see, all the tasks are now running at the same time.

[nathadfield]

Here's the same DAG running against master via breeze.

[kaxil]

@nathadfield Does it occur without clearing tasks instances?

[nathadfield]

@nathadfield Does it occur without clearing tasks instances?

Yes. I removed catchup and triggered the DAG several times via the CLI and all of the tasks were running at the same time.

[kaxil]

@nathadfield What metadata DB do you use?

Just wondering if it is related to #13278

[nathadfield]

#13278

@kaxil We're using Postgres so it's probably not related.

[kaxil]

Cool, I am looking at this today and tomorrow

[kaxil]

This happens in Master too but only happens when you manually trigger DAGs multiple time.

[kamac]

This also happens when triggering DAGs manually with create_dagrun. If the DagRun is created with state RUNNING, max_active_runs is not respected (the run isn't counted if you browse DAG's details page).

I've also tried creating the run with state SCHEDULED, but it's never run by the scheduler. (perhaps I'm doing it the wrong way - I'm still investigating that)

Here's the code for two DAGs to reproduce this: https://gist.github.com/kamac/7112af78f1a9004142903d4fe6e387d4

[ashb]

I've also tried creating the run with state SCHEDULED, but it's never run by the scheduler. (perhaps I'm doing it the wrong way - I'm still investigating that)

Right now the only states for Dag Runs are None, "running", or "failed" -- that's why the scheduler is never picking up that dag run.

[kamac]

@ashb I've tried creating DagRuns with state NONE. The tasks get run now (the last two instances), but max_active_runs is not respected.

I'm on airflow 1.10.13, SequentialExecutor

[kaxil]

@ashb I've tried creating DagRuns with state NONE. The tasks get run now (the last two instances), but max_active_runs is not respected.

I'm on airflow 1.10.13, SequentialExecutor

🤔 1.10.13 ?? Just noticed the original issue creator was on 1.10.11 so this is not a regression in 2.0 -- I though I read somewhere that it worked correctly for you @nathadfield on 1.10.x -- can you confirm please.

[nathadfield]

@kaxil It is definitely an issue on 1.10.14 as that is what we're running on our production system at the moment.

It does also seem to be an issue on 2.0 that I've got running on a locally running dev instance of our setup, but I could not replicate the problem on master via Breeze.

[kaxil]

@kaxil It is definitely an issue on 1.10.14 as that is what we're running on our production system at the moment.

It does also seem to be an issue on 2.0 that I've got running on a locally running dev instance of our setup, but I could not replicate the problem on master via Breeze.

I might ping you on Slack for more ad-hoc discussion, I was able to replicate it on Master too: #9975 (comment)

[nathadfield]

@kaxil Is this likely to only be fixed in 2.1 or might we also see it in a new 1.10 version for those people who are affected by this issue but cannot (or don't want to) move to 2.0 yet?

[Adam-Leyshon]

This issue is affecting 1.10.14.

Upgrading to 2.0 is not possible for us at this time since we have a number of internal blockers on plugins that we wrote that require refactoring.

We cleared a DAG run for 2019-01-01 and selected the all future tasks option,
Now it want to processes over 200 runs at the same time

We did not experience this issue in 1.10.9.

Instead it is now trying to run a single task for each of the active runs in parallel instead of waiting for the first run to complete.

What I expect to happen is for it to clear the tasks and then complete each run sequentially, our system requires data be loaded in that order, having multiple parallel runs is causing huge issues for us.

[kaxil]

looks like max_active_runs was broken further in 1.10.13 -- This will be fixed in 1.10.15 by #13803 . However like I mentioned in #9975 (comment) not all use-cases will be fixed.

Complete fix will require adding a new state to DagRun -- 'queued' -- similar to Taskinstance

[himabindu07]

verified by QA

[aran3]

I've been testing version 2.1.3 for a few days now, and while adding Queued state seem to help most cases I think this bug is not fully solved.
When manually triggering a dag that has max_active_runs=1 many times, it does happen that dags reaches more than dag run in "running" state at the same time.
In our case the dag has two tasks, a trigger and a sensor:

I will try to gather more specific information and update.

[kamushadenes]

Observing the same behavior on 2.1.2 with catchup=False, this has been blowing through my quotas.

[ephraimbuddy]

@aran3, I think your case has been fixed in #17786 where tasks can start running while the dagruns are still queued. This would lead to the queued dagrun entering the running state

[vumdao]

Issue does not happen on 2.1.4

[argemiront]

Is there a way to prevent the scheduler to queue new runs if there's an active run? I have a DAG that now and then overruns and because of this behaviour I'm seeing DAG runs piling up on the new UI through time due to this new behaviour

[ephraimbuddy]

@argemiront, If you are on 2.1.4 you can change this:
AIRFLOW__SCHEDULER__MAX_QUEUED_RUNS_PER_DAG=16
to a lower number

[argemiront]

AIRFLOW__SCHEDULER__MAX_QUEUED_RUNS_PER_DAG=16

thank you so much!

[dave-martinez]

Issue does not happen on 2.1.4

I tested this one with Docker Airflow 2.1.4 Python 3.7.
Only works when triggered via HTTP (UI/API)

However, when using TriggerDagRun Operator, it doesn't work, or is this is the intended behavior for TriggerDagRun?

[ephraimbuddy]

What you see there is queued run. The currently active run is 1 but there's also a queued run which doesn't count as an active run.

[aran3]

It is worth noting for anyone (such as us) that heavily relies on max_active_runs=1 - that this still happens in 2.1.4 when using cli dags trigger command or TriggerDagRunOperator and was supposedly fixed in #18583 (version 2.2.0)

[ephraimbuddy]

It is worth noting for anyone (such as us) that heavily relies on max_active_runs=1 - that this still happens in 2.1.4 when using cli dags trigger command or TriggerDagRunOperator and was supposedly fixed in #18583 (version 2.2.0)

This is now fixed

[DanielMorales9]

How can this be solved on a fully managed setup such as MWAA? MWAA only supports 1.10.12 and 2.0.2?
I am looking for a workaround here, any help will be appreciated.

[ashb]

@DanielMorales9 Not easily I'm afraid - by asking AWS to provide a more recent version, or use a different method than MWAA that providers quicker update cycles.

[stroykova]

Same problem in 2.1.3 with manually triggered dags. All of them run simultaneously. I will try 2.2.0

[ashb]

@stroykova Please let us know. (I'd try 2.2.2 than 2.2.0)

[stroykova]

2.2.2 is fine with this 🥳

[stas-snow]

I'm seeing this issue in 2.2.3.
catchup=True and max_active_runs=1.
DAG is triggered multiple times and multiple instances are running in parallel.

[GabeChurch]

If you are on an older version of airflow that has this problem you can add concurrency setting to you dag (ie concurrency = some_num ). That or depends_on_past = True

[nathadfield]

@alexstrimbeanu Coming in here with an attitude like that is unacceptable and is not going to help your cause but I'll give you the decency of replying.

You may notice that this particular issue is closed and, afaik, there isn't currently an open issue that documents this as a problem.

Maybe you would like to open one and provide it will all the necessary information so that someone can replicate the scenario?