Discard task state for a taskinstance when it is cleared

[amoghrajesh]

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

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:

Task state store contains this:

Set the variable rightly now:

airflow variables set handle_missing true

Clear with the default:

Next run:

State store:

Now if I cleared by overriding the checkbox for another try (ie: keeping task state):

Observe the logs:

---

• Read the 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) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[ashb]

Suggested change

	Clearing a task now discards its task state store entries
	Clearing a task now discards its task state store entries by default

I think? "By default" hopefully implying it's changeable?

[amoghrajesh]

yep, handled in b4937ef1b7

[ashb]

This feels like it could be a bug in some respects.

If the max number of retries for a durable task has failed, shouldn't some callback on the operator handle this automatically?

[amoghrajesh]

Yep, it is and not specific to durable tasks.

on_kill() only fires for SIGTERM, or execution_timeout being exceeded. A task that fails by raising never calls it. So a durable task that exhausts its retries can leave the external job running with nothing tracking it!

Checked the current durable operators. Eight of nine put cancellation only in on_kill (Glue, Redshift Data, BigQuery, Snowflake, Livy, SparkSubmit, both Databricks ones), so none clean up on failure. KPO is the exception, and by a different route: its cleanup sits in a finally in execute_sync honouring on_finish_action, so it runs on any exit. ResumableJobMixin has no cleanup path of its own.

I'd keep it out of this PR since it's terminal-failure cleanup rather than clear semantics, but if a terminally failed durable task cancelled its job, there'd be nothing left running for a later clear to duplicate, and most of this caveat goes away. I'll open an issue for that.

[amoghrajesh]

Created an issue for that #72128

[jedcunningham]

I'll let @vatsrahul1001 chime in, but imo we don't need a significant note for this one.

[amoghrajesh]

My reasoning for significant was the old behaviour shipped in 3.3.1, so anyone who clears a task today and expects it to resume from the checkpoint gets a different result after upgrading, without changing anything in their Dag. I'll wait for rahul too

[ashb]

I agree with Amogh here. It's a very simple code change but a possibly very big behaviour change.

[jedcunningham]

Probably don't need the front end of this comment, pretty intuitive. Could keep the "only after clear has suceeded" part, still optional imo.

[amoghrajesh]

Cool, shortened in b4937ef1b7

[jedcunningham]

Suggested change

	if not conf.getboolean("state_store", "clear_on_success", fallback=False):
	if not conf.getboolean("state_store", "clear_on_success"):

Don't need the fallback, defaults come from config already.

[amoghrajesh]

Sounds good. removed in b4937ef1b7