ferruzzi commented on code in PR #75253:
URL: https://github.com/apache/airflow/pull/75253#discussion_r4461683724
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -1538,14 +1539,52 @@ def process_executor_events(
)
continue
+ # The supervisor reports the outcome through PATCH
/callbacks/{id}/state, this is
+ # in case a supervisor dies before reporting.
+ if callback.state not in (CallbackState.QUEUED,
CallbackState.RUNNING):
+ cls.logger().debug(
+ "Callback %s is already %s; ignoring executor event with
state %s",
+ callback_id,
+ callback.state,
+ state,
+ )
+ continue
+
+ new_values: dict[str, Any] = {"state": state}
+ if state == CallbackState.FAILED:
+ new_values["output"] = str(info) if info else "Execution
failed"
+
+ # The QUEUED/RUNNING check above reads an ORM snapshot, and the
supervisor can report a
+ # terminal state through the API before this write lands.
Repeating it here stops a
+ # terminal row moving.
+ updated = cast(
+ "CursorResult",
+ session.execute(
+ update(Callback)
+ .where(
+ Callback.id == callback_uuid,
+ Callback.state.in_((CallbackState.QUEUED,
CallbackState.RUNNING)),
Review Comment:
I can't think of a better solution and that's a relatively big bite of work
to close a race condition that already exists. Are you alright with throwing
that in the backlog and getting this merged? I don't believe (I could be
wrong) this is making the existing state worse than main or introducing a new
bug, right?
--
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]
