apache · 1fanwang · Aug 12, 2026 · Aug 12, 2026
@@ -94,6 +94,7 @@
 from airflow.models.taskreschedule import TaskReschedule
 from airflow.models.trigger import Trigger, handle_event_submit
 from airflow.models.xcom import XComModel
+from airflow.sdk.bases.operatorlink import ATTEMPT_LINK_XCOM_KEY_PREFIX
 from airflow.serialization.definitions.assets import SerializedAsset, SerializedAssetUniqueKey
 from airflow.state import get_state_backend
 from airflow.triggers.base import TriggerEvent
@@ -296,6 +297,13 @@ def ti_run(
             if map_index is not None:
                 xcom_query = xcom_query.where(XComModel.map_index == map_index)
 
+            # A link that keeps one row per attempt describes an attempt that already ran,
+            # so those rows outlive the clear. The underscores are LIKE single-character
+            # wildcards and must be escaped, or unrelated keys are spared too.
+            xcom_query = xcom_query.where(
+                ~XComModel.key.like(ATTEMPT_LINK_XCOM_KEY_PREFIX.replace("_", r"\_") + "%", escape="\\")
+            )
+
             xcom_keys = list(session.scalars(xcom_query))
         task_reschedule_count = (
             session.scalar(

@@ -24,6 +24,7 @@
 import attrs
 
 from airflow.models.xcom import XComModel
+from airflow.sdk.bases.operatorlink import attempt_link_xcom_key
 from airflow.utils.log.logging_mixin import LoggingMixin
 from airflow.utils.session import create_session
 
@@ -42,6 +43,12 @@ class XComOperatorLink(LoggingMixin):
 
     name: str
     xcom_key: str
+    per_attempt: bool = False
+
+    def xcom_key_for_try(self, try_number: int) -> str:
+        if not self.per_attempt:
+            return self.xcom_key
+        return attempt_link_xcom_key(self.xcom_key, try_number)
 
     def get_link(self, operator: Operator, *, ti_key: TaskInstanceKey) -> str:
         """
@@ -57,7 +64,7 @@ def get_link(self, operator: Operator, *, ti_key: TaskInstanceKey) -> str:
         with create_session() as session:
             result = session.execute(
                 XComModel.get_many(
-                    key=self.xcom_key,
+                    key=self.xcom_key_for_try(ti_key.try_number),
                     run_id=ti_key.run_id,
                     dag_ids=ti_key.dag_id,
                     task_ids=ti_key.task_id,

@@ -1512,7 +1512,7 @@ def _is_excluded(cls, var: Any, attrname: str, op: SDKDAGNode) -> bool:
 
     @classmethod
     def _deserialize_operator_extra_links(
-        cls, encoded_op_links: dict[str, str]
+        cls, encoded_op_links: dict[str, str | dict[str, Any]]
     ) -> dict[str, XComOperatorLink]:
         """
         Deserialize Operator Links if the Classes are registered in Airflow Plugins.
@@ -1540,7 +1540,14 @@ def _deserialize_operator_extra_links(
             #     'raise_error': 'key'
             # }
 
-            op_predefined_extra_link = XComOperatorLink(name=name, xcom_key=xcom_key)
+            # A link that keeps one row per attempt serializes as a mapping; everything
+            # else stays the bare key string it has always been.
+            if isinstance(xcom_key, dict):
+                op_predefined_extra_link = XComOperatorLink(
+                    name=name, xcom_key=xcom_key["key"], per_attempt=xcom_key.get("per_attempt", False)
+                )
+            else:
+                op_predefined_extra_link = XComOperatorLink(name=name, xcom_key=xcom_key)
             op_predefined_extra_links.update({op_predefined_extra_link.name: op_predefined_extra_link})
 
         return op_predefined_extra_links
@@ -1560,7 +1567,12 @@ def _serialize_operator_extra_links(
         :param operator_extra_links: Operator Link
         :return: Serialized Operator Link
         """
-        return {link.name: link.xcom_key for link in operator_extra_links}
+        return {
+            link.name: {"key": link.xcom_key, "per_attempt": link.keeps_a_link_per_attempt}
+            if link.keeps_a_link_per_attempt
+            else link.xcom_key
+            for link in operator_extra_links
+        }
 
     @classmethod
     def serialize(cls, var: Any, *, strict: bool = False) -> Any:

@@ -27,6 +27,15 @@
     from airflow.sdk.types import TaskInstanceKey
 
 
+ATTEMPT_LINK_XCOM_KEY_PREFIX = "_link_attempt_"
+"""Prefix marking the XCom rows that hold one attempt's rendered operator link."""
+
+
+def attempt_link_xcom_key(xcom_key: str, try_number: int) -> str:
+    """Return the XCom key holding ``xcom_key``'s link as rendered for ``try_number``."""
+    return f"{ATTEMPT_LINK_XCOM_KEY_PREFIX}{try_number}_{xcom_key}"
+
+
 @attrs.define()
 class BaseOperatorLink(metaclass=ABCMeta):
     """Abstract base class that defines how we get an operator link."""
@@ -55,6 +64,21 @@ def xcom_key(self) -> str:
         """
         return f"_link_{self.__class__.__name__}"
 
+    keeps_a_link_per_attempt: ClassVar[bool] = False
+    """
+    Keep one row per attempt rather than a single row holding whichever ran last.
+
+    Set on links whose URL cannot be recomputed, such as one carrying a job id the remote
+    service mints per submission. A task's XComs are cleared before each attempt; the rows
+    behind a link that sets this are kept.
+    """
+
+    def xcom_key_for_try(self, try_number: int) -> str:
+        """Return the XCom key holding this link as rendered for ``try_number``."""
+        if not self.keeps_a_link_per_attempt:
+            return self.xcom_key
+        return attempt_link_xcom_key(self.xcom_key, try_number)
+
     @abstractmethod
     def get_link(self, operator: BaseOperator, *, ti_key: TaskInstanceKey) -> str:
         """

@@ -2326,7 +2326,8 @@ def finalize(
     # Pushing xcom for each operator extra links defined on the operator only.
     for oe in task.operator_extra_links:
         try:
-            link, xcom_key = oe.get_link(operator=task, ti_key=ti), oe.xcom_key  # type: ignore[arg-type]
+            link = oe.get_link(operator=task, ti_key=ti)  # type: ignore[arg-type]
+            xcom_key = oe.xcom_key_for_try(ti.try_number)
             log.debug("Setting xcom for operator extra link", link=link, xcom_key=xcom_key)
             _xcom_push_to_db(ti, key=xcom_key, value=link)
         except Exception: