ferruzzi commented on code in PR #70714:
URL: https://github.com/apache/airflow/pull/70714#discussion_r3817482939
##########
airflow-core/src/airflow/models/taskinstance.py:
##########
@@ -249,9 +247,16 @@ def _recalculate_dagrun_queued_at_deadlines(
return
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
+ new_deadline_time =
decode_deadline_reference(deadline_alert.reference).evaluate_with(
+ session=session,
+ interval=timedelta(seconds=deadline_alert.interval),
Review Comment:
deadline_alert.interval is a json value, you need to be deserializing the
deadline. Which also means your tests are hiding something if this is passing.
You also have to account for VariableIntervals now.
You'll need to do something like (untested code)
```python
for deadline, deadline_alert in results:
decoded_alert = decode_deadline_alert(
{
DeadlineAlertFields.REFERENCE: deadline_alert.reference,
DeadlineAlertFields.INTERVAL: deadline_alert.interval,
DeadlineAlertFields.CALLBACK: deadline_alert.callback_def,
}
)
interval = decoded_alert.interval
if isinstance(interval, VariableInterval):
interval = interval.resolve()
new_deadline_time = decoded_alert.reference.evaluate_with(
session=session,
interval=interval,
dagrun=dagrun,
dag_id=dagrun.dag_id,
run_id=dagrun.run_id,
)
```
--
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]
