kaxil commented on code in PR #72100:
URL: https://github.com/apache/airflow/pull/72100#discussion_r3963104184
##########
airflow-core/src/airflow/api_fastapi/core_api/services/public/task_instances.py:
##########
@@ -74,21 +80,29 @@ 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",
+ event,
dag_id=ti.dag_id,
run_id=ti.run_id,
task_id=ti.task_id,
map_index=ti.map_index,
)
except Exception:
log.warning(
- "Failed to clear task state on success",
+ "Failed to discard task state",
+ event=event,
Review Comment:
`event` is structlog's own first positional parameter, so this call raises
`TypeError: got multiple values for argument 'event'`. Airflow's own bound
logger in `shared/logging/src/airflow_shared/logging/structlog.py` defines
`meth(self, event, *args, **kw)`, same as vanilla structlog. The except branch
crashes instead of logging and skipping, so the first backend failure takes the
whole clear down with it. The docstring's reasoning doesn't quite hold either:
`backend.clear()` and `clear_task_instances()` share one uncommitted
transaction (`create_session` commits only when the request returns, and
`clear_task_instances` never flushes), so a DB error here rolls the clear back
whether you swallow it or not.
##########
airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py:
##########
@@ -231,6 +231,11 @@ class ClearTaskInstancesBody(StrictBaseModel):
"and finally ``False`` (the historical default for clear/rerun).",
)
prevent_running_task: bool = False
+ keep_task_state: bool = Field(
Review Comment:
The flag reaches both task clear dialogs, but two other clear surfaces still
discard with no way out: `useBulkClearTaskInstances` behind the multi-select
clear on the Task Instances page (it already plumbs `prevent_running_task`, so
the pattern is right there), and `airflowctl dags clear`, which hand-builds
this body in `dag_command.py` and has no matching arg in `DAG_COMMANDS`.
##########
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:
"Clearing discards them" reads as unconditional, but clearing a whole run
keeps task state: `perform_clear_dag_run` calls `dag.clear` directly and never
reaches this endpoint. Worth scoping the sentence to a task clear so nobody
reads the run-level Clear button into it.
##########
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:
`_get_db_backend()` skips `[workers] state_store_backend` by design, which
is why `clear_on_success` splits the work: the worker clears the custom backend
via `_clear_backend_only` and the server drops the DB rows. There's no worker
in the clear path, so on a custom backend this drops the ref row and leaves the
payload. Reads go through the DB so the start-over behaviour is still correct,
but is anything going to reclaim those objects?
--
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]
