amoghrajesh commented on code in PR #72100: URL: https://github.com/apache/airflow/pull/72100#discussion_r4142231240
########## airflow-core/docs/core-concepts/resumable-tasks.rst: ########## @@ -145,26 +145,59 @@ 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. + +This applies to clearing individual task instances through the REST API, the UI, or +``airflowctl dags clear`` (which clears every task instance in the matched Dag run(s) through this +same endpoint). Clearing an entire Dag run through the Clear Run dialog/API, marking a task as +failed or success (which clears downstream tasks as a side effect), and the core CLI's +``airflow tasks clear`` / ``airflow dags clear`` (which call ``clear_task_instances()`` directly and +never reach the discard endpoint) all still keep task state unconditionally today; see +`#72929 <https://github.com/apache/airflow/issues/72929>`_. + +**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, or defaulting, to leave the job running on kill keeps it alive, so a + fresh submission runs alongside it. Check the operator's own docs — for example + ``GlueJobOperator`` defaults ``stop_job_run_on_kill`` to ``False`` and so leaves the job running + unless you turn it on. +* Clearing a *failed* task never runs ``on_kill`` at all, so an external job that outlived the + worker is still running. +* A deferred task has no worker process to run ``on_kill`` on. Instead, the Triggerer cancels the + orphaned trigger and runs the trigger's ``on_kill``, bounded by ``[triggerer] on_kill_timeout``. + Most triggers cancel the external job there too. + ``GlueJobCompleteTrigger`` and ``LivyTrigger`` don't implement ``on_kill``, but neither writes a + job id to the state store when deferred either, so ``keep_task_state`` will not help for them. For + a deferred ``GlueJobOperator``, set ``durable=True`` so the next attempt reattaches regardless of Review Comment: [final review from kaxil](https://github.com/apache/airflow/pull/72100/commits/ce0ba2f98cc9ea3636166c728b7e0b95efbe4e05) ########## airflow-ctl/src/airflowctl/api/datamodels/generated.py: ########## @@ -297,6 +297,13 @@ class ClearTaskInstancesBody(BaseModel): ), ] = None prevent_running_task: Annotated[bool | None, Field(title="Prevent Running Task")] = False + keep_task_state: Annotated[ Review Comment: [final review from kaxil](https://github.com/apache/airflow/pull/72100/commits/ce0ba2f98cc9ea3636166c728b7e0b95efbe4e05) -- 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]
