amoghrajesh opened a new pull request, #73315:
URL: https://github.com/apache/airflow/pull/73315
<!-- 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".
-->
- [x] Yes claude sonnet
<!--
Generated-by: [Tool Name] following [the
guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
### What
Operator extra links are stored in xcom under one key per task instance, and
every retry deletes the task's xcoms before running. An extra link records what
one attempt did, so the next attempt destroying it means earlier tries lose
their link.
The extra links endpoint and the UI already pass `try_number` end to end,
but `XComOperatorLink.get_link` ignored it and read the single surviving row,
so the API looked try-aware while silently serving the latest try's link for
every try.
### Current behaviour
A task that fails once and succeeds on retry leaves a single xcom row
holding the second attempt's link. Opening try 1 in the UI shows try 2's link,
with nothing indicating it belongs to a different attempt. If the retry dies
before `finalize()` runs, no link exists at all.
This predates Airflow 3. Airflow 2 links read XCom for the runtime IDs they
needed `(job id, run id, cluster name)` and lost them to the same issue. The
rewrite to `XComOperatorLink` carried the gap forward rather than introducing
it, but also linked a single finished string written at a known point, which is
what makes a per-try key cheap now.
### Proposed change
Each attempt writes its link under a key of its own,
`_link_<LinkClass>__try_<n>`, and the retry wipe leaves operator link keys
alone. The resolver looks up the key for the try being viewed and falls back to
the unsuffixed key when there is no per try row.
### Changes of Note
**Filtering happens on the worker, not the API server.** The server lists
which xcom keys exist; the worker already holds the task object and knows its
link keys, so no Dag parsing is added to the `ti_run` path and the execution
API contract is untouched.
**The legacy row is kept, not wiped.** After upgrading, a row written before
this change survives and acts as the fallback for tries that have no per try
row. Old tries keep showing what they show today rather than losing their
button.
### User implications / backcompat
No schema change, no migration, no config.
- Runs that executed before this change resolve through the fallback and
render exactly as they do now.
- A new API server with an older worker degrades to current behaviour: the
worker writes only the unsuffixed key, the server falls back to it.
- A task with several tries now has one xcom row per try per link instead of
one. Link rows are not currently hidden from the xcom tab, so they are more
visible there.
### Testing
Unit tests cover the resolver reading the requested try, the fallback to the
unsuffixed key, the per try write in `finalize()`, and the wipe skipping link
keys while still clearing everything else. Each fails without the source change.
#### Manual validation
Using this DAG:
```python
import pendulum
from airflow.sdk import DAG, BaseOperator
from airflow.sdk.bases.operatorlink import BaseOperatorLink
class JobConsoleLink(BaseOperatorLink):
name = "Job console"
def get_link(self, operator, *, ti_key):
return f"https://example.com/job?try={ti_key.try_number}";
class FlakyJobOperator(BaseOperator):
operator_extra_links = (JobConsoleLink(),)
def execute(self, context):
if context["ti"].try_number == 1:
raise RuntimeError("boom on first try")
return "ok"
with DAG(
"test_link_per_try",
start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
schedule=None,
catchup=False,
):
FlakyJobOperator(task_id="run_job", retries=1,
retry_delay=pendulum.duration(seconds=5))
```
##### Simple situation: Clean database
Try 1:
<img width="2475" height="1225" alt="image"
src="https://github.com/user-attachments/assets/9d239db1-9e57-4a2f-a50d-e72eb8612cde";
/>
Try 2:
<img width="2475" height="1225" alt="image"
src="https://github.com/user-attachments/assets/a5f021c1-53df-4cf9-966c-7eab101b5877";
/>
<img width="2479" height="1218" alt="image"
src="https://github.com/user-attachments/assets/7eadd237-e2fc-48fd-be7b-bd6f8cc57bcf";
/>
##### Situation where old format was in xcom already: newly upgraded
<img width="1000" height="309" alt="image"
src="https://github.com/user-attachments/assets/25e92ca2-2113-4b32-a392-819c59b842f6";
/>
Try 1 and 2 were already run so they both show older behaviour: showing link
with `try=2`
<img width="2479" height="1218" alt="image"
src="https://github.com/user-attachments/assets/5fef551d-5251-43c0-80ed-2eab15ff96f2";
/>
<img width="2479" height="1218" alt="image"
src="https://github.com/user-attachments/assets/3050bcc9-c481-40ff-8342-7b04d428552f";
/>
On upgrade to this new version, the task is cleared and third try is
launched and it has link with `try=3`:
<img width="2479" height="1218" alt="image"
src="https://github.com/user-attachments/assets/12545e9c-8a98-4e66-9b6e-c97fc72f7806";
/>
<img width="2479" height="1218" alt="image"
src="https://github.com/user-attachments/assets/9eac908c-50ef-4309-a360-629fb71ebf3c";
/>
Upgrade and mixed version cases checked by hand:
- **Link written before this change, then both server and worker upgraded.**
The unsuffixed row is still resolved through the fallback and renders as it did
before, for every try.
- **Such a run cleared and re-run after upgrading.** The new attempt gets
its own per try link; earlier tries fall back to the pre upgrade row and keep
their button rather than going blank.
- **New API server with an older worker.** The worker writes only the
unsuffixed key, the server falls back to it, and behaviour matches current
Airflow.
---
* 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]
