ferruzzi commented on code in PR #75253:
URL: https://github.com/apache/airflow/pull/75253#discussion_r4460423492
##########
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:
Yeah, I see what you are saying. I think I can see a fix for it, but it
would be a bigger change worth a followup. I'll think on it more, but the best
answer I have right now would solve both of your points: We could maybe add a
new column to the `callback` table for the heartbeat timestamp the same way
`TaskInstance` has `TaskInstance.last_heartbeat_at`.
That would need a new column with a migration, a new PATCH endpoint to
update it with a cadwyn migration, a new heartbeat loop in the supervisor to
generate it, (three?) new config options to set the new heartbeat's timeout, a
new reaper to clean up anything that goes stale... then once that's all in we
can add the liveness check that makes sure that a FAILED event can only move a
RUNNING row if the heartbeat is state. I think that would work but it feels
heavy-handed... I'm sure there's a better way. Let me think on it some more.
--
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]
