seanghaeli commented on code in PR #76034:
URL: https://github.com/apache/airflow/pull/76034#discussion_r4556656106
##########
airflow-core/src/airflow/models/taskinstance.py:
##########
@@ -2652,8 +2652,20 @@ 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.
+ # Flush earlier event writes before the savepoint, outside
conflict recovery.
+ savepoint = session.begin_nested()
Review Comment:
This covers the asset insert, but the alias link next to it has the same
race. When the asset already exists and the alias isn't linked to it yet, both
writers see no link in `_add_asset_alias_association`, both insert the
`asset_alias_asset` row, and the second one fails on `asset_alias_asset_pkey`,
which rolls back its task completion. Found this in testing: your two-writer
setup fails that way when the target is declared as a concrete outlet first.
Mapped tasks emitting one known asset through an alias would hit it on their
first run. Could the link insert get the same savepoint treatment, or an insert
that ignores the conflict?
--
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]
