This is an automated email from the ASF dual-hosted git repository. martin-g pushed a commit to branch 2226-emit-ExecutorLost-after-launch_tasks in repository https://gitbox.apache.org/repos/asf/datafusion-ballista.git
commit 904ecebe823e7781b1fd73f92b0b62612446233a Author: Martin Tzvetanov Grigorov <[email protected]> AuthorDate: Thu Aug 6 16:47:19 2026 +0300 fix(scheduler): Emit ExecutorLost after failed launch_tasks() Fixes #2226 `launch_tasks` removes failed executors via `SchedulerState::remove_executor`, which resets affected stages but does NOT post an `ExecutorLost` event. The grace-period timer that fails jobs on an empty cluster is only armed inside the `ExecutorLost` event handler. If `launch_tasks` is the path that removes the last executor (e.g. AQE plans a new stage right as executors die), no `ExecutorLost` event is ever posted, the grace timer never fires, and the job hangs forever. Post `ExecutorLost` for each failed executor so the event handler's empty-cluster check runs. Calling `task_manager.executor_lost()` twice for the same executor is safe (idempotent: the second call finds nothing to reset). --- ballista/scheduler/src/state/mod.rs | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/ballista/scheduler/src/state/mod.rs b/ballista/scheduler/src/state/mod.rs index d86488e8f..514c99477 100644 --- a/ballista/scheduler/src/state/mod.rs +++ b/ballista/scheduler/src/state/mod.rs @@ -190,9 +190,16 @@ impl<T: 'static + AsLogicalPlan, U: 'static + AsExecutionPlan> SchedulerState<T, let state = self.clone(); tokio::spawn(async move { let mut if_revive = false; + // Collect executor IDs that failed during launch so we can + // post `ExecutorLost` events for them below. + let mut failed_executor_ids: Vec<String> = vec![]; match state.launch_tasks(schedulable_tasks).await { Ok(unassigned_executor_slots) => { if !unassigned_executor_slots.is_empty() { + failed_executor_ids = unassigned_executor_slots + .iter() + .map(|(id, _)| id.clone()) + .collect(); if let Err(e) = state .executor_manager .unbind_tasks(unassigned_executor_slots) @@ -208,6 +215,21 @@ impl<T: 'static + AsLogicalPlan, U: 'static + AsExecutionPlan> SchedulerState<T, if_revive = true; } } + + for executor_id in failed_executor_ids { + if let Err(e) = sender + .post_event(QueryStageSchedulerEvent::ExecutorLost( + executor_id.clone(), + Some(format!( + "Executor {executor_id} removed after launch failure" + )), + )) + .await + { + error!("Fail to send ExecutorLost event for {executor_id}: {e:?}"); + } + } + if if_revive && let Err(e) = sender .post_event(QueryStageSchedulerEvent::ReviveOffers) --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
