yuseok89 commented on code in PR #76682:
URL: https://github.com/apache/airflow/pull/76682#discussion_r4526883447
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -1638,6 +1638,28 @@ def _build_finished_tis(dag: SerializedDAG, rows) ->
list[FinishedTI]:
unfinished_tis = list(_filter_tis_and_exclude_removed(dag,
unfinished_tis))
finished_tis: list[SchedulingTI] = list(_build_finished_tis(dag,
finished_rows))
+ retry_timeout_expired = False
+ now = timezone.utcnow()
+ for ti in unfinished_tis:
+ if ti.state not in {
+ TaskInstanceState.UP_FOR_RETRY,
+ TaskInstanceState.SCHEDULED,
+ TaskInstanceState.QUEUED,
+ }:
+ continue
+ if ti.retry_timeout_deadline is None:
+ if ti.task.retry_timeout is None:
+ continue
+ failed_at = ti.end_date
+ if failed_at is None and ti.try_number > 1:
+ failed_at =
ti._get_previous_attempt_end_date(session=session)
+ if failed_at is None:
+ continue
+ ti.retry_timeout_deadline = failed_at + ti.task.retry_timeout
+ if ti.retry_timeout_deadline <= now:
+ ti.set_state(TaskInstanceState.FAILED, session=session)
Review Comment:
Would it be worth adding a test for a task with an `on_failure_callback`
whose deadline expires here?
--
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]
