aeroyorch commented on code in PR #65932:
URL: https://github.com/apache/airflow/pull/65932#discussion_r3188418143
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -1993,16 +2003,30 @@ def schedule_tis(
# execute it to run in the worker.
elif not ti.defer_task(session=session):
schedulable_ti_ids.append(ti.id)
- if ti.state == TaskInstanceState.UP_FOR_RESCHEDULE:
+ is_reschedule = ti.state == TaskInstanceState.UP_FOR_RESCHEDULE
+ if is_reschedule:
reschedule_ti_ids.add(ti.id)
if debug_try_number_check:
expected_try_number_by_ti_id[ti.id] = (
- ti.try_number
- if ti.state == TaskInstanceState.UP_FOR_RESCHEDULE
- else ti.try_number + 1,
+ ti.try_number if is_reschedule else ti.try_number + 1,
ti.try_number,
ti.state,
)
+ # Retry-only: first attempts use creation-time hook firing and
+ # reschedules do not bump try_number.
+ if ti.task is not None and ti.state ==
TaskInstanceState.UP_FOR_RETRY:
+ saved_try = ti.try_number
+ saved_values = tuple(getattr(ti, k) for k in
_RETRY_REFRESHED_TI_FIELDS)
+ ti.try_number = saved_try + 1
+ ti.refresh_from_task(ti.task)
+ ti.try_number = saved_try
+ overrides = {
+ k: getattr(ti, k)
+ for k, prev in zip(_RETRY_REFRESHED_TI_FIELDS,
saved_values)
+ if getattr(ti, k) != prev
+ }
+ if overrides:
+ mutated_overrides_by_ti_id[ti.id] = overrides
Review Comment:
I was trying to be consistent with the `schedulable_ti_ids` path, but you're
right, ORM unit-of-work is simpler. Already committed the change
##########
airflow-core/docs/administration-and-deployment/cluster-policies.rst:
##########
@@ -37,10 +37,12 @@ There are three main types of cluster policy:
task running in a DagRun. The ``task_policy`` defined is applied to all the
task instances that will be
executed in the future.
* ``task_instance_mutation_hook``: Takes a
:class:`~airflow.models.taskinstance.TaskInstance` parameter called
- ``task_instance``. The ``task_instance_mutation_hook`` applies not to a task
but to the instance of a task that
- relates to a particular DagRun. It is executed in a "worker", not in the Dag
file processor, just before the
- task instance is executed. The policy is only applied to the currently
executed run (i.e. instance) of that
- task.
+ ``task_instance``. The ``task_instance_mutation_hook`` applies not to a task
but to the instance of a task
+ that relates to a particular DagRun. It is executed scheduler-side, before
the task instance is dispatched
+ to an executor; the worker does not invoke it. The hook may fire more than
once for the same task instance
+ (on creation, on mapped expansion, on manual clear, and on natural retry
transitions), so implementations
+ should be idempotent. Mutated ``TaskInstance`` fields are persisted and
become the values used for dispatch
Review Comment:
rewrote the paragraph
--
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]
