This is an automated email from the ASF dual-hosted git repository.
ferruzzi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 07a0c4da099 Fix TypeError clearing a DAG run when a queued-at deadline
interval is stored as JSON (#70370)
07a0c4da099 is described below
commit 07a0c4da099d28b6b387fb1c7ec416b00ba763a4
Author: Hemkumar Chheda <[email protected]>
AuthorDate: Sat Oct 10 05:06:04 2026 +0530
Fix TypeError clearing a DAG run when a queued-at deadline interval is
stored as JSON (#70370)
* Fix TypeError clearing a DAG run when a queued-at deadline interval is
stored as JSON
Since 3.3.0 the deadline_alert.interval column is JSON (migration 0117,
#64751),
holding a serialized timedelta or VariableInterval.
_recalculate_dagrun_queued_at_deadlines
still passed that raw dict into timedelta(seconds=...), so clearing a DAG
run or task
instance whose DAG has a DagRunQueuedAtDeadline failed with
"TypeError: unsupported type for timedelta seconds component: dict" (HTTP
500).
Decode the interval the same way DagRun creation does
(decode_deadline_alert), then
resolve a VariableInterval to a timedelta before recalculating. The
existing test stored
the interval as a plain float and never exercised the JSON path; it now
stores the
serialized envelope and is parametrized over fixed-timedelta and
VariableInterval.
closes: #70368
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
Co-authored-by: ferruzzi <[email protected]>
Co-authored-by: Amogh Desai <[email protected]>
---
airflow-core/src/airflow/models/taskinstance.py | 26 +++-
airflow-core/src/airflow/serialization/decoders.py | 44 +++++-
.../src/airflow/serialization/definitions/dag.py | 26 +---
airflow-core/tests/unit/models/test_dagrun.py | 18 +++
.../tests/unit/models/test_taskinstance.py | 156 +++++++++++++++++++--
.../tests/unit/serialization/test_decoders.py | 84 +++++++++++
6 files changed, 313 insertions(+), 41 deletions(-)
diff --git a/airflow-core/src/airflow/models/taskinstance.py
b/airflow-core/src/airflow/models/taskinstance.py
index 6c2745f0c70..19934254e97 100644
--- a/airflow-core/src/airflow/models/taskinstance.py
+++ b/airflow-core/src/airflow/models/taskinstance.py
@@ -263,10 +263,30 @@ def _recalculate_dagrun_queued_at_deadlines(
if not results:
return
+ # Local import to avoid a circular import between models and serialization.
+ from airflow.serialization.decoders import decode_deadline_alert_model,
resolve_deadline_alert_interval
+
for deadline, deadline_alert in results:
- # We can't use evaluate_with() since the new queued_at is not written
to the DB yet.
- deadline_interval = timedelta(seconds=deadline_alert.interval)
- new_deadline_time = new_queued_at + deadline_interval
+ # We can't use evaluate_with() since the new queued_at is not written
to the DB yet, and
+ # interval is stored as JSON, so it has to be decoded rather than
passed to timedelta().
+ try:
+ interval = resolve_deadline_alert_interval(
+ decode_deadline_alert_model(deadline_alert), session=session
+ )
+ except (ValueError, TypeError):
+ # Either step can fail: the alert may hold a payload the decoder
refuses, or a
+ # variable-backed interval may point at a Variable that is missing
or non-numeric.
+ # Leave this deadline alone rather than failing the whole clear.
+ log.warning(
+ "Error while recalculating deadline %s for DagRun %s.%s,
leaving it unchanged",
+ deadline.id,
+ dagrun.dag_id,
+ dagrun.run_id,
+ exc_info=True,
+ )
+ continue
+
+ new_deadline_time = new_queued_at + interval
log.debug(
"Recalculating deadline %s for DagRun %s.%s: old=%s, new=%s",
diff --git a/airflow-core/src/airflow/serialization/decoders.py
b/airflow-core/src/airflow/serialization/decoders.py
index fb1651bfb01..8c42ec0c987 100644
--- a/airflow-core/src/airflow/serialization/decoders.py
+++ b/airflow-core/src/airflow/serialization/decoders.py
@@ -53,6 +53,9 @@ from airflow.serialization.helpers import (
)
if TYPE_CHECKING:
+ from sqlalchemy.orm import Session
+
+ from airflow.models.deadline_alert import DeadlineAlert as
DeadlineAlertModel
from airflow.partition_mappers.base import PartitionMapper
from airflow.partition_mappers.wait_policy import WaitPolicy
from airflow.partition_mappers.window import Window
@@ -308,7 +311,7 @@ def _decode_deadline_callback(raw_callback: Any) ->
AsyncCallback | SyncCallback
return callback_cls(callback_callable=_SerializedCallbackPath(path),
**fields)
-def decode_deadline_alert(encoded_data: dict):
+def decode_deadline_alert(encoded_data: dict) -> SerializedDeadlineAlert:
"""
Decode a previously serialized deadline alert.
@@ -337,6 +340,45 @@ def decode_deadline_alert(encoded_data: dict):
)
+def decode_deadline_alert_model(deadline_alert: DeadlineAlertModel) ->
SerializedDeadlineAlert:
+ """
+ Decode a ``DeadlineAlert`` ORM row into its serialized representation.
+
+ :meta private:
+ """
+ return decode_deadline_alert(
+ {
+ DeadlineAlertFields.NAME: deadline_alert.name,
+ DeadlineAlertFields.REFERENCE: deadline_alert.reference,
+ DeadlineAlertFields.INTERVAL: deadline_alert.interval,
+ DeadlineAlertFields.CALLBACK: deadline_alert.callback_def,
+ }
+ )
+
+
+def resolve_deadline_alert_interval(
+ alert: SerializedDeadlineAlert, *, session: Session
+) -> datetime.timedelta:
+ """
+ Resolve a decoded alert's interval to a ``timedelta``.
+
+ A ``SerializedVariableInterval`` reads its Airflow Variable here, so this
is only called at
+ the point a deadline is actually calculated. It raises ``ValueError`` if
the Variable is
+ missing or is not an integer number of seconds.
+
+ :param alert: The decoded alert whose interval should be resolved.
+ :param session: Existing SQLAlchemy Session. Required rather than
defaulted: both callers run
+ under the scheduler's ``prohibit_commit`` guard, so the open session
has to reach
+ ``Variable.get`` instead of ``provide_session`` handing back the same
scoped session and
+ rolling it back on exit.
+
+ :meta private:
+ """
+ if isinstance(alert.interval, SerializedVariableInterval):
+ return alert.interval.resolve(session=session)
+ return alert.interval
+
+
def decode_timetable(var: dict[str, Any]) -> CoreTimetable:
"""
Decode a previously serialized timetable.
diff --git a/airflow-core/src/airflow/serialization/definitions/dag.py
b/airflow-core/src/airflow/serialization/definitions/dag.py
index 0acb5a2faa3..6f02d62c51e 100644
--- a/airflow-core/src/airflow/serialization/definitions/dag.py
+++ b/airflow-core/src/airflow/serialization/definitions/dag.py
@@ -50,14 +50,9 @@ from airflow.models.deadline import Deadline
from airflow.models.deadline_alert import DeadlineAlert as DeadlineAlertModel
from airflow.models.taskinstancekey import TaskInstanceKey
from airflow.models.tasklog import LogTemplate
-from airflow.serialization.decoders import decode_deadline_alert
-from airflow.serialization.definitions.deadline import (
- DeadlineAlertFields,
- SerializedReferenceModels,
- SerializedVariableInterval,
-)
+from airflow.serialization.decoders import decode_deadline_alert_model,
resolve_deadline_alert_interval
+from airflow.serialization.definitions.deadline import
SerializedReferenceModels
from airflow.serialization.definitions.param import SerializedParamsDict
-from airflow.serialization.enums import DagAttributeTypes as DAT, Encoding
from airflow.timetables.base import DagRunInfo, DataInterval, TimeRestriction
from airflow.utils.helpers import prune_dict
from airflow.utils.session import NEW_SESSION, provide_session
@@ -751,21 +746,8 @@ class SerializedDAG:
if not deadline_alert:
continue
- deserialized_deadline_alert = decode_deadline_alert(
- {
- Encoding.TYPE: DAT.DEADLINE_ALERT,
- Encoding.VAR: {
- DeadlineAlertFields.REFERENCE:
deadline_alert.reference,
- DeadlineAlertFields.INTERVAL: deadline_alert.interval,
- DeadlineAlertFields.CALLBACK:
deadline_alert.callback_def,
- },
- }
- )
-
- interval = deserialized_deadline_alert.interval
-
- if isinstance(interval, SerializedVariableInterval):
- interval = interval.resolve(session=session)
+ deserialized_deadline_alert =
decode_deadline_alert_model(deadline_alert)
+ interval =
resolve_deadline_alert_interval(deserialized_deadline_alert, session=session)
if isinstance(deserialized_deadline_alert.reference,
SerializedReferenceModels.TYPES.DAGRUN):
deadline_time =
deserialized_deadline_alert.reference.evaluate_with(
diff --git a/airflow-core/tests/unit/models/test_dagrun.py
b/airflow-core/tests/unit/models/test_dagrun.py
index ea36d44b599..c80a9dfc3ac 100644
--- a/airflow-core/tests/unit/models/test_dagrun.py
+++ b/airflow-core/tests/unit/models/test_dagrun.py
@@ -1596,6 +1596,24 @@ class TestDagRun:
deadline = session.execute(select(Deadline)).scalars().one_or_none()
assert deadline.deadline_time == first_deadline_time
+ @mock.patch.object(Deadline, "prune_deadlines")
+ def test_dagrun_deadline_variable_interval_uses_callers_session(self, _,
session, deadline_test_dag):
+ """Caller's session must reach Variable.get."""
+ future_date = datetime.datetime.now() + datetime.timedelta(days=365)
+
+ scheduler_dag = deadline_test_dag(
+ deadline=DeadlineAlert(
+ reference=DeadlineReference.FIXED_DATETIME(future_date),
+ interval=VariableInterval("my_key"),
+ callback=AsyncCallback(empty_callback_for_deadline),
+ ),
+ )
+
+ with mock.patch.object(Variable, "get", return_value="60") as mock_get:
+ self.create_dag_run(dag=scheduler_dag, session=session)
+
+ mock_get.assert_called_once_with("my_key", session=session)
+
def test_dagrun_deadline_logs_when_reference_column_is_null(self, session,
deadline_test_dag, caplog):
scheduler_dag = deadline_test_dag(
deadline=DeadlineAlert(
diff --git a/airflow-core/tests/unit/models/test_taskinstance.py
b/airflow-core/tests/unit/models/test_taskinstance.py
index 0820c2f7881..29f98da417a 100644
--- a/airflow-core/tests/unit/models/test_taskinstance.py
+++ b/airflow-core/tests/unit/models/test_taskinstance.py
@@ -73,6 +73,7 @@ from airflow.models.taskinstance import (
find_relevant_relatives,
)
from airflow.models.taskreschedule import TaskReschedule
+from airflow.models.variable import Variable
from airflow.models.xcom import XComModel, XComModelV2, build_xcom_read_query
from airflow.providers.standard.operators.bash import BashOperator
from airflow.providers.standard.operators.empty import EmptyOperator
@@ -92,10 +93,11 @@ from airflow.sdk import (
)
from airflow.sdk.api.datamodels._generated import AssetEventResponse,
AssetResponse
from airflow.sdk.definitions.callback import AsyncCallback
-from airflow.sdk.definitions.deadline import DeadlineReference
+from airflow.sdk.definitions.deadline import DeadlineReference,
VariableInterval
from airflow.sdk.definitions.param import process_params
from airflow.sdk.definitions.taskgroup import TaskGroup
from airflow.sdk.execution_time.comms import AssetEventsResult
+from airflow.sdk.serde import serialize
from airflow.serialization.definitions.assets import SerializedAsset
from airflow.serialization.definitions.baseoperator import
SerializedBaseOperator
from airflow.serialization.definitions.dag import SerializedDAG
@@ -4597,8 +4599,24 @@ async def empty_callback_for_deadline():
pass
-def test_clear_task_instances_recalculates_dagrun_queued_deadlines(dag_maker,
session):
- """Test that clearing tasks recalculates all (and only) DAGRUN_QUEUED_AT
deadlines."""
[email protected](
+ "use_variable_interval",
+ [
+ pytest.param(False, id="fixed_timedelta_interval"),
+ pytest.param(True, id="variable_interval"),
+ ],
+)
+def test_clear_task_instances_recalculates_dagrun_queued_deadlines(dag_maker,
session, use_variable_interval):
+ """Test that clearing tasks recalculates all (and only) DAGRUN_QUEUED_AT
deadlines.
+
+ Since Airflow 3.3 the ``deadline_alert.interval`` column is JSON (a
serialized ``timedelta``
+ or ``VariableInterval``), so the recalculation must decode it instead of
passing the raw value
+ to ``timedelta()``. Storing the interval via ``serialize`` here mirrors
production and covers
+ both interval kinds.
+ """
+ variable_key = "deadline_interval_key"
+ variable_seconds = 3600
+
with dag_maker(
dag_id="test_recalculate_deadlines",
schedule=datetime.timedelta(days=1),
@@ -4618,29 +4636,52 @@ def
test_clear_task_instances_recalculates_dagrun_queued_deadlines(dag_maker, se
select(SerializedDagModel.id).where(SerializedDagModel.dag_id ==
dag.dag_id)
)
- deadline_configs = [
- (DeadlineReference.DAGRUN_QUEUED_AT, datetime.timedelta(hours=1)),
- (DeadlineReference.DAGRUN_QUEUED_AT, datetime.timedelta(hours=2)),
- (DeadlineReference.FIXED_DATETIME, datetime.timedelta(hours=1)),
+ if use_variable_interval:
+ queued_interval = VariableInterval(variable_key)
+ queued_resolved = datetime.timedelta(seconds=variable_seconds)
+ # Both DAGRUN_QUEUED_AT deadlines resolve to the same interval from
the Variable.
+ queued_configs = [
+ (DeadlineReference.DAGRUN_QUEUED_AT, queued_interval,
queued_resolved),
+ (DeadlineReference.DAGRUN_QUEUED_AT, queued_interval,
queued_resolved),
+ ]
+ else:
+ queued_configs = [
+ (
+ DeadlineReference.DAGRUN_QUEUED_AT,
+ datetime.timedelta(hours=1),
+ datetime.timedelta(hours=1),
+ ),
+ (
+ DeadlineReference.DAGRUN_QUEUED_AT,
+ datetime.timedelta(hours=2),
+ datetime.timedelta(hours=2),
+ ),
+ ]
+
+ # (reference type, interval stored on the alert, timedelta it resolves to)
+ deadline_configs = queued_configs + [
+ (DeadlineReference.FIXED_DATETIME, datetime.timedelta(hours=1),
datetime.timedelta(hours=1)),
]
- for deadline_type, interval in deadline_configs:
+ expected_resolved_by_alert = {}
+ for deadline_type, stored_interval, resolved_interval in deadline_configs:
if deadline_type == DeadlineReference.DAGRUN_QUEUED_AT:
reference =
DeadlineReference.DAGRUN_QUEUED_AT.serialize_reference()
- deadline_time = dag_run.queued_at + interval
+ deadline_time = dag_run.queued_at + resolved_interval
else: # FIXED_DATETIME
future_date = timezone.utcnow() + datetime.timedelta(days=7)
reference =
DeadlineReference.FIXED_DATETIME(future_date).serialize_reference()
- deadline_time = future_date + interval
+ deadline_time = future_date + resolved_interval
deadline_alert = DeadlineAlertModel(
serialized_dag_id=serialized_dag_id,
reference=reference,
- interval=interval.total_seconds(),
- callback_def={"path": f"{__name__}.empty_callback_for_deadline",
"kwargs": {}},
+ interval=serialize(stored_interval),
+ callback_def=serialize(AsyncCallback(empty_callback_for_deadline)),
)
session.add(deadline_alert)
session.flush()
+ expected_resolved_by_alert[deadline_alert.id] = resolved_interval
deadline = Deadline(
dagrun_id=dag_run.id,
@@ -4659,7 +4700,21 @@ def
test_clear_task_instances_recalculates_dagrun_queued_deadlines(dag_maker, se
}
tis = session.scalars(select(TI).where(TI.dag_id == dag.dag_id, TI.run_id
== dag_run.run_id)).all()
- clear_task_instances(tis, session)
+
+ # VariableInterval.resolve() reads the Airflow Variable during
recalculation.
+ with (
+ mock.patch.object(Variable, "get", return_value=str(variable_seconds))
+ if use_variable_interval
+ else contextlib.nullcontext()
+ ) as mock_variable_get:
+ clear_task_instances(tis, session)
+
+ if use_variable_interval:
+ # See resolve_deadline_alert_interval() for why the caller's session
has to be forwarded.
+ assert mock_variable_get.call_args_list == [
+ mock.call(variable_key, session=session),
+ mock.call(variable_key, session=session),
+ ]
dag_run = session.scalar(select(DagRun).where(DagRun.id == dag_run.id))
assert dag_run.queued_at > original_queued_at
@@ -4672,13 +4727,84 @@ def
test_clear_task_instances_recalculates_dagrun_queued_deadlines(dag_maker, se
for deadline in deadlines_after:
if deadline.deadline_time !=
deadline_times_by_alert[deadline.deadline_alert_id]:
recalculated_count += 1
- deadline_alert = session.get(DeadlineAlertModel,
deadline.deadline_alert_id)
- expected_time = dag_run.queued_at +
datetime.timedelta(seconds=deadline_alert.interval)
+ expected_time = dag_run.queued_at +
expected_resolved_by_alert[deadline.deadline_alert_id]
assert deadline.deadline_time == expected_time
assert recalculated_count == 2
+def
test_clear_task_instances_skips_deadline_with_unresolvable_interval(dag_maker,
session):
+ """A variable-backed interval that cannot be resolved must not abort the
clear.
+
+ ``SerializedVariableInterval.resolve()`` raises ``ValueError`` when the
Airflow Variable is
+ missing or is not an integer, and that happens while the DAG run is being
cleared. The clear
+ should still go through, leaving the unresolvable deadline at its old time.
+ """
+ with dag_maker(
+ dag_id="test_recalculate_deadlines_unresolvable",
+ schedule=datetime.timedelta(days=1),
+ ) as dag:
+ EmptyOperator(task_id="task_1")
+
+ dag_run = dag_maker.create_dagrun()
+ ti = dag_run.get_task_instance("task_1", session=session)
+ ti.set_state(TaskInstanceState.SUCCESS, session=session)
+
+ original_queued_at = timezone.utcnow() - datetime.timedelta(hours=2)
+ dag_run.queued_at = original_queued_at
+ session.flush()
+
+ serialized_dag_id = session.scalar(
+ select(SerializedDagModel.id).where(SerializedDagModel.dag_id ==
dag.dag_id)
+ )
+
+ deadline_alert = DeadlineAlertModel(
+ serialized_dag_id=serialized_dag_id,
+ reference=DeadlineReference.DAGRUN_QUEUED_AT.serialize_reference(),
+ interval=serialize(VariableInterval("missing_deadline_interval_key")),
+ callback_def=serialize(AsyncCallback(empty_callback_for_deadline)),
+ )
+ session.add(deadline_alert)
+ session.flush()
+
+ original_deadline_time = original_queued_at + datetime.timedelta(hours=1)
+ session.add(
+ Deadline(
+ dagrun_id=dag_run.id,
+ deadline_alert_id=deadline_alert.id,
+ deadline_time=original_deadline_time,
+ callback=AsyncCallback(empty_callback_for_deadline),
+ dag_id=dag_run.dag_id,
+ )
+ )
+ session.flush()
+
+ tis = session.scalars(select(TI).where(TI.dag_id == dag.dag_id, TI.run_id
== dag_run.run_id)).all()
+
+ with (
+ mock.patch.object(
+ Variable, "get",
side_effect=KeyError("missing_deadline_interval_key")
+ ) as mock_variable_get,
+ mock.patch("airflow.models.taskinstance.log") as mock_log,
+ ):
+ clear_task_instances(tis, session)
+
+ dag_run = session.scalar(select(DagRun).where(DagRun.id == dag_run.id))
+ assert dag_run.queued_at > original_queued_at
+
+ deadline = session.scalar(select(Deadline).where(Deadline.dagrun_id ==
dag_run.id))
+ assert deadline.deadline_time == original_deadline_time
+
+ # The alert decoded fine and the Variable lookup is what raised, so this
is still covering
+ # an unresolvable interval rather than an earlier failure that lands in
the same except.
+ mock_variable_get.assert_called_once_with("missing_deadline_interval_key",
session=session)
+
+ assert mock_log.warning.call_count == 1
+ msg, *args = mock_log.warning.call_args.args
+ assert "Error while recalculating deadline" in msg
+ assert args[0] == deadline.id
+
+
def test_get_dagrun_loaded_but_none_returns_dagrun(dag_maker, session):
"""
Test that `get_dagrun()` fetches `DagRun` from DB when the `dag_run`
diff --git a/airflow-core/tests/unit/serialization/test_decoders.py
b/airflow-core/tests/unit/serialization/test_decoders.py
new file mode 100644
index 00000000000..56ab174f261
--- /dev/null
+++ b/airflow-core/tests/unit/serialization/test_decoders.py
@@ -0,0 +1,84 @@
+# 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.
+
+from __future__ import annotations
+
+import datetime
+
+from airflow.models.deadline_alert import DeadlineAlert as DeadlineAlertModel
+from airflow.models.variable import Variable
+from airflow.sdk.definitions.callback import AsyncCallback
+from airflow.sdk.definitions.deadline import DeadlineReference
+from airflow.sdk.serde import serialize
+from airflow.serialization.decoders import decode_deadline_alert_model,
resolve_deadline_alert_interval
+from airflow.serialization.definitions.deadline import (
+ SerializedDeadlineAlert,
+ SerializedReferenceModels,
+ SerializedVariableInterval,
+)
+
+
+async def empty_callback_for_deadline():
+ pass
+
+
+def _alert(interval):
+ return SerializedDeadlineAlert(
+ reference=SerializedReferenceModels.DagRunQueuedAtDeadline(),
+ interval=interval,
+ callback=None,
+ )
+
+
+class TestResolveDeadlineAlertInterval:
+ def test_forwards_session_to_variable_lookup(self, mocker):
+ """Both callers resolve intervals while holding an open transaction,
so the session they
+ pass has to reach ``Variable.get``. See
``resolve_deadline_alert_interval()``."""
+ mock_get = mocker.patch.object(Variable, "get", return_value="42")
+ session = mocker.MagicMock()
+
+ resolved = resolve_deadline_alert_interval(
+ _alert(SerializedVariableInterval(key="test_interval")),
session=session
+ )
+
+ assert resolved == datetime.timedelta(seconds=42)
+ mock_get.assert_called_once_with("test_interval", session=session)
+
+ def test_fixed_interval_is_returned_without_a_variable_lookup(self,
mocker):
+ mock_get = mocker.patch.object(Variable, "get")
+
+ interval = datetime.timedelta(hours=1)
+
+ assert resolve_deadline_alert_interval(_alert(interval),
session=mocker.MagicMock()) == interval
+ mock_get.assert_not_called()
+
+
+class TestDecodeDeadlineAlertModel:
+ def test_row_name_is_carried_into_the_serialized_alert(self):
+ """The ORM row has a ``name`` column, so a general "decode this row"
helper has to read it.
+
+ Neither caller looks at ``.name`` today, which is exactly why dropping
it here would go
+ unnoticed until a third caller trusts the helper's name.
+ """
+ row = DeadlineAlertModel(
+ name="my_deadline",
+ reference=DeadlineReference.DAGRUN_QUEUED_AT.serialize_reference(),
+ interval=serialize(datetime.timedelta(hours=1)),
+ callback_def=serialize(AsyncCallback(empty_callback_for_deadline)),
+ )
+
+ assert decode_deadline_alert_model(row).name == "my_deadline"
