Andrushika commented on code in PR #66854:
URL: https://github.com/apache/airflow/pull/66854#discussion_r3605197903
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -525,14 +535,28 @@ def ti_update_state(
task_id=task_id,
map_index=map_index,
)
+ session.commit()
except Exception:
+ session.rollback()
log.warning(
"Failed to clear task state on success",
dag_id=dag_id,
run_id=run_id,
task_id=task_id,
)
+ # Register asset events now that the TI state is committed and the row
lock released, so
+ # the registration work never holds the task_instance lock. Only a
genuine RUNNING->SUCCESS
+ # transition reaches here (a duplicate SUCCESS->SUCCESS short-circuits
earlier), so a
+ # completion is never registered twice.
+ if isinstance(ti_patch_payload, TISuccessStatePayload):
+ _register_asset_changes_or_enqueue(
Review Comment:
The process could die after "state success commit" and before enqueue; when
this happens, the event will be dropped.
Imagine a scenario: `state=success` has been committed, but the process was
killed here; when the worker retries, it sees that the state is "success," so
it skips the subsequent asset registration. If the outbox is not being
implemented, it still creates the silent drop.
--
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]
