ramitkataria commented on code in PR #75253:
URL: https://github.com/apache/airflow/pull/75253#discussion_r4449070205
##########
task-sdk/src/airflow/sdk/execution_time/callback_supervisor.py:
##########
@@ -489,9 +496,23 @@ def supervise_callback(
exit_code=exit_code,
duration=end - start,
)
- if exit_code != 0:
- raise RuntimeError(f"Callback subprocess exited with code
{exit_code}")
- return exit_code
+ if exit_code == 0:
+ terminal_state = CallbackTerminalState.SUCCESS
+ else:
+ terminal_output = f"Callback subprocess exited with code
{exit_code}"
+ except Exception as e:
+ terminal_output = f"Callback supervisor error: {type(e).__name__}:
{e}"
+ log.exception("Callback supervisor error", workload_id=id)
+ raise
finally:
+ try:
+ client.callbacks.finish(UUID(id), state=terminal_state,
output=terminal_output)
Review Comment:
The execution token from `/run` seems to only live for `[execution_api]
jwt_expiration_time` (600s by default), and since the callback supervisor
doesn't heartbeat, I don't think anything refreshes it while the callback runs.
So a callback that runs longer than ~10 min without reading a Connection or
Variable would get a 403 here, which isn't retried and gets swallowed by the
`except`. On executors with no executor-event fallback, that would leave the
row `RUNNING`. Maybe the supervisor could refresh the token periodically?
##########
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:
Could a duplicate delivery (e.g. Celery `acks_late` redelivery) end up
overriding a real success here? The second supervisor's `/run` would get a 409
and fail, this `UPDATE` would write FAILED since the row is `RUNNING`, and then
the original supervisor's `finish(SUCCESS)` gets a 409 and its SUCCESS event is
ignored. Before this change the later SUCCESS would have overwritten it. Not
sure what the cleanest fix is, since this path still needs to fail a `RUNNING`
row whose supervisor died, so mostly flagging it in case it's worth handling or
noting.
--
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]
