From 8535447fee1f430e6aa637ae5d550ac018dabc69 Mon Sep 17 00:00:00 2001 From: Vamsi-klu Date: Sat, 11 Jul 2026 06:46:06 +0000 Subject: [PATCH 1/2] Fix TimeSensor dag_version churn by removing start_from_trigger support TimeSensor computed target_datetime using datetime.now() at DAG-parse time and baked it into start_trigger_args.trigger_kwargs.moment. Since that field is serialized into serialized_dag, the dag hash changed on every parse, creating a new DagVersion each scheduler loop. This removes the start_from_trigger feature from TimeSensor (consistent with TimeDeltaSensor which never had it because its target depends on runtime context) and moves target_datetime computation from __init__ to execution time (poke/execute). Wall-clock behavior is preserved, but no volatile value enters serialization. Fixes: #69543 --- providers/standard/docs/sensors/datetime.rst | 2 +- .../providers/standard/sensors/time.py | 83 +++++++++++-------- 2 files changed, 50 insertions(+), 35 deletions(-) diff --git a/providers/standard/docs/sensors/datetime.rst b/providers/standard/docs/sensors/datetime.rst index 2a45ecb6bd869..3045a7ffc129f 100644 --- a/providers/standard/docs/sensors/datetime.rst +++ b/providers/standard/docs/sensors/datetime.rst @@ -41,7 +41,7 @@ TimeSensor Use the :class:`~airflow.providers.standard.sensors.time_sensor.TimeSensor` to end sensing after time specified. ``TimeSensor`` can be run in deferrable mode, if a Triggerer is available. -Time will be evaluated against ``data_interval_end`` if present for the Dag run, otherwise ``run_after`` will be used. +Time is evaluated against the wall-clock date in the Dag's timezone at execution time (poke or trigger time), not against ``data_interval_end`` or ``run_after``. For interval-relative behavior, use :class:`~airflow.providers.standard.sensors.time_delta.TimeDeltaSensor`. .. exampleinclude:: /../src/airflow/providers/standard/example_dags/example_sensors.py :language: python diff --git a/providers/standard/src/airflow/providers/standard/sensors/time.py b/providers/standard/src/airflow/providers/standard/sensors/time.py index e32a98d97d995..28cd2272bcb73 100644 --- a/providers/standard/src/airflow/providers/standard/sensors/time.py +++ b/providers/standard/src/airflow/providers/standard/sensors/time.py @@ -24,7 +24,6 @@ from airflow.exceptions import AirflowProviderDeprecationWarning from airflow.providers.common.compat.sdk import BaseSensorOperator, conf, timezone from airflow.providers.standard.triggers.temporal import DateTimeTrigger -from airflow.triggers.base import StartTriggerArgs if TYPE_CHECKING: from airflow.sdk import Context @@ -34,6 +33,11 @@ class TimeSensor(BaseSensorOperator): """ Waits until the specified time of the day. + The time is evaluated against the current date in the Dag's timezone + at execution time (poke or deferral), not at Dag-parse time. This avoids + dag_version churn that previously occurred when target_datetime was + baked into serialized start_trigger_args. + :param target_time: time after which the job succeeds :param deferrable: whether to defer execution @@ -43,53 +47,65 @@ class TimeSensor(BaseSensorOperator): """ - start_trigger_args = StartTriggerArgs( - trigger_cls="airflow.providers.standard.triggers.temporal.DateTimeTrigger", - trigger_kwargs={"moment": "", "end_from_trigger": False}, - next_method="execute_complete", - next_kwargs=None, - timeout=None, - ) - start_from_trigger = False - def __init__( self, *, target_time: datetime.time, deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False), - start_from_trigger: bool = False, end_from_trigger: bool = False, trigger_kwargs: dict[str, Any] | None = None, **kwargs, ) -> None: - super().__init__(**kwargs) - - # Create a "date-aware" timestamp that will be used as the "target_datetime". This is a requirement - # of the DateTimeTrigger - - # Get date considering dag.timezone - aware_time = timezone.coerce_datetime( - datetime.datetime.combine( - datetime.datetime.now(self.dag.timezone), target_time, self.dag.timezone + # Backwards compatibility: previously supported start_from_trigger which caused dag_version churn. + # Pop it if present and warn, instead of failing on unexpected kwarg. + start_from_trigger = kwargs.pop("start_from_trigger", None) + if start_from_trigger is not None: + warnings.warn( + "start_from_trigger is deprecated and no longer supported for TimeSensor. " + "It has been ignored. Target time is now always evaluated at execution time. " + "If you need trigger-time evaluation optimization, please open an issue.", + AirflowProviderDeprecationWarning, + stacklevel=2, ) - ) - - # Now that the dag's timezone has made the datetime timezone aware, we need to convert to UTC - self.target_datetime = timezone.convert_to_utc(aware_time) + super().__init__(**kwargs) + self.target_time = target_time self.deferrable = deferrable - self.start_from_trigger = start_from_trigger self.end_from_trigger = end_from_trigger - - if self.start_from_trigger: - self.start_trigger_args.trigger_kwargs = dict( - moment=self.target_datetime, end_from_trigger=self.end_from_trigger + # Store trigger_kwargs for API compatibility but not used for hash + self._trigger_kwargs = trigger_kwargs + + def _get_target_datetime(self) -> datetime.datetime: + """Compute target datetime at execution time, not parse time.""" + # Use Dag timezone if available, else UTC; matches prior coerce logic + dag_timezone = None + try: + if hasattr(self, "dag") and self.dag is not None: + dag_timezone = self.dag.timezone + except Exception: + dag_timezone = None + # Fallback: if dag timezone not set, coerce_datetime will handle via default + if dag_timezone is None: + # timezone.coerce_datetime expects a tzinfo, but when None it will use default + # Use UTC as safe fallback for datetime.now() + now_date = datetime.datetime.now(timezone.utc).date() + # combine with target_time, let coerce handle tz conversion + aware_time = timezone.coerce_datetime(datetime.datetime.combine(now_date, self.target_time)) + else: + aware_time = timezone.coerce_datetime( + datetime.datetime.combine(datetime.datetime.now(dag_timezone), self.target_time, dag_timezone) ) + return timezone.convert_to_utc(aware_time) + + @property + def target_datetime(self) -> datetime.datetime: + """Backward-compatible property, now computed at call time.""" + return self._get_target_datetime() def execute(self, context: Context) -> None: if self.deferrable: self.defer( trigger=DateTimeTrigger( - moment=self.target_datetime, # This needs to be an aware timestamp + moment=self._get_target_datetime(), end_from_trigger=self.end_from_trigger, ), method_name="execute_complete", @@ -101,10 +117,9 @@ def execute_complete(self, context: Context, event: Any = None) -> None: return None def poke(self, context: Context) -> bool: - self.log.info("Checking if the time (%s) has come", self.target_datetime) - - # self.target_date has been converted to UTC, so we do not need to convert timezone - return timezone.utcnow() > self.target_datetime + target_datetime = self._get_target_datetime() + self.log.info("Checking if the time (%s) has come", target_datetime) + return timezone.utcnow() > target_datetime class TimeSensorAsync(TimeSensor): From 98eeb777faa673e7a8c99f28eef6b72d78ec52b4 Mon Sep 17 00:00:00 2001 From: probe Date: Sun, 12 Jul 2026 19:55:28 +0000 Subject: [PATCH 2/2] Stabilize TimeSensor serialization cleanup TimeSensor no longer supports trigger-start evaluation because it caused parse-time serialized Dag churn, so the compatibility path needs deterministic tests and should not keep unused trigger kwargs around. --- .../providers/standard/sensors/time.py | 33 +++++----------- .../tests/unit/standard/sensors/test_time.py | 38 ++++++++++++++++++- 2 files changed, 46 insertions(+), 25 deletions(-) diff --git a/providers/standard/src/airflow/providers/standard/sensors/time.py b/providers/standard/src/airflow/providers/standard/sensors/time.py index 28cd2272bcb73..d2ccd9df1b744 100644 --- a/providers/standard/src/airflow/providers/standard/sensors/time.py +++ b/providers/standard/src/airflow/providers/standard/sensors/time.py @@ -35,7 +35,7 @@ class TimeSensor(BaseSensorOperator): The time is evaluated against the current date in the Dag's timezone at execution time (poke or deferral), not at Dag-parse time. This avoids - dag_version churn that previously occurred when target_datetime was + Dag version churn that previously occurred when target_datetime was baked into serialized start_trigger_args. :param target_time: time after which the job succeeds @@ -56,14 +56,12 @@ def __init__( trigger_kwargs: dict[str, Any] | None = None, **kwargs, ) -> None: - # Backwards compatibility: previously supported start_from_trigger which caused dag_version churn. - # Pop it if present and warn, instead of failing on unexpected kwarg. start_from_trigger = kwargs.pop("start_from_trigger", None) if start_from_trigger is not None: warnings.warn( "start_from_trigger is deprecated and no longer supported for TimeSensor. " "It has been ignored. Target time is now always evaluated at execution time. " - "If you need trigger-time evaluation optimization, please open an issue.", + "Use deferrable=True to defer the sensor instead.", AirflowProviderDeprecationWarning, stacklevel=2, ) @@ -71,29 +69,16 @@ def __init__( self.target_time = target_time self.deferrable = deferrable self.end_from_trigger = end_from_trigger - # Store trigger_kwargs for API compatibility but not used for hash - self._trigger_kwargs = trigger_kwargs + # Accepted for compatibility only; storing it would reintroduce serialized Dag hash churn. + del trigger_kwargs def _get_target_datetime(self) -> datetime.datetime: """Compute target datetime at execution time, not parse time.""" - # Use Dag timezone if available, else UTC; matches prior coerce logic - dag_timezone = None - try: - if hasattr(self, "dag") and self.dag is not None: - dag_timezone = self.dag.timezone - except Exception: - dag_timezone = None - # Fallback: if dag timezone not set, coerce_datetime will handle via default - if dag_timezone is None: - # timezone.coerce_datetime expects a tzinfo, but when None it will use default - # Use UTC as safe fallback for datetime.now() - now_date = datetime.datetime.now(timezone.utc).date() - # combine with target_time, let coerce handle tz conversion - aware_time = timezone.coerce_datetime(datetime.datetime.combine(now_date, self.target_time)) - else: - aware_time = timezone.coerce_datetime( - datetime.datetime.combine(datetime.datetime.now(dag_timezone), self.target_time, dag_timezone) - ) + dag_timezone = getattr(getattr(self, "dag", None), "timezone", None) or timezone.utc + now_date = datetime.datetime.now(dag_timezone).date() + aware_time = timezone.coerce_datetime( + datetime.datetime.combine(now_date, self.target_time, dag_timezone) + ) return timezone.convert_to_utc(aware_time) @property diff --git a/providers/standard/tests/unit/standard/sensors/test_time.py b/providers/standard/tests/unit/standard/sensors/test_time.py index 1989c35d91b58..35707e8524d27 100644 --- a/providers/standard/tests/unit/standard/sensors/test_time.py +++ b/providers/standard/tests/unit/standard/sensors/test_time.py @@ -23,10 +23,12 @@ import pytest import time_machine -from airflow.models.dag import DAG +from airflow.exceptions import AirflowProviderDeprecationWarning from airflow.providers.common.compat.sdk import TaskDeferred from airflow.providers.standard.sensors.time import TimeSensor from airflow.providers.standard.triggers.temporal import DateTimeTrigger +from airflow.sdk import DAG +from airflow.serialization.serialized_objects import LazyDeserializedDAG from tests_common.test_utils.compat import timezone @@ -135,3 +137,37 @@ def test_execute_complete_accepts_event(self): op.execute_complete(context={}, event={"status": "success"}) except TypeError as e: pytest.fail(f"TypeError raised: {e}") + + def test_start_from_trigger_is_ignored_with_warning(self): + with pytest.warns(AirflowProviderDeprecationWarning, match="start_from_trigger is deprecated"): + op = TimeSensor( + task_id="test", + target_time=time(10, 0), + start_from_trigger=True, + trigger_kwargs={"moment": "2020-01-01T10:00:00Z"}, + ) + + assert getattr(op, "start_from_trigger", False) is False + assert not hasattr(op, "_trigger_kwargs") + + def test_ignored_start_from_trigger_keeps_serialized_dag_hash_stable(self): + def build_hash() -> str: + with pytest.warns(AirflowProviderDeprecationWarning, match="start_from_trigger is deprecated"): + with DAG( + dag_id="test_ignored_start_from_trigger_hash", + schedule=None, + start_date=datetime(2020, 1, 1), + ) as dag: + TimeSensor( + task_id="test", + target_time=time(10, 0), + start_from_trigger=True, + ) + return LazyDeserializedDAG.from_dag(dag).hash + + with time_machine.travel("2025-01-01 00:00:00", tick=False): + first_hash = build_hash() + with time_machine.travel("2025-01-02 00:00:00", tick=False): + second_hash = build_hash() + + assert first_hash == second_hash