Operator extra links from earlier attempts are deleted on retry

[1fanwang]

Apache Airflow version

3.2.2

What happened?

Operator extra links are backed by XCom rows. BaseOperatorLink.xcom_key defaults to
_link_<ClassName>, and get_link() is handed a TaskInstanceKey that carries a
try_number.

When a task instance transitions to running, the execution API collects every XCom key
for that task instance into TIRunContext.xcom_keys_to_clear, and the worker deletes each
one. There is no filter — the only exemption is deferral:

airflow/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py

Lines 285 to 293 in 537feaf

	xcom_query = select(XComModel.key).where(
	XComModel.dag_id == ti.dag_id,
	XComModel.task_id == ti.task_id,
	XComModel.run_id == ti.run_id,
	)
	if map_index is not None:
	xcom_query = xcom_query.where(XComModel.map_index == map_index)
	xcom_keys = list(session.scalars(xcom_query))

So a retry deletes the _link_* rows written by every previous attempt. An extra link can
only ever resolve for the attempt that ran last.

This collides with a change that shipped in the same release. #65661 (closing #65354) made
the extra-links endpoint try-aware: pass try_number and it resolves the link against that
attempt's TaskInstanceHistory. The read path now asks for attempt N's link while the
write path guarantees only the latest attempt's row exists, so for any earlier attempt the
endpoint returns nothing.

It bites hardest when the external URL cannot be recomputed — a job id the remote service
mints per submission, so each attempt gets a different one. Once the row is gone the link
to that attempt's logs is unrecoverable. Providers linking to per-attempt logs in EMR,
Glue, Databricks, Dataproc or Livy are all in this shape; none of them stores anything
per-attempt today, so the older attempts' links are simply lost.

There is also no way to work around it in a link class. xcom_key is a plain property with
no access to ti_key, so the framework's own fetch cannot vary by attempt; the xcom table
has no try_number column, so per-attempt rows can only be expressed by encoding the try
into the key string; and the collection above takes every key regardless, so an encoded key
is deleted too.

I'm not proposing a shape — the options trade off against each other and the call is yours.
Exempting a key prefix from the collection is the smallest. Giving xcom_key access to
ti_key would make per-attempt links first class but changes a public interface. Pointing
links at the task state store fits where durable cross-retry state is already heading
(#69914, #71211), but that store is aimed at job identity rather than link rendering.

Happy to put up a PR once there's a preferred direction.

What you think should happen instead?

An extra link resolved for attempt N should return attempt N's URL, which is what the
try-aware endpoint added in #65661 implies.

How to reproduce

1. Use any operator with an extra link whose URL is built from a value pushed at runtime
   (_link_<ClassName>), or push the key by hand.
2. Let attempt 1 run and write it.
3. Make the task fail so it retries.
4. While attempt 2 is running, query the metastore:

SELECT `key` FROM xcom WHERE dag_id = '<dag>' AND task_id = '<task>' AND run_id = '<run>';

Only attempt 2's _link_* row is present. Requesting the link for try_number=1 through
the extra-links endpoint returns nothing.

Operating System

Linux

Versions of Apache Airflow Providers

Not provider specific — the behaviour is in airflow-core.

Deployment

Official Apache Airflow Helm Chart

Deployment details

Reproduced on 3.2.2 with the KubernetesExecutor and a MySQL metastore. The code path is
unchanged on main at 537feaf.

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct