amoghrajesh opened a new pull request, #72100:
URL: https://github.com/apache/airflow/pull/72100
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull
request.
Write a good git commit message following this guide:
https://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the
documentation.
For user-facing UI changes, please attach before/after screenshots (or a
short
screen recording) so reviewers can assess the visual impact.
Feel free to ping (in general) for the review if you do not see reaction for
a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we
sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR,
please
change below checkbox to `[X]` followed by the name of the tool, uncomment
the "Generated-by".
-->
- [ ] Yes (please specify the tool below)
<!--
Generated-by: [Tool Name] following [the
guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
## Summary
Clearing a task instance now discards its `task_state_store` entries, so the
next attempt starts over instead of resuming from a checkpoint or reconnecting
to an external job recorded by the attempt that was cleared.
Clearing a task instance now discards its `task_state_store` entries, so the
next attempt starts
over. Retries are unaffected and still resume.
## Why
A retry and a clear were treated identically. Both kept the task's state,
both resumed. But they mean different things.
A retry happens because infrastructure failed. Nothing about the work
changed, so carrying on from the checkpoint is exactly right, and that is what
crash recovery is for. A clear happens because a human intervened, and the
usual reason a human clears a task is that something *did* change: the code,
the upstream data, a connection, a config value.
So the behaviour was tuned for the case where nothing changed, and then
applied to the case where
something had.
What that cost in practice:
**Fixing a bug and clearing left the bug's output in place.** A task gets
through files 1 to 6, hits bad data on the 7th, you fix the transform and
clear. It resumed at 7. Files 1 to 6 still held output from the code you had
just fixed, now silently mixed with the corrected work. The task went green,
and nothing anywhere said otherwise.
**Clearing a succeeded durable task did nothing at all.** The operator read
back the stored job id, saw the external job had already finished, and returned
the stored result in a couple of seconds. This needs no unusual configuration,
and there is no reading of "clear" under which doing nothing is what the user
asked for.
The two failure modes are asymmetric, which is what decides the default
rather than just moving the
problem elsewhere. Discarding when you wanted to resume costs repeated work
you can watch happen.
Resuming when you wanted a fresh start produces wrong output you cannot see.
## What changed
- `keep_task_state` on `ClearTaskInstancesBody`, defaulting to `false`
- The discard runs in `post_clear_task_instances`, inside the existing `if
not dry_run:` block and
after `clear_task_instances` succeeds, so a preview discards nothing and a
failed clear cannot
take the task state with it
- `_clear_task_state_store_on_success` refactored into a shared
`discard_task_state_store` helper:
two callers, two gates, one implementation
- A "Keep task state and resume" checkbox in both clear dialogs, unchecked
by default, next to the
existing "Prevent rerun if task is running"
- Concept docs rewritten. The section previously asserted the opposite,
under the heading "Clearing
a task is treated the same as a retry"
## When to tick the box
Two situations, both documented.
**Nothing changed and you only want the task to carry on.** Retries normally
cover this, so you reach it when retries are exhausted.
**An external job is still running.** Most operators cancel theirs in
`on_kill`, so clearing a *running* task leaves nothing to reconnect to. But
clearing a *failed* task never runs `on_kill`, so a job that outlived its
worker is still going, and discarding the stored id submits a second one. Same
for operators configured to leave the job alive, such as
`KubernetesPodOperator` with `on_kill_action="keep_pod"`.
## Compatibility
This changes behaviour introduced in 3.3 and released in 3.3.1. Anyone
relying on clear-to-resume needs `keep_task_state=true`.
A single default rather than per-operator behaviour is acceptable precisely
because the user keeps an override: if Airflow decided silently per operator, a
wrong guess would be unrecoverable, whereas a wrong default is one checkbox.
## Tests
Trying to run a dag like this:
```
from __future__ import annotations
import logging
from datetime import datetime
from airflow.sdk import DAG, NEVER_EXPIRE, Variable, task
log = logging.getLogger("airflow.task")
ROWS = ["100", "200", "N/A", "400", "500"]
with DAG(
dag_id="clear_after_fix",
schedule=None,
start_date=datetime(2026, 1, 1),
catchup=False,
tags=["task-state-store"],
doc_md=__doc__,
):
@task(retries=0)
def transform_amounts(**context):
store = context["task_state_store"]
handle_missing = Variable.get("handle_missing", default="false") ==
"true"
done = set(store.get("done", default=[]))
if done:
log.info("Resuming, %d of %d rows already done", len(done),
len(ROWS))
else:
log.info("Starting fresh, %d rows to transform", len(ROWS))
total = 0
for row in ROWS:
if row in done:
log.info(" %-5s skipped, already done", row)
continue
if row == "N/A" and not handle_missing:
raise ValueError(f"cannot transform {row!r}: fix the
transform and clear this task")
total += 0 if row == "N/A" else int(row)
done.add(row)
store.set("done", sorted(done), retention=NEVER_EXPIRE)
log.info(" %-5s transformed", row)
log.info("Done. %d rows, total %d", len(done), total)
return total
transform_amounts()
```
This dag showcases some data transformation and mimics a case where bad data
came in as `N/A` controlled by a variable value.
So showing it:
First run:
<img width="2498" height="967" alt="image"
src="https://github.com/user-attachments/assets/a842df07-01eb-40cf-bb43-d03637f194c6";
/>
Task state store contains this:
<img width="2497" height="1108" alt="image"
src="https://github.com/user-attachments/assets/f542cfc6-c856-4984-bf9c-4aa04b9c53c3";
/>
Set the variable rightly now:
`airflow variables set handle_missing true`
Clear with the default:
<img width="894" height="420" alt="image"
src="https://github.com/user-attachments/assets/1b30c9e5-ad33-4a3c-9442-bcfd25622a95";
/>
Next run:
<img width="2497" height="1108" alt="image"
src="https://github.com/user-attachments/assets/869bb050-b6ed-418d-9029-6e1741a49dfe";
/>
State store:
<img width="2497" height="1108" alt="image"
src="https://github.com/user-attachments/assets/b1144a11-9438-4268-8533-9167054d001b";
/>
Now if I cleared by overriding the checkbox for another try (ie: keeping
task state):
<img width="2497" height="1108" alt="image"
src="https://github.com/user-attachments/assets/b48c0651-86d9-408a-bd35-d7dcefa9a796";
/>
Observe the logs:
<img width="2497" height="1108" alt="image"
src="https://github.com/user-attachments/assets/602f771d-03bd-47f1-9585-b6a5d2f6c76b";
/>
---
* Read the **[Pull Request
Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)**
for more information. Note: commit author/co-author name and email in commits
become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal
([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals))
is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License
Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment:
`{pr_number}.significant.rst`, in
[airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments).
You can add this file in a follow-up commit after the PR is created so you
know the PR number.
--
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]
