nagasrisai commented on code in PR #69959:
URL: https://github.com/apache/airflow/pull/69959#discussion_r3713955961
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py:
##########
@@ -433,17 +434,35 @@ def set_xcom(
# means loading the serialized dag and that seems like a relatively costly
operation for minimal benefit
# (the mapped task would fail in a moment as it can't be expanded anyway.)
try:
- # We expect serialised value from the caller - sdk, do not serialise
in here
- XComModel.set(
- key=key,
- value=value,
- run_id=run_id,
- task_id=task_id,
- dag_id=dag_id,
- map_index=map_index,
- serialize=False,
- dag_result=dag_result,
- session=session,
+ # Use a savepoint so that an IntegrityError on the XCom write does not
+ # roll back the task-map merge that may have already been flushed
above.
+ with session.begin_nested():
+ # We expect serialised value from the caller - sdk, do not
serialise in here
+ XComModel.set(
+ key=key,
+ value=value,
+ run_id=run_id,
+ task_id=task_id,
+ dag_id=dag_id,
+ map_index=map_index,
+ serialize=False,
+ dag_result=dag_result,
+ session=session,
+ )
+ except IntegrityError:
Review Comment:
@kaxil - Really good catch, thank you. You're right — a FK violation on the
task_instance cascade would also produce an `IntegrityError` here, and
swallowing it silently would return 200 when the XCom was never actually
stored. I'll narrow the guard to unique-constraint violations only, mirroring
`_UniqueConstraintErrorHandler`, and let everything else propagate. Will push
the fix shortly.
--
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]
