aeroyorch commented on code in PR #65932:
URL: https://github.com/apache/airflow/pull/65932#discussion_r4007103027
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -2202,6 +2202,17 @@ def schedule_tis(
debug_try_number_check = self.log.isEnabledFor(logging.DEBUG)
expected_try_number_by_ti_id: dict[UUID, tuple[int, int, str | None]]
= {}
for ti in schedulable_tis:
+ if ti.state == TaskInstanceState.UP_FOR_RETRY:
+ if TYPE_CHECKING:
+ assert ti.task
+ # Weight strategies and the mutation hook must see the
upcoming try. The updates below
+ # still own the increment, so try_number is restored.
+ try_number = ti.try_number
+ ti.try_number = try_number + 1
Review Comment:
Thanks for the review! I tried to find a way to keep this consistent, and I
see two options:
- Leave `try_number ` as the failed attempt, as you suggest, and clarify
this in the cluster policy docs.
- Refresh after the guarded bulk `UPDATE` that moves the task instance to
`SCHEDULED`. We read back `try_number` for the retries that are now `SCHEDULED`
(an extra `SELECT`, only when a batch has retries), load it, and then call
`refresh_from_task`. Then nothing is set temporarily, and the hook sees the
real next attempt.
I am not sure which of these options fits better for you, or if you have a
better idea in mind.
--
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]
