amoghrajesh commented on code in PR #70370:
URL: https://github.com/apache/airflow/pull/70370#discussion_r4070061048
##########
airflow-core/src/airflow/serialization/decoders.py:
##########
@@ -338,6 +341,43 @@ 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.REFERENCE: deadline_alert.reference,
+ DeadlineAlertFields.INTERVAL: deadline_alert.interval,
+ DeadlineAlertFields.CALLBACK: deadline_alert.callback_def,
Review Comment:
This helper is named as a general "decode a DeadlineAlert ORM row", but it
skips the row's `name`. The model has a `name` column and
`decode_deadline_alert` reads it (line 337 just above), so
`SerializedDeadlineAlert.name` comes back as None even when the row has one.
Nothing reads `.name` on either of your two callers, so this isn't a bug as
of today but its a trap for the next caller. Worth adding:
```
DeadlineAlertFields.NAME: deadline_alert.name,
```
##########
airflow-core/tests/unit/models/test_taskinstance.py:
##########
@@ -4130,8 +4130,28 @@ 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.
+ """
+ from airflow.models.variable import Variable
+ from airflow.sdk.definitions.deadline import VariableInterval
+ from airflow.sdk.serde import serialize
Review Comment:
Top level import pls.
##########
airflow-core/src/airflow/serialization/decoders.py:
##########
@@ -338,6 +341,43 @@ 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.REFERENCE: deadline_alert.reference,
+ DeadlineAlertFields.INTERVAL: deadline_alert.interval,
+ DeadlineAlertFields.CALLBACK: deadline_alert.callback_def,
+ }
+ )
+
+
+def resolve_deadline_alert_interval(
+ alert: SerializedDeadlineAlert, *, session: Session | None = None
+) -> 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. 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.
Review Comment:
Small thing: this session rationale is written out four times - here, and
again in test_dagrun.py:1568, test_taskinstance.py:4248, and
test_decoders.py:41. Keep it here as the source of truth and let the tests
point at it
##########
airflow-core/tests/unit/models/test_taskinstance.py:
##########
@@ -4205,13 +4263,88 @@ 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.
+ """
+ from airflow.models.variable import Variable
+ from airflow.sdk.definitions.deadline import VariableInterval
+ from airflow.sdk.serde import serialize
Review Comment:
Top level import pls.
##########
airflow-core/src/airflow/serialization/decoders.py:
##########
@@ -338,6 +341,43 @@ 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.REFERENCE: deadline_alert.reference,
+ DeadlineAlertFields.INTERVAL: deadline_alert.interval,
+ DeadlineAlertFields.CALLBACK: deadline_alert.callback_def,
+ }
+ )
+
+
+def resolve_deadline_alert_interval(
+ alert: SerializedDeadlineAlert, *, session: Session | None = None
Review Comment:
The signature makes `session` optional, but the docstring right below spends
three lines explaining that a caller who omits it gets its session rolled back
under `prohibit_commit`. Both callers bypass it.
Can this be `*, session: Session` with no default? That turns a silent
misuse into a TypeError at the caller, which is the whole point of the three
tests you added for this.
--
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]
