kaxil commented on code in PR #72100: URL: https://github.com/apache/airflow/pull/72100#discussion_r4008056932
########## airflow-core/docs/core-concepts/resumable-tasks.rst: ########## @@ -145,26 +145,48 @@ 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. Clearing an entire Dag run, and marking a task +as failed or success (which clears downstream tasks as a side effect), 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 Review Comment: Still open from #discussion_r3979115182: deferred tasks have dropped off this page entirely. The text this replaced closed with a sentence about `deferrable=True`, and all three bullets here are 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 there 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`). `GlueJobCompleteTrigger` and `LivyTrigger` don't implement it, which is exactly the pair `task-state-store.rst:288` names, so as written this page gives an author of a deferrable durable operator no way to work out which bullet they are in. ########## airflow-core/newsfragments/72100.significant.rst: ########## @@ -0,0 +1,52 @@ +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. + +This only applies to clearing individual task instances (the task-instance clear endpoint / dialog, +and ``airflowctl dags clear``, which clears every task instance in the matched Dag run(s) through +the same endpoint). Clearing an entire Dag run through the "Clear Run" dialog/API, and marking a +task as failed or success (which clears downstream tasks as a side effect), still keep task state +unconditionally today; extending discard-by-default to those paths is tracked in +`#72929 <https://github.com/apache/airflow/issues/72929>`_. + +**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. + +Operators with durable execution are worth particular attention. Clearing a *failed* task never runs +``on_kill``, so an external job that outlived its worker is still running, and discarding the stored +id means submitting a second one. The same applies to operators configured to leave their job alive +on kill, such as ``KubernetesPodOperator`` with ``on_kill_action="keep_pod"``. + +On the CLI, ``airflowctl dags clear`` discards task state the same way, since it clears every task +instance in the matched Dag run(s) through the same endpoint, but it has no ``--keep-task-state`` +equivalent yet to opt back in. ``airflow dags clear`` and ``airflow tasks clear`` (airflow-core, Review Comment: `airflow dags clear` carries no deprecation marker at all: `dag_clear` at `cli/commands/dag_command.py:134` has only `action_cli` / `providers_configuration_loaded` / `provide_session`, while the commands around it that do have an airflowctl equivalent (`trigger` :83, `delete` :111, `pause` :244, `list` :537) each carry `@deprecated_for_airflowctl`. `airflow tasks clear` does, so the claim is right for one of the two. This rewrite also deleted the sentence that named `airflowctl tasks clear --keep-task-state`, and that flag ships as of this PR. `cli_config.py:705-735` emits one `--flag` per `ClearTaskInstancesBody.model_fields` entry with `BooleanOptionalAction` for bools, and `:519` lists the model in `field_bool_default_datamodels` so the default follows the field, so adding `keep_task_state` created the flag on its own. `airflowctl tasks clear --help` prints `--keep-task-state, --no-keep-task-state` at this HEAD. As it now reads, the paragraph says only that `airflowctl dags clear` "has no `--keep-task-state` equivalent yet", so a reader of the 3.4.0 release notes concludes the CLI has no opt-out at all, when the command that maps 1:1 to this endpoint has exactly that. Putting that sentence back is the fix, and it is a different thing from the `dags clear` flag you decided against. Still open from the earlier thread: the core docs footprint is thinner than a `significant` change wants. `core-concepts/dag-run.rst:233` "Re-run Tasks" is the page that actually teaches clearing: it still enumerates the 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. `task-state-store.rst` carries the change in a single line inside its *Deferrable tasks* subsection, while its own lifecycle section at `:304`, "Automatic cleanup (`clear_on_success`)", and the `/administration-and-deployment/task-and-asset-state-store` page it links to at `:313` both still present `clear_on_success` and retention as the ways entries get removed. This also makes `PUT` on a task instance delete task-state rows that the dedicated endpoint gates behind `DELETE` (`routes/public/task_state_store.py:277` vs `routes/public/task_instances.py:862`). `SimpleAuthManager` does not distinguish the two (`simple_auth_manager.py:276-282`), and clearing already deletes `TaskReschedule` rows under the same permission, so this is defensible, but it is a permission-surface change worth a line rather than leaving a FAB operator to find it. ########## airflow-core/docs/core-concepts/resumable-tasks.rst: ########## @@ -145,26 +145,48 @@ 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. Clearing an entire Dag run, and marking a task Review Comment: This paragraph and the newsfragment disagree about `airflowctl dags clear`. The newsfragment carves it out as discarding, correctly -- it builds a `ClearTaskInstancesBody` per matched run and posts it to the task-instance endpoint (`airflowctl/ctl/commands/dag_command.py:317-326`) -- but a CLI user reading "Clearing an entire Dag run ... still keep task state unconditionally today" here will conclude the opposite. This page is where they will look once the release note has scrolled past, and `airflowctl dags clear` is the one clear path that discards with no way to opt back in, so it is the one most worth naming rather than the one left out. ########## airflow-core/docs/core-concepts/task-state-store.rst: ########## @@ -285,7 +285,7 @@ If the worker process crashes, the task instance is retried. Task store data wri Deferrable tasks ~~~~~~~~~~~~~~~~ -Once a task defers, the Triggerer handles continuity across poke cycles. Use task state store in deferrable tasks only when you need to survive an operator-initiated clear, not for normal poke continuity. +Once a task defers, the Triggerer handles continuity across poke cycles, and a cleared task's trigger is cancelled via ``on_kill`` before the next attempt starts. Most durable operators implement ``on_kill`` to cancel the external job there too, so the next attempt finds nothing left to reconnect to either way. The state store still matters for the small set of triggers that don't implement ``on_kill`` (for example ``GlueJobCompleteTrigger`` and ``LivyTrigger``): for those, keep task state (``keep_task_state``) when clearing so the next attempt reconnects to the job still running instead of submitting a duplicate. Review Comment: Also still open from #discussion_r3979115182. "before the next attempt starts" is an ordering guarantee the code does not make. Clearing sets `ti.state = None` (`models/taskinstance.py:444`) and nothing in the clear path touches the triggerer. `run_once` calls `load_triggers()` before `clean_unused()` (`triggerer_job_runner.py:711`, `:717`) and `ids_for_triggerer` selects on `Trigger.triggerer_id` without filtering on task-instance state (`models/trigger.py:352`), so the trigger only drops out of `requested_trigger_ids` on the *following* iteration, and `on_kill` is then bounded by `[triggerer] on_kill_timeout` (30s). Meanwhile the scheduler sees `state = None` and enqueues, and no dep check consults `trigger_id`. So a new attempt can start while the previous external job is still being cancelled, which is the opposite of what this sentence promises. Smaller, on the same line: sentence one says "Most durable **operators** implement `on_kill`" inside a *Deferrable tasks* subsection, where the operator's `on_kill` is the one thing that cannot run. Sentence two already says "triggers", so it is a consistency fix rather than a wrong claim. The named exception pair is right, for what it's worth: `grep -rln "def on_kill" providers/ | grep /triggers/` returns ten files and neither `amazon/aws/triggers/glue.py` nor `apache/livy/triggers/livy.py` is among them. ########## airflow-core/src/airflow/ui/src/components/Clear/TaskInstance/ClearTaskInstanceDialog.tsx: ########## @@ -93,18 +93,21 @@ const ClearTaskInstanceDialog = (props: Props) => { const future = selectedOptions.includes("future"); const upstream = selectedOptions.includes("upstream"); const downstream = selectedOptions.includes("downstream"); + const [keepTaskState, setKeepTaskState] = useState(false); Review Comment: Non-blocking, and a follow-up PR is a fine answer. The line right below this seeds `preventRunningTask` from `useClearPreventRunningTaskDefault` (`hooks/useUserSettings.ts:56`, key at `constants/localStorage.ts:35`), which users set once on the settings page (`pages/Settings/Settings.tsx:175`). `keepTaskState` is a bare `useState(false)` with no equivalent anywhere -- `grep -rn KEEP_TASK_STATE ui/src` is empty. Since this PR is what makes discarding the default, the people most affected are exactly the ones who will want to flip it back per-session and now have to tick the box on every clear, in three dialogs. Same treatment as its neighbour would close that, and it stays a UI preference rather than a `[state_store]` option, which would overlap `clear_on_success`. Unrelated, while you are in this footer: `Modal.tsx:73` lays `footerActions` out as `row-reverse` with no `flexWrap`, and with the run-on-latest option showing it now carries four items plus Cancel. The screenshots in the PR description are all at 2497px. One at phone width with a version-drifted Dag would settle it. ########## airflow-core/src/airflow/api_fastapi/core_api/services/public/task_instances.py: ########## @@ -73,20 +89,27 @@ def _clear_task_state_store_on_success(tis: Sequence[TI], session: Session) -> N ) try: backend.clear(scope=scope, session=session) - log.info( - "Cleared task state on success", - dag_id=ti.dag_id, - run_id=ti.run_id, - task_id=ti.task_id, - map_index=ti.map_index, - ) + discarded_count += 1 except Exception: log.warning( - "Failed to clear task state on success", + "Failed to discard task state", + discard_event=event, dag_id=ti.dag_id, run_id=ti.run_id, task_id=ti.task_id, + map_index=ti.map_index, + exc_info=True, ) + break + if discarded_count: + log.info(event, task_instance_count=discarded_count) Review Comment: After the break above this still fires, so a partial discard is reported under the same event name the all-clear case uses, with a count that reads as a total and no denominator. The route never learns about the failure either: it falls through to the note patch, then the unconditional re-query at `routes/public/task_instances.py:1008-1013`, and returns 200 listing every task instance as cleared. That is observable on the default dev backend, not just a theoretical backend asymmetry. Patching `MetastoreBackend._clear_task_state_store` to raise and running the clear against two mapped indexes under breeze on SQLite: when the first call raises, the request still returns 200 and reports `total_entries=2`, both `task_state_store` rows survive, and both task instances come back at state `None`, so the clear itself committed. When the second call raises instead, the warning is immediately followed by `Discarded task state on clear task_instance_count=1` and still a 200, so the surviving row is invisible to the caller and the log line reads as a success. Worth noting the first case is quieter still: `if discarded_count:` suppresses the summary entirely, so the only trace is a warning. On Postgres the aborted transaction makes the re-query raise and the request 500s with a full rollback, so one code path has two opposite user-visible outcomes, and the non-Postgres one is the silen t stale-resume this PR exists to prevent. On Postgres the aborted transaction makes that re-query raise and the request 500s with a full rollback, so one code path has two opposite user-visible outcomes, and the non-Postgres one is the silent stale-resume this PR exists to prevent. Two smaller things in the same refactor. In the `clear_on_success` caller the summary fires once per entity, since `BulkTaskInstanceService._perform_update` calls `_patch_task_instance_state` per entity (`:501-509`), so a bulk mark-success now emits a run of `task_instance_count=1` lines carrying no `dag_id`/`run_id`/`task_id` where the pre-PR per-TI line carried all three: same volume, less information. And the first `backend.clear` is the first statement after `clear_task_instances`, so it carries that autoflush, and a flush failure originating in the clear gets reported as "Failed to discard task state". No test makes `backend.clear` raise, so none of this branch is covered. ########## airflow-core/src/airflow/api_fastapi/core_api/services/public/task_instances.py: ########## @@ -73,20 +89,27 @@ def _clear_task_state_store_on_success(tis: Sequence[TI], session: Session) -> N ) try: backend.clear(scope=scope, session=session) - log.info( - "Cleared task state on success", - dag_id=ti.dag_id, - run_id=ti.run_id, - task_id=ti.task_id, - map_index=ti.map_index, - ) + discarded_count += 1 except Exception: log.warning( - "Failed to clear task state on success", + "Failed to discard task state", + discard_event=event, dag_id=ti.dag_id, run_id=ti.run_id, task_id=ti.task_id, + map_index=ti.map_index, + exc_info=True, ) + break Review Comment: This break is what I asked for in round 2, and it reached a second caller neither of us accounted for. At the base, `_clear_task_state_store_on_success` had its own loop whose `except` logged and carried on to the next task instance (`git show 17ffe26979:...task_instances.py`, the function at :62 with its `except` at :83). Now that it delegates here, the two mark-as-success paths (`:289` and `:324`) stop at the first failure, so with `[state_store] clear_on_success = True` every later task instance keeps state the config explicitly asks to clear. Breaking is right for the clear route, where a failed statement on Postgres poisons everything after it in the request. On the success path the old continue was the deliberate behaviour and this silently drops it. A `stop_on_error` flag, `True` from the clear route and `False` from `_clear_task_state_store_on_success`, keeps both. Worth noting there is a second server-side implementation of the same config option that this PR leaves alone: `execution_api/routes/task_instances.py:529-552` clears the scope inline for a single task instance and swallows a failure to carry on with the request. Before this PR both sites logged and carried on, so after it the failure handling for one config option differs between them. Neither direction is covered today: the `clear_on_success` tests at `test_task_instances.py:5671` and `:5700` never make a `backend.clear()` call fail mid-loop. -- 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]
