potiuk commented on code in PR #66141:
URL: https://github.com/apache/airflow/pull/66141#discussion_r3215318912
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -1261,17 +1261,30 @@ def process_executor_events(
)
continue
+ # Callback state transitions are now driven by the supervisor
through
+ # the Execution API (POST /callbacks/{id}/run, PATCH
/callbacks/{id}/state).
+ # The in-process events from the executor are kept as a fallback
safety
+ # net for cases where the supervisor crashed before reporting a
terminal state
+
+ need_to_modify = False
+
if state == CallbackState.RUNNING:
- callback.state = CallbackState.RUNNING
cls.logger().info("Callback %s is currently running",
callback_id)
elif state == CallbackState.SUCCESS:
- callback.state = CallbackState.SUCCESS
cls.logger().info("Callback %s completed successfully",
callback_id)
+ if callback.state == CallbackState.RUNNING:
+ callback.state = CallbackState.SUCCESS
+ need_to_modify = True
elif state == CallbackState.FAILED:
- callback.state = CallbackState.FAILED
- callback.output = str(info) if info else "Execution failed"
- cls.logger().error("Callback %s failed: %s", callback_id,
callback.output)
- session.add(callback)
+ callback_output = str(info) if info else "Execution failed"
+ cls.logger().error("Callback %s failed: %s", callback_id,
callback_output)
+ if callback.state == CallbackState.RUNNING:
+ callback.state = CallbackState.FAILED
+ callback.output = callback_output
+ need_to_modify = True
Review Comment:
*Flagged by the adversarial reviewer (Codex). Cross-checked — the trace
holds.*
**[high] Executor callback failures are ignored while the DB row is still
QUEUED**
> The scheduler now only persists executor `FAILED` / `SUCCESS` events when
the callback row is already `RUNNING`. But the worker reports `RUNNING` through
the executor event channel *before* `supervise_callback()` calls
`client.callbacks.start()`, and that `RUNNING` event no longer updates the DB.
If the start API call fails before it changes the row to `RUNNING`, the worker
raises and emits a `FAILED` event while the database row is still `QUEUED`.
This branch then logs the failure but does not save it, leaving the callback in
`QUEUED` forever; `_enqueue_executor_callbacks()` only selects `PENDING`
callbacks, so the callback is neither retried nor marked failed.
>
> **Recommendation:** Make the executor-event fallback handle `QUEUED` as
well as `RUNNING` for terminal `FAILED` / `SUCCESS` events, or restore
persisting the `RUNNING` event when the current DB state is `QUEUED`. Add a
regression test where `callbacks.start()` raises and the scheduler receives a
`FAILED` callback event.
This pairs with my [finding #1
above](https://github.com/apache/airflow/pull/66141#discussion_r4259746620)
(the dead-code `RUNNING` branch). Both are symptoms of the same constraint —
the fallback assuming the API has already advanced the row to `RUNNING`. A
clean resolution would probably (a) keep the `RUNNING` branch live and update
the DB when DB-state is `QUEUED`, **or** (b) loosen the terminal-state branches
to accept `QUEUED → terminal` directly. The forward-only invariant the rest of
the PR relies on still holds either way: a callback in a terminal state never
regresses; what changes is that `QUEUED → terminal` becomes a legal one-step
jump in the safety net.
Failure modes worth testing for:
- `client.callbacks.start()` raises before the route runs (network, DNS,
5xx) — current head leaves the row stuck.
- Worker process killed (`SIGKILL`) between the executor emitting `RUNNING`
and `supervise_callback()` reaching `client.callbacks.start()` — same
stuck-`QUEUED` outcome.
- Worker process killed after `start()` returns but before subprocess exits
— the supervisor's `finish()` in `finally` runs (good), but a hard kill
bypasses `finally` too. Currently OK because DB is `RUNNING` and the executor's
terminal event drives the fallback, but worth a parametrized test in
`test_scheduler_job.py`'s forward-only matrix.
--
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]
