apache / apache/airflow

only_new clear can silently miss real new tasks after a prior "run on latest version" clear

Open
#71,455 1 comment 0 reactions 0 assignees View on GitHub
area:core kind:bug
Dominant language
Python
Stars
46.9k
Forks
17.8k
Avg merge
2d 9h
Merged PRs (30d)
472

Description

## Problem

`_get_new_task_ids` (`airflow-core/src/airflow/models/taskinstance.py:270-309`), which powers the `only_new` clear feature added in #59764, computes "tasks added since this run's version" like this:

```python
# Use created_dag_version_id directly to get the DAG version the run was
# originally created with. We cannot use get_dag_for_run here because it
# falls back to the latest version when bundle_version is not set (e.g.
# LocalDagBundle), which would make current_dag == latest_dag and the diff
# always empty.
current_dag = None
if dag_run.created_dag_version_id:
current_dag = scheduler_dagbag.get_dag(version_id=dag_run.created_dag_version_id, session=session)
new_task_ids = set(latest_dag.task_ids) - set(current_dag.task_ids) if current_dag else set()
```

The comment explicitly documents the assumption that `created_dag_version_id` means "the version the run was originally created with." That assumption was correct when the field was immutable. Since #54984, `clear_task_instances(..., run_on_latest_version=True)` mutates `created_dag_version_id` to the latest version for a pinned run (while only bumping `dag_version_id` on the specific task instances that were cleared — see `taskinstance.py:477`). No new task-instance rows get created by that operation; it only updates the pointer and the already-existing, cleared TIs.

Once that has happened, `created_dag_version_id == latest`, so `current_dag == latest_dag`, and `new_task_ids` becomes `latest_dag.task_ids - latest_dag.task_ids` = the empty set — even if the run genuinely never got a task instance created for a task that was added between the run's true original version and latest.

## Repro

1. Create a pinned DagRun at version V1 with tasks A and B.
2. Deploy V2, which adds task C (V2 task_ids = {A, B, C}).
3. Clear task A with `run_on_latest_version=True`. Now `dag_run.created_dag_version_id == V2`; the run still has no TI for task C.
4. Run an `only_new` clear on this DagRun.

## Expected

Task C is identified as new and a TaskInstance is created for it.

## Actual

`_get_new_task_ids` returns an empty set (`current_dag` resolves to V2, same as `latest_dag`), so task C is silently never added, even though the run's actual task-instance rows never picked it up.

## Suggested fix

Determine "already has a TI for task X" by checking the run's actual `TaskInstance` rows directly (e.g. `{ti.task_id for ti in dag_run.get_task_instances(session=session)}`) rather than by diffing against a dag-version pointer (`created_dag_version_id`) that can no longer be trusted to represent "the version whose task set defines this run's existing task instances."

## Context

This is one concrete consequence of `created_dag_version_id`'s contract having drifted from "the version recorded at DagRun creation" (its documented meaning) to "the version this run should currently run at" (its behavior since #54984). See the companion issue tracking that broader contract problem: #71453

---
Drafted-by: Claude Code (Sonnet 5) (no human review before posting)

Contributor guide

Open the contributing guide

Research direction

Start in airflow-core/src/airflow/models/taskinstance.py:270-309 at _get_new_task_ids, then read clear_task_instances around line 477 and trace how created_dag_version_id changes. Reproduce the V1/V2 sequence from the issue and verify that only_new identifies task C and creates its TaskInstance without relying on the mutated version pointer.

Written by the indexing model from the issue text.

Assessment

Tech stack
python
Domain
backend, data-engineering
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Quiet
Clarity
Clearly specified
Newbie friendliness
52/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.