bingqin2 commented on code in PR #76534:
URL: https://github.com/apache/airflow/pull/76534#discussion_r4544985313
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -718,7 +718,11 @@ def ti_update_state(
query = TI.duration_expression_update(timezone.utcnow(), query,
session.bind)
query = query.values(state=(updated_state := TaskInstanceState.FAILED))
if ti is not None:
- _handle_fail_fast_for_dag(ti=ti, dag_id=dag_id, session=session,
dag_bag=dag_bag)
+ try:
+ with session.begin_nested():
+ _handle_fail_fast_for_dag(ti=ti, dag_id=dag_id,
session=session, dag_bag=dag_bag)
+ except Exception:
+ log.exception("Failed to apply fail_fast while setting the
task to failed")
Review Comment:
```suggestion
try:
_handle_fail_fast_for_dag(ti=ti, dag_id=dag_id,
session=session, dag_bag=dag_bag)
except Exception:
log.exception("Failed to apply fail_fast while setting the
task to failed")
session.rollback()
```
`TaskInstance.error()` commits, so a savepoint context manager can't wrap
this call. The rollback drops whatever fail_fast flushed before it failed, as
the savepoint did.
--
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]
