shahar1 commented on code in PR #73005:
URL: https://github.com/apache/airflow/pull/73005#discussion_r4023204974
##########
airflow-core/src/airflow/models/xcom.py:
##########
@@ -245,6 +253,7 @@ def set(
dag_id=dag_id,
map_index=map_index,
dag_result=dag_result,
+ mapped_length=mapped_length,
Review Comment:
I reproduced the following regression with a same-length* XCom edit:
<details>
<summary>Example Dag</summary>
```python
@dag(schedule=None)
def xcom_edit_example():
@task
def emit():
return [1, 2, 3]
@task
def slow():
time.sleep(300) # Allows time to edit the XCom.
@task
def consume(value):
print(value)
values = emit()
waiting_task = slow()
consumers = consume.expand(value=values)
waiting_task >> consumers
xcom_edit_example()
```
</details>
1. Trigger the Dag, assume that `emit` and `slow` run in parallel.
2. After `emit` succeeds, edit its `return_value` XCom from `[1, 2, 3]` to
`[9, 9, 9]` through the UI. `consume` is still waiting for `slow`, so it hasn’t
expanded yet.
3. Wait for `slow` to finish.
**On main:** the recorded length remains `3`, and `consume` expands into
three tasks.
**On this PR:** the edit clears the recorded length, and `consume` becomes
`UPSTREAM_FAILED`.
\* Length-changing edits are already broken on main - the recorded length
isn't updated, so you silently get the wrong number of mapped tasks. This PR
just fails instead.
---
For this PR, passing `mapped_length=xcom_entry.mapped_length` through the
existing `XComModel.set call` restores the current behaviour.
Long term - maybe we should restrict editing an XCom that has a recorded
mapped length?
--
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]
