kaxil commented on code in PR #72100: URL: https://github.com/apache/airflow/pull/72100#discussion_r3979115182
########## 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. +* Clearing a *failed* task never runs ``on_kill`` at all, so an external job that outlived the + worker is still running. Review Comment: `task-state-store.rst:288` covers it now, thanks, but a few things on that sentence. The deferred case is still missing from this page itself: the three bullets are all about the worker's `on_kill`, which is the one thing that never runs for a deferred task, because there is no worker process to signal. What cancels the job is the *trigger's* `on_kill` -- an orphaned trigger goes into `cancelling_triggers` (`triggerer_job_runner.py:999`), `cancel_triggers` cancels it with `_USER_ACTION_CANCEL_MSG` (`:1456`), and `run_trigger` then awaits `trigger.on_kill()` (`:1706`). That is also why the new sentence names both actors: it credits "durable operators" for the cancellation but lists the exceptions by trigger, and the trigger is the correct half. Two smaller things on the same line. "before the next attempt starts" is an ordering guarantee the code does not make: `clear_task_instances` sets `ti.state = None` (`models/taskinstance.py:444`) and nothing in the clear path touches the triggerer, so the scheduler can queue the next attempt while the triggerer is still working through `cancel_triggers` under a 30s `_ON_CANCEL_TIMEOUT` (`:154`, `:1706`). And the exception list is right for the shipped durable operators -- Glue and Livy are the only two of the seven whose triggers don't override `BaseTrigger.on_kill` -- but it reads as a statement about triggers in general, so an author with their own deferrable operator would take it as covering them. ########## airflow-core/newsfragments/72100.significant.rst: ########## @@ -0,0 +1,39 @@ +Clearing a task now discards its task state store entries by default + +Clearing a task instance discards its ``task_state_store`` entries, so the next attempt starts from +the beginning instead of resuming from a checkpoint or reconnecting to an external job recorded by +the attempt that was cleared. + +Retries are unaffected. They keep task state exactly as before, which is what crash recovery relies +on. Only a deliberate clear discards. + +**Why** + +Clearing means "run this again". A checkpoint records how far a task got, not what it got there +with, so resuming after the code or the upstream data changed left work done before the fix in place +and silently mixed it with the corrected work. Clearing a task whose external job had already +succeeded was worse: the operator read the stored result back and returned in seconds having run +nothing. + +**Keeping the old behaviour** + +Pass ``keep_task_state=True`` to the clear task instances endpoint, or tick "keep task state" in the +clear dialog. Use it when nothing about the inputs or the code changed and the task should carry on +where it stopped, or when an external job is still running and you want the next attempt to +reconnect rather than submit a duplicate. Review Comment: On taking the flag back off `airflowctl dags clear`: I have left that inline on the `keep_task_state` field in `generated.py`, since the command still posts this body and now discards with no opt-out. Separately, "deprecated" overstates the marker. `deprecated_for_airflowctl`'s own docstring says the command "stays in the ``airflow`` CLI as a supported entry point" and "emits no user-facing deprecation warning at runtime", recorded "for maintainers only". So from a user's side `airflow tasks clear` is live and warning-free, and it now silently keeps task state while the UI, the API and `airflowctl tasks clear` discard it. Worth saying that plainly rather than leaning on "deprecated", which reads as "you shouldn't be using it anyway". One more on discoverability: `core-concepts/dag-run.rst:233` "Re-run Tasks" is the page that actually teaches clearing, and it is untouched. It enumerates the clear dialog options with no "Keep task state and resume", never mentions the new default, and at `:258` hands the reader `airflow tasks clear`, one of the paths that keeps state. Outside the two rewritten pages and the generated REST reference the flag has nowhere to be found. -- 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]
