Scheduler fails when loading SubDAG after upgrade to 2.2.1

[sushi30]

Apache Airflow version

2.2.1 (latest released)

Operating System

Centos 7

Versions of Apache Airflow Providers

No response

Deployment

Official Apache Airflow Helm Chart

Deployment details

No response

What happened

After upgrade to Airflow 2.2.1, an existing DAG failed with the following error after which the scheduler crashes:

[2021-11-03 14:52:39,739] {dagrun.py:715} WARNING - Failed to record duration of <DagRun housekeeping_unified_test.add_partitions_unified_test_public @ 2021-11-03 13:37:54.190276+00:00: scheduled__2021-11-03T13:37:54.190276+00:00, externally triggered: True>: start_date is not set.
[2021-11-03 14:52:39,743] {scheduler_job.py:644} ERROR - Exception when executing SchedulerJob._run_scheduler_loop
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 628, in _execute
    self._run_scheduler_loop()
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 709, in _run_scheduler_loop
    num_queued_tis = self._do_scheduling(session)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 792, in _do_scheduling
    callback_to_run = self._schedule_dag_run(dag_run, session)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1044, in _schedule_dag_run
    self._update_dag_next_dagruns(dag, dag_model, active_runs)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 935, in _update_dag_next_dagruns
    data_interval = dag.get_next_data_interval(dag_model)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/models/dag.py", line 629, in get_next_data_interval
    return self.infer_automated_data_interval(dag_model.next_dagrun)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/models/dag.py", line 669, in infer_automated_data_interval
    end = cast(DeltaDataIntervalTimetable, self.timetable)._get_next(start)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/timetables/interval.py", line 278, in _get_next
    return convert_to_utc(current + self._delta)
TypeError: unsupported operand type(s) for +: 'NoneType' and 'datetime.timedelta'

What you expected to happen

1. DAG to continue running.
2. Scheduler should not fail.

How to reproduce

Try to run this DAG:

from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.operators.subdag import SubDagOperator
from airflow.utils.dates import days_ago

default_args = {
    "owner": "airflow",
}


def create_subdag(dag):
    with DAG(dag_id=f"{dag.dag_id}.subdag", start_date=days_ago(0)) as dag:
        DummyOperator(task_id="task")
    return dag


with DAG(
    dag_id="example_subdag",
    default_args=default_args,
    schedule_interval="@hourly",
    start_date=days_ago(2),
    tags=["example"],
) as dag:
    SubDagOperator(task_id="subdag", subdag=create_subdag(dag))

Anything else

Every time this DAG is turned on.

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

[sushi30]

As a workaround. Using this config and deleting the DAG allows it to continue running:

def create_subdag(dag):
    with DAG(dag_id=f"{dag.dag_id}.subdag", start_date=days_ago(0), schedule_interval="@once", catchup=False) as dag:
        DummyOperator(task_id="task")
    return dag

[RikHeijdens]

I'm not sure if by deleting the DAG you mean, delete the DAG and all history associated with it, but I was able to recover the scheduler from a crash loop by setting the schedule_interval to @once and then deleting the DAG's entry from the dag_code table. This would then allow the scheduler to recover, without having to delete all of the history.

[sushi30]

Installing Airflow after #19307 did the trick. So I guess the issue can be closed.