ashb commented on code in PR #65932:
URL: https://github.com/apache/airflow/pull/65932#discussion_r4017790214
##########
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:
Okay, I've dug in to this and found that things are... a bit messier than
they should be.
| Point | Live TI UUID | Live try_number | Live state | TIH |
|---|---|---|---|---|
| Attempt executing | A | N | RUNNING | — |
| Retry transition commits | B | N | UP_FOR_RETRY | A, try N archived |
| Waiting through retry delay | B | N | UP_FOR_RETRY | A, try N |
| Scheduler schedules retry | B | N+1 | SCHEDULED | A, try N |
| Executor queues retry | B | N+1 | QUEUED | A, try N |
| Replacement worker starts | B | N+1 | RUNNING | A, try N |
I want to change things so rows (B,N) don't ever exist.
--
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]
