You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by ag...@apache.org on 2026/09/27 17:08:49 UTC
(datafusion-ballista) branch main updated: fix(scheduler): fail jobs whose tasks cannot be prepared (#2492)
This is an automated email from the ASF dual-hosted git repository.
andygrove pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-ballista.git
The following commit(s) were added to refs/heads/main by this push:
new 8ebfa6e59 fix(scheduler): fail jobs whose tasks cannot be prepared (#2492)
8ebfa6e59 is described below
commit 8ebfa6e593c4622a8602ad741c40a455f13a8978
Author: Andy Grove <ag...@apache.org>
AuthorDate: Sun Sep 27 11:08:44 2026 -0600
fix(scheduler): fail jobs whose tasks cannot be prepared (#2492)
A task whose definition cannot be prepared was logged and dropped. It was
never sent to an executor, so no status ever came back for it: the job
stayed Running for the life of the scheduler and its vcores were never
refunded. Report such jobs alongside the ones executors reject, so they
are failed and their slots freed the same way.
---
ballista/scheduler/src/state/mod.rs | 95 +++++++++++++++++++++++++++-
ballista/scheduler/src/state/task_manager.rs | 28 ++++++--
2 files changed, 114 insertions(+), 9 deletions(-)
diff --git a/ballista/scheduler/src/state/mod.rs b/ballista/scheduler/src/state/mod.rs
index b544a2be2..b01b8b760 100644
--- a/ballista/scheduler/src/state/mod.rs
+++ b/ballista/scheduler/src/state/mod.rs
@@ -215,8 +215,9 @@ impl<T: 'static + AsLogicalPlan, U: 'static + AsExecutionPlan> SchedulerState<T,
if let Err(e) = sender
.post_event(QueryStageSchedulerEvent::JobRunningFailed {
job_id: job,
- fail_message: "task serialization failed by executor"
- .to_string(),
+ fail_message:
+ "task could not be prepared or was rejected by executor"
+ .to_string(),
queued_at: timestamp_millis(),
failed_at: timestamp_millis(),
})
@@ -608,4 +609,94 @@ mod tests {
Ok(())
}
+
+ /// A task whose definition cannot be prepared is never sent to an
+ /// executor, so no status will ever come back for it. Its job has to be
+ /// failed and its slot refunded, or the job stays `Running` for the life of
+ /// the scheduler and the executor permanently loses the vcore.
+ #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+ async fn launch_tasks_fails_jobs_whose_tasks_cannot_be_prepared() -> Result<()> {
+ let good_job = JobId::from("job-good");
+ let doomed_job = JobId::from("job-doomed");
+ let orphan = JobId::from("job-orphan");
+
+ let state: SchedulerState<LogicalPlanNode, PhysicalPlanNode> =
+ SchedulerState::new_with_task_launcher(
+ test_cluster_context(),
+ BallistaCodec::default(),
+ "localhost:50050".to_owned(),
+ Arc::new(SchedulerConfig::default()),
+ Arc::new(RejectOne {
+ reject: JobId::from("no-such-job"),
+ }),
+ );
+
+ let vcores = 8;
+ state
+ .executor_manager
+ .register_executor(
+ ExecutorMetadata {
+ id: "executor-1".to_string(),
+ host: String::default(),
+ port: 0,
+ grpc_port: 0,
+ specification: ExecutorSpecification::default().with_vcores(vcores),
+ os_info: ExecutorOperatingSystemSpecification::default(),
+ },
+ ExecutorData {
+ executor_id: "executor-1".to_string(),
+ total_vcores: vcores,
+ available_vcores: vcores,
+ },
+ )
+ .await?;
+
+ let ctx = state
+ .session_manager
+ .create_or_update_session("session", &SessionConfig::new_with_ballista())
+ .await?;
+ for job_id in [&good_job, &doomed_job] {
+ state
+ .task_manager
+ .queue_job(job_id, "", timestamp_millis())?;
+ state
+ .task_manager
+ .submit_job(
+ job_id,
+ "",
+ ctx.clone(),
+ &agg_plan(),
+ timestamp_millis(),
+ None,
+ )
+ .await?;
+ }
+
+ let mut bound = state
+ .executor_manager
+ .bind_schedulable_tasks(state.task_manager.get_running_job_cache())
+ .await?;
+
+ // Point one job's tasks at a job the task manager has never heard of,
+ // which preparation refuses. Any preparation failure, such as a plan
+ // the physical codec cannot encode, lands in the same branch.
+ let mut orphaned = 0;
+ for (_, task) in bound.iter_mut() {
+ if task.key.job_id == doomed_job {
+ task.key.job_id = orphan.clone();
+ orphaned += 1;
+ }
+ }
+ assert!(orphaned > 0 && bound.len() > orphaned);
+
+ let (tx_event, _rx_event) = tokio::sync::mpsc::channel(100);
+ let sender = EventSender::new(tx_event);
+ let (unassigned_slots, failed_jobs) = state.launch_tasks(bound, &sender).await?;
+
+ assert_eq!(failed_jobs, HashSet::from([orphan]));
+ let freed: u32 = unassigned_slots.iter().map(|(_, n)| *n).sum();
+ assert_eq!(freed as usize, orphaned);
+
+ Ok(())
+ }
}
diff --git a/ballista/scheduler/src/state/task_manager.rs b/ballista/scheduler/src/state/task_manager.rs
index e0ef644bf..b99a1d11b 100644
--- a/ballista/scheduler/src/state/task_manager.rs
+++ b/ballista/scheduler/src/state/task_manager.rs
@@ -832,7 +832,14 @@ impl<T: 'static + AsLogicalPlan, U: 'static + AsExecutionPlan> TaskManager<T, U>
}
}
- /// Launch the given tasks on the specified executor
+ /// Launch the given tasks on the specified executor.
+ ///
+ /// Returns the jobs that cannot run: those the executor rejected, and
+ /// those whose task definitions could not be prepared here. Tasks of the
+ /// latter were never sent anywhere and will never report a status, so the
+ /// caller has to fail the job and refund the tasks' slots, exactly as for a
+ /// rejection. `Err` is only for the launch RPC itself failing, which says
+ /// the executor is sick, not the plan.
pub(crate) async fn launch_multi_task(
&self,
executor: &ExecutorMetadata,
@@ -840,20 +847,27 @@ impl<T: 'static + AsLogicalPlan, U: 'static + AsExecutionPlan> TaskManager<T, U>
executor_manager: &ExecutorManager,
) -> Result<HashSet<JobId>> {
let mut multi_tasks = vec![];
+ let mut failed_jobs = HashSet::new();
for stage_tasks in tasks {
+ let job_id = stage_tasks.first().map(|task| task.key.job_id.clone());
match self.prepare_multi_task_definition(stage_tasks) {
Ok(stage_tasks) => multi_tasks.extend(stage_tasks),
- Err(e) => error!("Fail to prepare task definition: {e:?}"),
+ Err(e) => {
+ error!("Fail to prepare task definition: {e:?}");
+ failed_jobs.extend(job_id);
+ }
}
}
if !multi_tasks.is_empty() {
- self.launcher
- .launch_tasks(executor, multi_tasks, executor_manager)
- .await
- } else {
- Ok(HashSet::new())
+ failed_jobs.extend(
+ self.launcher
+ .launch_tasks(executor, multi_tasks, executor_manager)
+ .await?,
+ );
}
+
+ Ok(failed_jobs)
}
#[allow(dead_code)]
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org