kaxil commented on code in PR #72100:
URL: https://github.com/apache/airflow/pull/72100#discussion_r3968150764
##########
airflow-core/src/airflow/api_fastapi/core_api/services/public/task_instances.py:
##########
@@ -74,21 +88,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,
Review Comment:
One INFO per task instance on an endpoint that emits no API-server log lines
today adds up on a bulk clear, and since `_clear_task_state_store` throws away
the `CursorResult` and `clear()` returns `None`, the identical line fires for
every task instance that never wrote any state. A single line with a count
after the loop, or a rowcount from the backend so it only fires when something
was actually dropped, would say more. The warning below could also use
`exc_info=True` and the `map_index` that the success branch includes, since
without either there is nothing to identify the row that kept its checkpoint.
##########
airflow-core/src/airflow/ui/src/components/Clear/TaskInstance/ClearTaskInstanceDialog.tsx:
##########
@@ -93,6 +93,7 @@ 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:
Nothing ever resets this: `setKeepTaskState` appears only in the three
`useState` declarations and the three checkbox handlers, and none of the
dialogs unmount on close (`ClearTaskInstanceButton.tsx:100-118` gates on
`taskInstance`, not `open`), so a tick survives close, reopen, and a move to a
different task instance, and the next clear quietly resumes from a checkpoint.
`note` is reset on both open and close five lines below, and
`preventRunningTask` is sticky on purpose because it is seeded from the
`useClearPreventRunningTaskDefault` localStorage preference, so a bare
`useState` here looks like an omission rather than the house pattern. While you
are in this block, the checkbox at line 215 also lands on the Confirm side of
the footer gap, since `preventRunningTask` carries `marginRight: "auto"` inside
the `row-reverse` flex from `Modal.tsx:73`.
##########
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**
Review Comment:
The sweep is not finished outside core. `grep -rn "treated the same as a
retry" providers/` returns six shipped pages still carrying the heading this PR
deletes and the paragraph under it, each cross-linking back to this page:
amazon glue.rst:206, amazon redshift/redshift_data.rst:102, databricks
run_now.rst:116 and submit_run.rst:194, snowflake.rst:192, apache spark
operators.rst:226. Those are exactly the provider families that write
`task_state_store`, and the paragraph tells the reader clearing does not delete
the stored run id and that `clear_on_success` is what restores resubmission, so
a Glue user following it now gets a second Glue job.
##########
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:
Worth a sentence saying this is the `airflowctl` clear only. The core
`airflow dags clear` goes straight to `clear_task_instances` through
`_bulk_clear_runs` (`cli/commands/dag_command.py:236`), so it keeps task state
and has no `--keep-task-state`, leaving two shipped CLIs with the same command
name and opposite defaults. Neither the newsfragment nor the docs mention the
new flag either.
##########
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:
The rewrite dropped the deferred case the old text called out, and
`core-concepts/task-state-store.rst:288` still tells deferrable authors to
reach for the store "only when you need to survive an operator-initiated
clear", which is the thing that stops happening by default here. A deferred
clear does still cancel the job for most durable operators, through the
trigger's `on_kill` rather than the worker's, since `cancel_triggers()` cancels
with `_USER_ACTION_CANCEL_MSG` and `run_trigger` then awaits
`trigger.on_kill()`. So the ones that actually need the box ticked are those
whose trigger does not implement it, `GlueJobCompleteTrigger` and `LivyTrigger`
today.
##########
airflow-core/src/airflow/api_fastapi/core_api/services/public/task_instances.py:
##########
@@ -59,10 +59,24 @@
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:
+ """
+ Discard the task state store entries of each task instance.
+
+ A failure to discard one task instance is logged and skipped rather than
raised, so one bad
Review Comment:
On PostgreSQL this is not what happens. A failed statement aborts the
transaction, so once one `backend.clear()` raises, every later iteration raises
too and is swallowed, and then the note patch or the trailing re-query back in
the route raises uncaught and the whole clear rolls back behind a 500. The
per-entry resilience holds on SQLite only, so either scope the sentence to that
or stop the loop on the first failure.
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py:
##########
@@ -4339,6 +4339,72 @@ def test_clear_dry_run_does_not_set_note(self,
test_client, session):
ti_id = response_data["task_instances"][0]["id"]
_check_task_instance_note(session, ti_id, {"content":
"placeholder-note", "user_id": None})
+ def _seed_task_state(self, session, dag_id):
+ """Store one task state key for the single TI created by these
tests."""
+ ti = session.scalars(select(TaskInstance).where(TaskInstance.dag_id ==
dag_id)).one()
+ MetastoreBackend().set(
+ TaskScope(dag_id=ti.dag_id, run_id=ti.run_id, task_id=ti.task_id,
map_index=ti.map_index),
+ "job_id",
+ "app_1234",
+ session=session,
+ )
+ session.commit()
+
+ def _task_state_rows(self, session, dag_id):
+ return
session.scalars(select(TaskStateStoreModel).where(TaskStateStoreModel.dag_id ==
dag_id)).all()
+
+ @pytest.mark.db_test
+ @pytest.mark.parametrize(
+ ("payload_extra", "expect_kept"),
+ [
+ pytest.param({}, False, id="default-discards"),
+ pytest.param({"keep_task_state": True}, True, id="keep-preserves"),
+ pytest.param({"keep_task_state": False}, False,
id="explicit-false-discards"),
Review Comment:
This param asserts a Pydantic default rather than behaviour:
`keep_task_state` is a plain-defaulted bool and nothing reads
`model_fields_set` for it the way `run_on_latest_version` needs, so omitting
the key and sending `false` are the same object by the time the route sees it.
What is missing instead is the loop and the `map_index` scoping in
`_discard_task_state_store`, since every new test uses a single unmapped TI,
and `_task_state_rows` filters on `dag_id` alone so it cannot tell "discarded
the target" from "discarded everything". Swapping this case for a mapped task
at two or three indexes would cover both, though `_seed_task_state` uses
`.one()` so it needs `.first()` or an explicit ti first.
--
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]
