amoghrajesh commented on code in PR #72100:
URL: https://github.com/apache/airflow/pull/72100#discussion_r4013163175
##########
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:
Handled in [comments from
kaxil](https://github.com/apache/airflow/pull/72100/commits/04276bf6fc7a80865e87ee319b12ca75fb575c1b)
##########
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:
Handled in [comments from
kaxil](https://github.com/apache/airflow/pull/72100/commits/04276bf6fc7a80865e87ee319b12ca75fb575c1b)
##########
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:
Handled in [comments from
kaxil](https://github.com/apache/airflow/pull/72100/commits/04276bf6fc7a80865e87ee319b12ca75fb575c1b)
##########
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:
Handled in [comments from
kaxil](https://github.com/apache/airflow/pull/72100/commits/04276bf6fc7a80865e87ee319b12ca75fb575c1b)
##########
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:
Handled in [comments from
kaxil](https://github.com/apache/airflow/pull/72100/commits/04276bf6fc7a80865e87ee319b12ca75fb575c1b)
--
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]
