hkc-8010 commented on code in PR #70370:
URL: https://github.com/apache/airflow/pull/70370#discussion_r4065987948
##########
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:
Took your second option. It now reads "Error while recalculating deadline %s
for DagRun %s.%s, leaving it unchanged" and the cause is left to exc_info, and
I updated the comment above the try to match.
You are right that the decoder trips it, and it is not hypothetical. On the
last rebase #72651 made callback decoding strict, the fixtures' hand-written
callback_def stopped being valid, and this message reported an unresolvable
interval while the interval was fine.
##########
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:
Done, with one addition. Your message and deadline.id assertions are in,
reworded for the new log line from the other thread, plus:
```python
mock_variable_get.assert_called_once_with("missing_deadline_interval_key",
session=session)
```
That is the one with teeth, and it comes out of your other comment. Since
the try covers decode too, this test passes when decode fails before the
interval is ever read, which is what it was actually doing right after the
#72651 rebase. I checked by putting the old raw-dict callback_def back: it now
fails with "Expected 'get' to be called once. Called 0 times." where before it
stayed green.
--
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]
