seanghaeli commented on code in PR #70714:
URL: https://github.com/apache/airflow/pull/70714#discussion_r4066784820
##########
airflow-core/src/airflow/models/taskinstance.py:
##########
@@ -249,9 +248,27 @@ 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
+ 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 not isinstance(interval, timedelta):
+ interval = interval.resolve()
Review Comment:
Looks good and I think it's mostly ready to merge, there's just one more
thing that needs to be taken care of since new code has merged into main.
#71968 added a session argument to `SerializedVariableInterval.resolve()`, so
your `interval.resolve()` must become `interval.resolve(session=session)`.
Your call site runs inside `clear_task_instances`, which is midway through a
batch of database changes it saves all at once at the end. Without the session,
the Variable lookup opens the same connection and saves that half-finished
batch early. #71968 already converted the other call site to pass it; this new
one should follow.
--
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]
