yakovlef commented on code in PR #76827:
URL: https://github.com/apache/airflow/pull/76827#discussion_r4549666134
##########
airflow-core/src/airflow/models/taskinstance.py:
##########
@@ -2652,8 +2652,12 @@ def _asset_event_extras_from_aliases() ->
dict[tuple[SerializedAssetUniqueKey, s
)
if event is None:
ti.log.info("Dynamically creating AssetModel %s",
asset_key)
- session.add(AssetModel.from_serialized(asset))
- session.flush() # So event can set up its asset fk.
+ # Another task can create the same asset between our
lookup and this insert.
+ try:
+ with session.begin_nested():
+ session.add(AssetModel.from_serialized(asset))
+ except IntegrityError:
+ ti.log.info("AssetModel %s was created concurrently;
reusing it", asset_key)
asset_manager.register_asset_change(
Review Comment:
good point, added an error log there. raising would fail the task, and the
old code ignored this case too
--
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]
