kaxil commented on code in PR #72100:
URL: https://github.com/apache/airflow/pull/72100#discussion_r3979102917
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py:
##########
@@ -986,6 +987,10 @@ def _collect_relatives(run_id: str, direction:
Literal["upstream", "downstream"]
except AirflowClearRunningTaskException as e:
raise HTTPException(status.HTTP_409_CONFLICT, str(e)) from e
+ # After the clear has succeeded, so a failed clear cannot take the
task state with it.
+ if not body.keep_task_state:
+ _discard_task_state_store(task_instances, session,
event="Discarded task state on clear")
Review Comment:
Following on from the doc scoping in the earlier round, this stays the only
call site, so the new default covers the task-level clear and nothing else.
`perform_clear_dag_run` goes straight to `dag.clear()`
(`services/public/dag_run.py:147`) and on to `clear_task_instances`
(`serialization/definitions/dag.py:1349`), so the Clear Run dialog keeps state,
and so do mark-as-success/failed downstream clearing (`dag.py:873`, `:978`),
`airflow dags clear` and `airflow tasks clear`. Clearing a whole run after
fixing Dag code is the scenario the newsfragment leads with, and it is the one
still resuming from the stale checkpoint. `resumable-tasks.rst:158` also tells
the reader to "tick the corresponding box in the clear dialog" without saying
the run dialog has no box. Threading `keep_task_state` through
`clear_task_instances()` would make every path converge; if that is
deliberately out of scope, the release note should say which surfaces actually
changed.
##########
airflow-core/src/airflow/ui/src/components/Clear/TaskInstance/ClearGroupTaskInstanceDialog.tsx:
##########
@@ -67,6 +67,7 @@ export const ClearGroupTaskInstanceDialog = ({ onClose, open,
taskInstance }: Pr
const future = selectedOptions.includes("future");
const upstream = selectedOptions.includes("upstream");
const downstream = selectedOptions.includes("downstream");
+ const [keepTaskState, setKeepTaskState] = useState(false);
Review Comment:
The reset you added to `ClearTaskInstanceDialog` didn't reach this dialog.
`ClearTaskInstanceButton.tsx:97` renders it gated on `isGroup`, not `open`, so
the component stays mounted and nothing ever puts `keepTaskState` back to
false: a tick survives close, reopen, and a move to a different task group, and
the next clear quietly resumes from a checkpoint. This one has no
`useEffect`/`onCloseDialog` pair at all (it doesn't reset `note` either), so it
needs the reset wired on both open and close rather than the one-line addition
that worked next door.
##########
airflow-core/docs/core-concepts/resumable-tasks.rst:
##########
@@ -145,26 +145,42 @@ existing job on retry instead of submitting a new one.
For more details and a working example, see
:class:`~airflow.sdk.ResumableJobMixin`.
-**Clearing a task is treated the same as a retry**
-
-Clearing a task instance does not delete its ``task_state_store`` rows -- they
are only removed
-when the ``dag_run`` itself is deleted, or by :ref:`airflow state-store clean
-<task-and-asset-state-store-cleanup>`. For a checkpointed task this is usually
what you want:
-clearing resumes from the last checkpoint rather than starting over.
-
-For an operator with durable execution, it means clearing a task whose
external job already
-succeeded reads that stored result back and returns immediately, without
resubmitting the job. If
-you want clearing to always resubmit regardless of a prior success, set
-``[state_store] clear_on_success = True``, which deletes a task's state store
rows automatically
-when it moves to ``SUCCESS`` (see
:doc:`/administration-and-deployment/task-and-asset-state-store`).
-
-This does not guarantee the external job is still there to reconnect to,
though. Clearing a task
-that is actively running (``deferrable=False``) stops the worker process,
which runs the
-operator's ``on_kill``. Most operators with durable execution cancel the
external job there by
-default, so the next attempt finds it already stopped instead of still running
-- an operator that
-leaves the job running by default on kill is the exception, check its own
docs. Deferred tasks
-(``deferrable=True``) don't have this problem: there is no actively polling
worker process for the
-clear to interrupt.
+**Retries resume, clearing starts over**
+
+A retry keeps the task's ``task_state_store`` entries, which is what makes
crash recovery work: the
+next attempt reads the checkpoint or the external job id written by the
attempt before it.
+
+Clearing a task discards them. Clearing means "run this again", and a
checkpoint records how far a
+task got, not what it got there with. If you fixed the code or the upstream
data and cleared the
+task, resuming would leave the work done before the fix in place and silently
mix it with the
+corrected work. So by default a cleared task starts from the beginning.
+
+To resume from the checkpoint instead, set ``keep_task_state`` when clearing,
or tick the
+corresponding box in the clear dialog. That is the right choice when nothing
about the inputs or the
+code changed and you only want the task to carry on where it stopped.
+
+**Clearing a task that submitted an external job**
+
+For an operator with durable execution the stored value is an external job id,
so discarding it has
+a different consequence: the next attempt submits a new job rather than
reconnecting to the existing
+one.
+
+Whether that matters depends on what happened to the job:
+
+* Most operators cancel the external job in ``on_kill``, so clearing a
*running* task stops the job
+ and there is nothing left to reconnect to. Submitting a fresh one is the
only option anyway.
+* An operator configured to leave the job running on kill (for example
+ ``KubernetesPodOperator`` with ``on_kill_action="keep_pod"``) keeps it
alive, so a fresh submission
+ runs alongside it. Check the operator's own docs.
Review Comment:
`GlueJobOperator` belongs on this bullet rather than the one above it.
`stop_job_run_on_kill` defaults to `False` (`glue.py:205`, assigned `:262`) and
`on_kill` is a no-op unless it is set (`:386-393`), so clearing a running Glue
task leaves the AWS run going while this PR discards the run id that would have
let the next attempt reconnect. As written the page routes Glue into "nothing
left to reconnect to" and the reader pays for a second job. Worth having the
bullet say that cancel-on-kill being off by default is a thing to check, not
only an opt-in like `keep_pod`, since Glue is one of the operators that
actually writes `task_state_store`.
##########
airflow-ctl/src/airflowctl/api/datamodels/generated.py:
##########
@@ -300,6 +300,13 @@ class ClearTaskInstancesBody(BaseModel):
),
] = None
prevent_running_task: Annotated[bool | None, Field(title="Prevent Running
Task")] = False
+ keep_task_state: Annotated[
Review Comment:
The field is on the model now, but `airflowctl dags clear` has no way to set
it. That command hand-builds a `ClearTaskInstancesBody` at
`commands/dag_command.py:317-326` and posts it to `/clearTaskInstances`, so it
discards task state, and the last commit took `--keep-task-state` back off it
(8 lines out of `cli_config.py`, 1 out of `dag_command.py`, plus its tests).
Unlike `tasks clear`, its arg list in `DAG_COMMANDS` is written by hand rather
than derived from this datamodel, so it will not pick the flag up on its own.
That leaves `airflowctl dags clear` discarding with no opt-out, which is a
regression against the version earlier in this PR. Worth putting the arg back
and naming it in the newsfragment next to `airflowctl tasks clear`.
##########
airflow-core/src/airflow/ui/src/pages/TaskInstances/BulkClearTaskInstancesButton.tsx:
##########
@@ -42,6 +42,7 @@ const BulkClearTaskInstancesButton = ({ clearSelections,
selectedTaskInstances }
const { onClose, onOpen, open } = useDisclosure();
const [selectedOptions, setSelectedOptions] =
useState<Array<string>>(["downstream"]);
const [note, setNote] = useState<string | null>(null);
+ const [keepTaskState, setKeepTaskState] = useState(false);
Review Comment:
Same stale tick here: `handleClose` resets `note` but not `keepTaskState`,
and this button sits in the Task Instances toolbar so it never unmounts. Select
a different set of task instances after one keep-state clear and the box is
still ticked. While you're in this block, the two checkboxes end up in opposite
orders between the dialogs: this Flex renders `keep, prevent`, while
`ClearTaskInstanceDialog`'s footer is `row-reverse` (`Modal.tsx:73`) so the
same pair reads `prevent, keep` there.
--
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]
