amoghrajesh commented on code in PR #72100: URL: https://github.com/apache/airflow/pull/72100#discussion_r3966392320
########## 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 discards them. Clearing means "run this again", and a checkpoint records how far a task Review Comment: Handled in [handling review comments from kaxil](https://github.com/apache/airflow/pull/72100/commits/d13e6c3553b9c21056ba392cfbb67a6b67563e19) ########## airflow-core/src/airflow/api_fastapi/core_api/services/public/task_instances.py: ########## @@ -59,10 +59,16 @@ log = structlog.get_logger(__name__) -def _clear_task_state_store_on_success(tis: Sequence[TI], session: Session) -> None: - """Clear task state store rows for each TI if clear_on_success is enabled.""" - if not conf.getboolean("state_store", "clear_on_success", fallback=False): - return +def _discard_task_state_store(tis: Sequence[TI], session: Session, *, event: str) -> None: Review Comment: Handled in [handling review comments from kaxil](https://github.com/apache/airflow/pull/72100/commits/d13e6c3553b9c21056ba392cfbb67a6b67563e19) with documentation -- 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]
