jason810496 commented on code in PR #72498:
URL: https://github.com/apache/airflow/pull/72498#discussion_r3948167580
##########
airflow-core/src/airflow/timetables/interval.py:
##########
@@ -111,19 +111,20 @@ def next_dagrun_info(
# Data interval starts from the end of the previous interval.
start = align_last_data_interval_end
- # CronTriggerTimetable stores its runs as point-in-time intervals
- # (start == end == logical_date). After a switch to a
- # CronDataIntervalTimetable the aligned `start` lands back on that
- # same logical_date, so without this guard we'd propose a run
- # identical to the existing one — which collides with the
- # (dag_id, logical_date) unique constraint and leaves the scheduler
- # looping on "run already exists; skipping dagrun creation" until
- # the next period elapses. Advance one period to skip past it.
- if (
- last_automated_data_interval.start ==
last_automated_data_interval.end
- and start == last_automated_data_interval.start
- ):
+ # A schedule change (e.g. a coarser cron) can realign `start` onto
or
+ # before the previous run's start, colliding with the (dag_id,
+ # logical_date) unique constraint and stalling the scheduler on
"run
+ # already exists; skipping dagrun creation". One retry past it is
+ # provably enough for both shipped subclasses; fail loudly instead
of
+ # retrying indefinitely, which would hang the scheduler for every
Dag
+ # if `_get_next` ever stopped strictly advancing.
+ if start <= last_automated_data_interval.start:
start = self._get_next(start)
+ if start <= last_automated_data_interval.start:
+ raise ValueError(
+ f"{type(self).__name__}._get_next did not advance past
"
Review Comment:
Addressed in 95e782701b.
##########
airflow-core/tests/unit/jobs/test_schedule_change_stall.py:
##########
@@ -0,0 +1,174 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Regression test for https://github.com/apache/airflow/issues/66754: changing a
Dag's
+cron to a coarser one (e.g. hourly to daily) must not permanently stall DagRun
creation.
+Calls the DagRun-creation and TI-scheduling code paths directly
+(``SchedulerJobRunner._create_dag_runs``, ``DagRun.update_state``,
+``DagRun.schedule_tis``) against a real Dag re-sync
(``SerializedDAG.bulk_write_to_db``
+via ``dag_maker``) and real ``DagRun``/``TaskInstance`` rows, not just
+``CronDataIntervalTimetable`` in isolation, to prove the stall reproduces at
the
+database level and that the guard in ``timetables/interval.py`` resolves it.
+"""
+
+from __future__ import annotations
+
+import datetime
+
+import pytest
+import time_machine
+from sqlalchemy import select
+
+from airflow._shared.timezones import timezone
+from airflow.jobs.job import Job
+from airflow.jobs.scheduler_job_runner import SchedulerJobRunner
+from airflow.models import DagRun
+from airflow.providers.standard.operators.bash import BashOperator
+from airflow.utils.state import DagRunState, TaskInstanceState
+
+from tests_common.test_utils.db import (
+ clear_db_assets,
+ clear_db_backfills,
+ clear_db_callbacks,
+ clear_db_dags,
+ clear_db_deadline,
+ clear_db_import_errors,
+ clear_db_jobs,
+ clear_db_pools,
+ clear_db_runs,
+ clear_db_triggers,
+)
+from tests_common.test_utils.mock_executor import MockExecutor
+
+pytestmark = pytest.mark.db_test
+
+DAG_ID = "schedule_change_stall_coarser_cron"
+START_DATE = timezone.datetime(2026, 5, 4)
+
+
+def _clean_db():
+ clear_db_dags()
+ clear_db_runs()
+ clear_db_backfills()
+ clear_db_pools()
+ clear_db_import_errors()
+ clear_db_jobs()
+ clear_db_assets()
+ clear_db_deadline()
+ clear_db_callbacks()
+ clear_db_triggers()
+
+
[email protected](autouse=True)
+def clean_db():
+ _clean_db()
+ yield
+ _clean_db()
+
+
+def _make_runner():
+ job = Job()
+ return SchedulerJobRunner(job=job, executors=[MockExecutor()])
Review Comment:
Addressed in cb21bdd6fa.
##########
airflow-core/tests/unit/jobs/test_schedule_change_stall.py:
##########
@@ -0,0 +1,174 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Regression test for https://github.com/apache/airflow/issues/66754: changing a
Dag's
+cron to a coarser one (e.g. hourly to daily) must not permanently stall DagRun
creation.
+Calls the DagRun-creation and TI-scheduling code paths directly
+(``SchedulerJobRunner._create_dag_runs``, ``DagRun.update_state``,
+``DagRun.schedule_tis``) against a real Dag re-sync
(``SerializedDAG.bulk_write_to_db``
+via ``dag_maker``) and real ``DagRun``/``TaskInstance`` rows, not just
+``CronDataIntervalTimetable`` in isolation, to prove the stall reproduces at
the
+database level and that the guard in ``timetables/interval.py`` resolves it.
+"""
+
+from __future__ import annotations
+
+import datetime
+
+import pytest
+import time_machine
+from sqlalchemy import select
+
+from airflow._shared.timezones import timezone
+from airflow.jobs.job import Job
+from airflow.jobs.scheduler_job_runner import SchedulerJobRunner
+from airflow.models import DagRun
+from airflow.providers.standard.operators.bash import BashOperator
+from airflow.utils.state import DagRunState, TaskInstanceState
+
+from tests_common.test_utils.db import (
+ clear_db_assets,
+ clear_db_backfills,
+ clear_db_callbacks,
+ clear_db_dags,
+ clear_db_deadline,
+ clear_db_import_errors,
+ clear_db_jobs,
+ clear_db_pools,
+ clear_db_runs,
+ clear_db_triggers,
+)
+from tests_common.test_utils.mock_executor import MockExecutor
+
+pytestmark = pytest.mark.db_test
+
+DAG_ID = "schedule_change_stall_coarser_cron"
+START_DATE = timezone.datetime(2026, 5, 4)
+
+
+def _clean_db():
+ clear_db_dags()
+ clear_db_runs()
+ clear_db_backfills()
+ clear_db_pools()
+ clear_db_import_errors()
+ clear_db_jobs()
+ clear_db_assets()
+ clear_db_deadline()
+ clear_db_callbacks()
+ clear_db_triggers()
+
+
[email protected](autouse=True)
+def clean_db():
+ _clean_db()
+ yield
+ _clean_db()
+
+
+def _make_runner():
+ job = Job()
+ return SchedulerJobRunner(job=job, executors=[MockExecutor()])
+
+
+@time_machine.travel(START_DATE, tick=False)
+def test_coarser_schedule_change_does_not_stall_dagrun_creation(dag_maker,
session):
+ # 1. First Dag scheduling: hourly, catchup=True.
+ with dag_maker(
+ dag_id=DAG_ID,
+ schedule="0 * * * *",
+ start_date=START_DATE,
+ catchup=True,
+ max_active_runs=1,
+ session=session,
+ ):
+ BashOperator(task_id="do_something", bash_command="true")
+
+ dag_model = dag_maker.dag_model
+ assert dag_model.next_dagrun == START_DATE
+ assert dag_model.next_dagrun_create_after == START_DATE +
datetime.timedelta(hours=1)
+
+ runner = _make_runner()
+
+ # Tick once the hourly run becomes due.
+ with time_machine.travel(START_DATE + datetime.timedelta(hours=1),
tick=False):
+ runner._create_dag_runs([dag_model], session)
+ session.flush()
+
+ runs = session.scalars(select(DagRun).where(DagRun.dag_id == DAG_ID)).all()
+ assert len(runs) == 1, f"expected exactly one DagRun after the first tick,
got {runs}"
+ hourly_run = runs[0]
+ assert hourly_run.logical_date == START_DATE
+ assert hourly_run.data_interval_end == START_DATE +
datetime.timedelta(hours=1)
+
+ # 2. The first TI of the first Dag scheduling already ran.
+ ti = hourly_run.get_task_instances(session=session)[0]
+ ti.state = TaskInstanceState.SUCCESS
+ session.merge(ti)
+ session.flush()
+ hourly_run.dag = dag_maker.serialized_dag
+ hourly_run.update_state(session=session)
+ session.flush()
+ assert hourly_run.state == DagRunState.SUCCESS
+
+ # 3. Change the Dag scheduling (the second Dag): hourly to daily, drop
end_date.
Review Comment:
Addressed in e2025f8750.
##########
airflow-core/tests/unit/timetables/test_interval_timetable.py:
##########
@@ -73,12 +77,12 @@ def test_no_catchup_first_starts_at_current_time(
)
@time_machine.travel(pendulum.DateTime(2021, 9, 7, 15, tzinfo=utc))
def test_zero_length_last_interval_does_not_re_emit_logical_date(catchup:
bool) -> None:
- """A zero-length ``data_interval`` (``start == end``) on the previous run
- must not cause ``next_dagrun_info`` to re-emit that run's logical_date.
-
- These appear when a DAG was scheduled by ``CronTriggerTimetable`` and later
- switched to ``CronDataIntervalTimetable``. Without the guard the scheduler
- loops on "run already exists; skipping dagrun creation".
+ """A zero-length ``data_interval`` (``start == end``) on the previous run
must not
+ cause ``next_dagrun_info`` to re-emit that run's logical_date. These
appear when a
+ DAG was scheduled by ``CronTriggerTimetable`` and later switched to
Review Comment:
Addressed in a6bb67e69a.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
