2.1.3/4 queued dag runs changes catchup=False behaviour

Apache Airflow version

2.1.4 (latest released)

Operating System

Amazon Linux 2

Versions of Apache Airflow Providers

No response

Deployment

Docker-Compose

Deployment details

No response

What happened

Say, for example, you have a DAG that has a sensor. This DAG is set to run every minute, with max_active_runs=1, and catchup=False.

This sensor may pass 1 or more times per day.

Previously, when this sensor is satisfied once per day, there is 1 DAG run for that given day. When the sensor is satisfied twice per day, there are 2 DAG runs for that given day.

With the new queued dag run state, new dag runs will be queued for each minute (up to AIRFLOW__CORE__MAX_QUEUED_RUNS_PER_DAG), this seems to be against the spirit of catchup=False.

This means that if a dag run is waiting on a sensor for longer than the schedule_interval, it will still in effect 'catchup' due to the queued dag runs.

What you expected to happen

No response

How to reproduce

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

this seems to be against the spirit of catchup=False

What is your understanding to catchup? This does not match my understanding to it.

https://airflow.apache.org/docs/apache-airflow/stable/dag-run.html?highlight=catchup#catchup

From the documentation’s description, catchup only matters for runs missed their turn. But since the scheduler is able to schedule those runs at the pace of once a minute, those successfully scheduled runs did not miss their turn, so catchup does not matter to them. The behaviour you described for 2.1.2 and below sounds like a bug to me.

Per the documentation:

"When turned off, the scheduler creates a DAG run only for the latest interval."

With the queued state, it creates dag runs for the latest interval and the next 16 intervals (by default)

I'm having the exactly same issue in the latest version, it breaks many of our DAGs that rely on the catchup setting.

We have DAGs scheduled every minute, most of them should finish within 1 minute, but sometimes it could exceed, e.g. network issue, external service slow, etc. When this happen, it will cumulate the DagRuns that we don't expect. We have internal backfill mechanism already.

The document said "Turning catchup off is great if your DAG performs catchup internally." in section "catchup"

Can you try the setting AIRFLOW__SCHEDULER__MAX_QUEUED_DAG_RUNS_PER_DAG=1

@ephraimbuddy

@ephraimbuddy thanks I'll try this out. I can accept it as workaround, but still it's conflicting with the idea of catch-up.

@uranusjr

@uranusjr
RE

From the documentation’s description, catchup only matters for runs missed their turn. But since the scheduler is able to schedule those runs at the pace of once a minute, those successfully scheduled runs did not miss their turn, so catchup does not matter to them. The behaviour you described for 2.1.2 and below sounds like a bug to me.

Oftentimes the usage of catchup=False is a case where execution_date does not matter, and there should only be one instance of the task running at a given time.

An example would be an incremental process. Each dag run, you process the new data that has arrived since last run.

So if you have your dag scheduled every 10 minutes let's say, and it takes an hour for some reason, you don't want it to create 6 new queued dag runs while it is waiting.

It should only queue up one at most.

There are many kinds of pipelines that use this pattern.

But the current behavior (verified in 2.2.0 as well) is, even when catchup=False, it will queue up a bunch of dag runs (up to max_queued_runs_per_dag). And this should not happen.

I encountered this today with a stuck dag (task had been stuck in running state for a day or so for some reason). 16 dag runs were queued up and waiting to go. But this should not be.

Along these lines... @ephraimbuddy in your PR #18897 you explicitly mention the catchup=True case, but from my understanding, what this issue is about is the catchup=False case... just wanted to bring that to your attention in case it's either a typo or a misunderstanding.

The PR captures both cases where catchup is False or true. It's annoying that it broke. For now limit it with max_queued_dagruns_per_dag, at least you'll not have a lot of dagruns

The PR captures both cases where catchup is False or true. It's annoying that it broke. For now limit it with max_queued_dagruns_per_dag, at least you'll not have a lot of dagruns

OK nice! I figured but just wanted to mention in case.

Thanks

@jedcunningham

So i have been also been working on a fix for this and the MR #18897 does not fix the issue.

It's very simple to recreate with a test dag:

import logging
import time
from datetime import datetime

from airflow import DAG
from airflow.operators.python import PythonOperator


def some_task():
    time.sleep(5)
    logging.info("Slept for 5")


dag = DAG(
    dag_id="aa_catchup_1",
    schedule_interval='@daily',
    catchup=True,
    start_date=datetime(2021, 9, 9),
    max_active_runs=1,
    max_active_tasks=1,
)


with dag:
    PythonOperator(
        task_id="some_task",
        python_callable=some_task
    )

Launch this dag
let it schedule a couple of dag runs (to simulate it being behind schedule)
Pause it.
Edit the dag file to catchup=False
Restart the scheduler
Confirm in the UI that the dag details now read catchup=False
** Now we have a dag that is catchup=False and is behind schedule **
Enable the dag
The next dag run it schedules will still not be the latest it will simply be the next dagrun from the current state

@jedcunningham or @ephraimbuddy would you kindly re-open this issue an i can provide a PR

Since this is catchup True and max_active_runs 1,
Only one dagrun is created. No extra dagruns would be created so it won't queue dagruns as you think.

If you take a look at the instructions beneath the example dag in my comment this is only used to get the dag behind schedule. It should then be amended to catchup=False

That's true.

Here is the fix #19130

Fixes apache#18487

Navigation Menu

Repository navigation

2.1.3/4 queued dag runs changes catchup=False behaviour #18487

Apache Airflow version

Operating System

Versions of Apache Airflow Providers

Deployment

Deployment details

What happened

What you expected to happen

How to reproduce

Anything else

Are you willing to submit PR?

Code of Conduct

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

2.1.3/4 queued dag runs changes catchup=False behaviour #18487

Description

Apache Airflow version

Operating System

Versions of Apache Airflow Providers

Deployment

Deployment details

What happened

What you expected to happen

How to reproduce

Anything else

Are you willing to submit PR?

Code of Conduct

Activity

uranusjr commented on Sep 23, 2021

StewartThomson commented on Sep 23, 2021

widewing commented on Oct 7, 2021

ephraimbuddy commented on Oct 7, 2021

widewing commented on Oct 8, 2021

dstandish commented on Oct 14, 2021

ephraimbuddy commented on Oct 14, 2021

dstandish commented on Oct 14, 2021

robinedwards commented on Oct 21, 2021

ephraimbuddy commented on Oct 21, 2021

robinedwards commented on Oct 21, 2021

ephraimbuddy commented on Oct 21, 2021

robinedwards commented on Oct 21, 2021

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions