Andrushika commented on code in PR #70951:
URL: https://github.com/apache/airflow/pull/70951#discussion_r3704044918
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -529,6 +530,12 @@ def ti_update_state(
task_id=task_id,
)
+ # Release the task_instance row lock before running listener callbacks.
+ session.commit()
+
+ for callback in asset_callbacks:
Review Comment:
Sorry, my wording was not clear... I meant a failure during registration
itself, before the callbacks are handed back.
For example, if 9 out of 10 outlets are processed and the last one raises a
non-DB exception, `asset_callbacks` here stays empty:
```python
asset_callbacks = ()
try:
# raises inside, while processing the 10th outlet
query, updated_state, asset_callbacks =
_create_ti_state_update_query_and_update_state(...)
# so this assignment never happens and asset_callbacks is still ()
except Exception:
... # caught, TI marked FAILED, execution continues
session.commit() # the 9 events are already in the transaction, so they are
committed here
for callback in asset_callbacks: # empty, the loop never runs
callback()
```
So the 9 events are committed, but their listeners are never called.
Probably still fine as best effort, just wanted to make sure this case was
considered!
--
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]
