[I] Operator extra links from earlier attempts are deleted on retry [airflow]

["1fanwang (via GitHub)" <gi...@apache.org>]

1fanwang opened a new issue, #71471:
URL: https://github.com/apache/airflow/issues/71471

   ### 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:
   
   https://github.com/apache/airflow/blob/537feafb3697338b3988d826b9c9b45da06b1f23/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py#L285-L293
   
   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:
   
   ```sql
   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 537feafb3697338b3988d826b9c9b45da06b1f23.
   
   ### Are you willing to submit PR?
   
   - [X] Yes I am willing to submit a PR!
   
   ### Code of Conduct
   
   - [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
   


-- 
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: commits-unsubscribe@airflow.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org