This is an automated email from the ASF dual-hosted git repository.
milenkovicm 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 8778c5496 fix: REST API does not show running jobs (#1703)
8778c5496 is described below
commit 8778c54967165862d2ae4f580212facbcbebafcd
Author: gittihub-jpg <[email protected]>
AuthorDate: Sun May 17 08:40:42 2026 +0200
fix: REST API does not show running jobs (#1703)
* fix(api): show running jobs in REST API (#1696)
Fixes #1696
Signed-off-by: gittihub-jpg <[email protected]>
* fix: address PR feedback for apache/datafusion-ballista#1696
Signed-off-by: gittihub-jpg <[email protected]>
* fix: apply inline suggestion — update get_all_jobs doc comment
Add '/failed' to the list of job states in the get_all_jobs docstring,
as requested by maintainer review on PR #1703.
Signed-off-by: gittihub-jpg <[email protected]>
* fix: address remaining PR review comments
- Deprecate get_jobs() with note to use get_all_jobs instead; update
doc to clarify it only covers the active cache (running/queued)
- Extract start_time/end_time from job status variants instead of
hardcoding 0: running jobs get started_at, successful/failed jobs
get started_at and ended_at, queued jobs stay 0 (no start time yet)
- Remove unused status filter in get_all_jobs_with_status for the
job_status fallback path
- Add warn! when a job_id is not found in any of the three lookups
Co-Authored-By: Claude Sonnet 4.6 <[email protected]>
* address comments from review
---------
Signed-off-by: gittihub-jpg <[email protected]>
Co-authored-by: Claude Sonnet 4.6 <[email protected]>
Co-authored-by: Marko Milenković <[email protected]>
---
ballista/scheduler/src/api/handlers.rs | 9 ++++--
ballista/scheduler/src/cluster/memory.rs | 11 +++++++
ballista/scheduler/src/cluster/mod.rs | 3 ++
ballista/scheduler/src/state/task_manager.rs | 46 ++++++++++++++++++++++++++--
4 files changed, 64 insertions(+), 5 deletions(-)
diff --git a/ballista/scheduler/src/api/handlers.rs
b/ballista/scheduler/src/api/handlers.rs
index 3e3329c75..2a6d674aa 100644
--- a/ballista/scheduler/src/api/handlers.rs
+++ b/ballista/scheduler/src/api/handlers.rs
@@ -312,7 +312,7 @@ pub async fn get_jobs<
) -> Result<impl IntoResponse, SchedulerErrorResponse> {
let state = &data_server.state;
- let jobs = state.task_manager.get_jobs().await.map_err(|e| {
+ let jobs = state.task_manager.get_all_jobs().await.map_err(|e| {
tracing::error!("Error occurred while getting jobs, reason: {e:?}");
SchedulerErrorResponse::new(StatusCode::INTERNAL_SERVER_ERROR)
})?;
@@ -327,8 +327,11 @@ pub async fn get_jobs<
// calculate progress based on completed stages for now, but we
could use completed
// tasks in the future to make this more accurate
- let percent_complete =
- ((job.completed_stages as f32 / job.num_stages as f32) *
100_f32) as u8;
+ let percent_complete = if job.num_stages == 0 {
+ 0
+ } else {
+ ((job.completed_stages as f32 / job.num_stages as f32) *
100_f32) as u8
+ };
JobResponse {
job_id: job.job_id.to_string(),
job_name: job.job_name.to_string(),
diff --git a/ballista/scheduler/src/cluster/memory.rs
b/ballista/scheduler/src/cluster/memory.rs
index 6875d4de8..54eba10b1 100644
--- a/ballista/scheduler/src/cluster/memory.rs
+++ b/ballista/scheduler/src/cluster/memory.rs
@@ -467,6 +467,17 @@ impl JobState for InMemoryJobState {
.collect())
}
+ async fn get_all_jobs(&self) -> Result<HashSet<String>> {
+ let mut all_jobs: HashSet<String> = self
+ .queued_jobs
+ .iter()
+ .map(|pair| pair.key().clone())
+ .collect();
+ all_jobs.extend(self.running_jobs.iter().map(|pair|
pair.key().clone()));
+ all_jobs.extend(self.completed_jobs.iter().map(|pair|
pair.key().clone()));
+ Ok(all_jobs)
+ }
+
fn accept_job(&self, job_id: &str, job_name: &str, queued_at: u64) ->
Result<()> {
self.queued_jobs
.insert(job_id.to_string(), (job_name.to_string(), queued_at));
diff --git a/ballista/scheduler/src/cluster/mod.rs
b/ballista/scheduler/src/cluster/mod.rs
index 80b3ba1ee..5c4767538 100644
--- a/ballista/scheduler/src/cluster/mod.rs
+++ b/ballista/scheduler/src/cluster/mod.rs
@@ -300,6 +300,9 @@ pub trait JobState: Send + Sync {
/// Returns the set of all active job IDs.
async fn get_jobs(&self) -> Result<HashSet<String>>;
+ /// Returns the set of all job IDs including running, queued, and
completed jobs.
+ async fn get_all_jobs(&self) -> Result<HashSet<String>>;
+
/// Returns the status of the specified job.
async fn get_job_status(&self, job_id: &str) -> Result<Option<JobStatus>>;
diff --git a/ballista/scheduler/src/state/task_manager.rs
b/ballista/scheduler/src/state/task_manager.rs
index b668e8342..20aa6990e 100644
--- a/ballista/scheduler/src/state/task_manager.rs
+++ b/ballista/scheduler/src/state/task_manager.rs
@@ -346,8 +346,12 @@ impl<T: 'static + AsLogicalPlan, U: 'static +
AsExecutionPlan> TaskManager<T, U>
Arc::new(ret)
}
- /// Get a list of active job ids
- pub async fn get_jobs(&self) -> Result<Vec<JobOverview>> {
+ /// Get a list of jobs from the active cache only (running and queued
jobs).
+ ///
+ /// Unlike [`Self::get_all_jobs`], this does not include completed or
failed jobs
+ /// that have been evicted from the cache. Prefer [`Self::get_all_jobs`]
for a
+ /// complete view.
+ pub async fn get_running_jobs(&self) -> Result<Vec<JobOverview>> {
let job_ids = self.state.get_jobs().await?;
let mut jobs = vec![];
@@ -366,6 +370,44 @@ impl<T: 'static + AsLogicalPlan, U: 'static +
AsExecutionPlan> TaskManager<T, U>
Ok(jobs)
}
+ /// Get all jobs optionally filtered by status.
+ /// When `status` is None, returns all jobs regardless of status.
+ pub async fn get_all_jobs(&self) -> Result<Vec<JobOverview>> {
+ let job_ids = self.state.get_all_jobs().await?;
+
+ let mut jobs = vec![];
+ for job_id in &job_ids {
+ if let Some(cached) = self.get_active_execution_graph(job_id) {
+ let graph = cached.read().await;
+ jobs.push(graph.deref().into());
+ } else if let Some(graph) =
self.state.get_execution_graph(job_id).await? {
+ jobs.push((&graph).into());
+ } else if let Some(job_status) =
self.state.get_job_status(job_id).await? {
+ let (start_time, end_time) = match &job_status.status {
+ Some(job_status::Status::Running(r)) => (r.started_at, 0),
+ Some(job_status::Status::Successful(s)) => (s.started_at,
s.ended_at),
+ Some(job_status::Status::Failed(f)) => (f.started_at,
f.ended_at),
+ // Queued jobs have no start or end time yet
+ _ => (0, 0),
+ };
+ jobs.push(JobOverview {
+ job_id: job_status.job_id.clone(),
+ job_name: job_status.job_name.clone(),
+ status: job_status,
+ start_time,
+ end_time,
+ num_stages: 0,
+ completed_stages: 0,
+ });
+ } else {
+ warn!(
+ "Job {job_id} not found in active cache, execution graph,
or job status"
+ );
+ }
+ }
+ Ok(jobs)
+ }
+
/// Get the status of of a job. First look in the active cache.
/// If no one found, then in the Active/Completed jobs, and then in Failed
jobs
pub async fn get_job_status(&self, job_id: &str) ->
Result<Option<JobStatus>> {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
