nagasrisai opened a new pull request, #69959:
URL: https://github.com/apache/airflow/pull/69959
Closes #69956
## Summary
When two task runners (concurrent execution or task retry) write the same
XCom key at roughly the same time, both call `XComModel.set` which does a
DELETE followed by an INSERT. Both DELETEs succeed (they each see the record
that was just written by the peer or see nothing), then both try to INSERT the
same primary key. The second INSERT hits the unique constraint and the
Execution API converts the `IntegrityError` to an HTTP 409, causing the task to
fail.
### Fix
Wrap the `XComModel.set` call in a **savepoint** (`session.begin_nested()`)
so that a constraint violation only rolls back the XCom write and not any
earlier work in the same request (e.g. the `task_map` merge). The
`IntegrityError` is caught and handled with an explicit UPDATE that overwrites
the existing entry with the latest value, which is the correct semantic for a
retry.
```python
try:
with session.begin_nested():
XComModel.set(...)
except IntegrityError:
# concurrent write already committed; overwrite with latest value
session.execute(update(XComModel).where(...).values(value=value,
dag_result=dag_result))
```
## Changes
- `airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py`
- Add `update` to the `sqlalchemy` import
- Add `IntegrityError` import from `sqlalchemy.exc`
- Wrap `XComModel.set` in `session.begin_nested()` and catch
`IntegrityError` with an UPDATE fallback
--
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]
