slice-soupam commented on code in PR #72297:
URL: https://github.com/apache/airflow/pull/72297#discussion_r4433243061
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py:
##########
@@ -860,7 +863,12 @@ def _find_writer_id(
dag_bag: DBDagBag,
) -> UUID | None:
"""Resolve the task instance a write targets with the rule reads use,
working from the caller."""
- caller = session.get(TaskInstance, token.id,
execution_options={"include_all_attempts": True})
+ # Dag-parse tokens identify a parsing attempt, not a task instance;
resolve by coordinates only.
+ caller = (
+ session.get(TaskInstance, token.id,
execution_options={"include_all_attempts": True})
+ if isinstance(token, TIToken)
+ else None
Review Comment:
Thanks — addressed by pinning the write to the originating attempt UUID.
**Change:**
- `SetXCom` now carries optional `task_instance_id` (supervisor schema bump
`AddTaskInstanceIdToSetXCom`). Callback / own-TI paths populate it from the
attempt that issued the write.
- The Execution API accepts it as a query param on `POST /xcoms`. When
present, `_find_writer_id()` loads that attempt (including archived), validates
it matches the URL coordinates, and writes there — no fallback to a successor
after clear/retry.
- Without `task_instance_id`, behavior stays coordinate-based (existing
task-path semantics).
**Regression:**
`test_dag_parse_set_with_task_instance_id_pins_archived_attempt` archives a TI,
creates the successor, POSTs with the archived id, and asserts the XCom lands
on the archived attempt while the successor’s XCom remains unset.
--
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]
