ferruzzi commented on code in PR #70370:
URL: https://github.com/apache/airflow/pull/70370#discussion_r4065632327
##########
airflow-core/tests/unit/models/test_taskinstance.py:
##########
@@ -4205,13 +4263,78 @@ 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
+
+ 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")),
+ 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
+ assert mock_log.warning.call_count == 1
Review Comment:
Non-blocking thought: More than just "something threw an error", we can
(should?) check that it's the error we expected (not some random breaking
change later), and that it's _our_ deadline that threw it. That last one maybe
overkill, but doesn't hurt?
```suggestion
assert mock_log.warning.call_count == 1
msg, *args = mock_log.warning.call_args.args
assert "interval could not be resolved" in msg
assert args[0] == deadline.id
```
##########
airflow-core/src/airflow/models/taskinstance.py:
##########
@@ -248,10 +248,29 @@ 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):
+ # A variable-backed interval resolves against an Airflow Variable
that may be missing
+ # or non-numeric. Leave this deadline alone rather than failing
the whole clear.
+ log.warning(
Review Comment:
Non-blocking thought: Only worth it if you are making other changes
already, but the message might be a little misleading. The `try` is covering
decode and resolve, so if the decoder refuses the callback, I'm pretty sure
that ends up tripping this message too.
You have `exc_info=True`, so the actual stack trace gets bubbled up and
they'll see the real cause, so it's not critical. Something like "deadline
alert could not be decoded" or drop the exact cause entirely and say "Error
while recalculating deadline" might be more accurate than "interval could not
be resolved". (and the same update to the comment just before the `try` would
be appropriate)
--
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]
