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 e95c435b3 fix(scheduler): route launch-failure executor removal
through ExecutorLost (#2240)
e95c435b3 is described below
commit e95c435b36ba7f7250e235f1f4aefc8b93a77771
Author: Andy Grove <[email protected]>
AuthorDate: Fri Aug 7 03:42:14 2026 -0600
fix(scheduler): route launch-failure executor removal through ExecutorLost
(#2240)
* fix(scheduler): route launch-failure executor removal through ExecutorLost
The scheduler discovers a dead executor in two ways, and only one of them
armed the empty-cluster grace timer added for #2029. The heartbeat reaper
posts `ExecutorLost`, whose handler rolls back the affected graphs, cancels
their tasks and, when no executors remain, fails the jobs left behind. A
failing task launch instead took `SchedulerState::remove_executor`, which
did
the rollback inline and posted nothing.
That path also deletes the executor's heartbeat, so the reaper can never
rediscover the executor and post the event later. When a whole cluster died
while tasks were being launched onto it, no executor-loss event ever reached
the handler and the job ran forever, which is the intermittent hang behind
`killing_every_executor_terminates_the_job::case_2_aqe_on`.
Post the event from that path too, so both removal paths converge on one
handler.
Closes #2226
* refactor(scheduler): unify the two executor-removal paths
Fold SchedulerServer::remove_executor into SchedulerState::remove_executor
so the reaper, the executor_stopped RPC and a failing task launch all share
one implementation instead of two that merely agree.
Drive the new regression test through a real failing launch, injected via
the test task launcher, rather than by calling remove_executor directly, so
it also covers launch_tasks reaching that path.
---
ballista/scheduler/src/scheduler_server/grpc.rs | 3 +-
ballista/scheduler/src/scheduler_server/mod.rs | 28 ++++--------
.../src/scheduler_server/query_stage_scheduler.rs | 50 ++++++++++++++++++++++
ballista/scheduler/src/state/mod.rs | 42 ++++++++++--------
ballista/scheduler/src/test_utils.rs | 47 ++++++++++++++------
chaos-testing/README.md | 15 +++++++
6 files changed, 135 insertions(+), 50 deletions(-)
diff --git a/ballista/scheduler/src/scheduler_server/grpc.rs
b/ballista/scheduler/src/scheduler_server/grpc.rs
index 3b41e918c..7ccf84b3b 100644
--- a/ballista/scheduler/src/scheduler_server/grpc.rs
+++ b/ballista/scheduler/src/scheduler_server/grpc.rs
@@ -737,7 +737,6 @@ impl<T: 'static + AsLogicalPlan, U: 'static +
AsExecutionPlan> SchedulerGrpc
executor_id, reason
);
- let executor_manager = self.state.executor_manager.clone();
let event_sender =
self.query_stage_event_loop.get_sender().map_err(|e| {
let msg = format!("Get query stage event loop error due to {e:?}");
error!("{msg}");
@@ -745,7 +744,7 @@ impl<T: 'static + AsLogicalPlan, U: 'static +
AsExecutionPlan> SchedulerGrpc
})?;
Self::remove_executor(
- executor_manager,
+ self.state.clone(),
event_sender,
&executor_id,
Some(reason),
diff --git a/ballista/scheduler/src/scheduler_server/mod.rs
b/ballista/scheduler/src/scheduler_server/mod.rs
index ca439dfeb..67c175dee 100644
--- a/ballista/scheduler/src/scheduler_server/mod.rs
+++ b/ballista/scheduler/src/scheduler_server/mod.rs
@@ -35,13 +35,11 @@ use crate::cluster::{BallistaCluster,
ClusterStateEventStream, JobStateEventStre
use crate::config::SchedulerConfig;
use crate::metrics::SchedulerMetricsCollector;
use ballista_core::serde::scheduler::{ExecutorData, ExecutorMetadata};
-use log::{debug, error, warn};
+use log::{debug, warn};
use crate::scheduler_server::event::{QueryStageSchedulerEvent, SubmitPlan};
use crate::scheduler_server::query_stage_scheduler::QueryStageScheduler;
-use crate::state::executor_manager::ExecutorManager;
-
use crate::state::SchedulerState;
use crate::state::task_manager::TaskLauncher;
@@ -364,7 +362,7 @@ impl<T: 'static + AsLogicalPlan, U: 'static +
AsExecutionPlan> SchedulerServer<T
// If executor is expired, remove it immediately
Self::remove_executor(
- state.executor_manager.clone(),
+ state.clone(),
sender_clone,
&executor_id,
Some(stop_reason.clone()),
@@ -389,8 +387,11 @@ impl<T: 'static + AsLogicalPlan, U: 'static +
AsExecutionPlan> SchedulerServer<T
Ok(())
}
+ /// Removes an executor after `wait_secs`, in the background. The removal
+ /// itself is [`SchedulerState::remove_executor`], so this path and the one
+ /// taken when a task launch fails cannot drift apart.
pub(crate) fn remove_executor(
- executor_manager: ExecutorManager,
+ state: Arc<SchedulerState<T, U>>,
event_sender: EventSender<QueryStageSchedulerEvent>,
executor_id: &str,
reason: Option<String>,
@@ -402,20 +403,9 @@ impl<T: 'static + AsLogicalPlan, U: 'static +
AsExecutionPlan> SchedulerServer<T
// Wait for `wait_secs` before removing executor
tokio::time::sleep(Duration::from_secs(wait_secs)).await;
- // Update the executor manager immediately here
- if let Err(e) = executor_manager
- .remove_executor(&executor_id, reason.clone())
- .await
- {
- error!("error removing executor {executor_id}: {e:?}");
- }
-
- if let Err(e) = event_sender
-
.post_event(QueryStageSchedulerEvent::ExecutorLost(executor_id, reason))
- .await
- {
- error!("error sending ExecutorLost event: {e:?}");
- }
+ state
+ .remove_executor(&executor_id, reason, &event_sender)
+ .await;
});
}
diff --git a/ballista/scheduler/src/scheduler_server/query_stage_scheduler.rs
b/ballista/scheduler/src/scheduler_server/query_stage_scheduler.rs
index ef3d118e8..2ccf5a9da 100644
--- a/ballista/scheduler/src/scheduler_server/query_stage_scheduler.rs
+++ b/ballista/scheduler/src/scheduler_server/query_stage_scheduler.rs
@@ -796,6 +796,56 @@ mod tests {
Ok(())
}
+ /// The same outcome as
`test_running_job_fails_when_all_executors_are_lost`,
+ /// but the cluster's death is discovered by a failing task launch instead
of
+ /// by the heartbeat reaper. That path removes the executor — heartbeat
+ /// included — so the reaper can never rediscover it afterwards, making the
+ /// launch failure the only chance the scheduler gets to notice the cluster
+ /// is empty. See
<https://github.com/apache/datafusion-ballista/issues/2226>
+ #[tokio::test]
+ async fn test_running_job_fails_when_launch_failure_loses_last_executor()
-> Result<()>
+ {
+ let plan = test_plan(10);
+
+ let metrics_collector = Arc::new(TestMetricsCollector::default());
+
+ let mut test = SchedulerTest::new(
+ SchedulerConfig::default()
+ .with_scheduler_policy(TaskSchedulingPolicy::PushStaged)
+ .with_no_executors_grace_period_seconds(0),
+ metrics_collector.clone(),
+ 1,
+ 1,
+ None,
+ )
+ .await?;
+
+ // The only executor dies before its first task is launched, so the
+ // scheduler learns of the loss from `launch_multi_task` rather than
+ // from an expired heartbeat.
+ test.make_launches_fail("virtual-executor-0");
+
+ let job_id = test.submit("", &plan).await?;
+
+ let job_id_ref = &job_id;
+ let test_ref = &test;
+ let failed = await_condition(Duration::from_millis(100), 50, || async
move {
+ let status = test_ref.job_status(job_id_ref).await?;
+ Ok(matches!(
+ status.and_then(|s| s.status),
+ Some(job_status::Status::Failed(_))
+ ))
+ })
+ .await?;
+ assert!(
+ failed,
+ "job should be failed after a launch failure lost the last
executor, but status was {:?}",
+ test.job_status(&job_id).await?
+ );
+
+ Ok(())
+ }
+
#[tokio::test]
async fn test_running_job_survives_partial_executor_loss() -> Result<()> {
let plan = test_plan(10);
diff --git a/ballista/scheduler/src/state/mod.rs
b/ballista/scheduler/src/state/mod.rs
index d86488e8f..f4a6ebec4 100644
--- a/ballista/scheduler/src/state/mod.rs
+++ b/ballista/scheduler/src/state/mod.rs
@@ -190,7 +190,7 @@ impl<T: 'static + AsLogicalPlan, U: 'static +
AsExecutionPlan> SchedulerState<T,
let state = self.clone();
tokio::spawn(async move {
let mut if_revive = false;
- match state.launch_tasks(schedulable_tasks).await {
+ match state.launch_tasks(schedulable_tasks, &sender).await {
Ok(unassigned_executor_slots) => {
if !unassigned_executor_slots.is_empty() {
if let Err(e) = state
@@ -222,33 +222,37 @@ impl<T: 'static + AsLogicalPlan, U: 'static +
AsExecutionPlan> SchedulerState<T,
/// Remove an executor.
/// 1. The executor related info will be removed from [`ExecutorManager`]
- /// 2. All of affected running execution graph will be rolled backed
- /// 3. All of the running tasks of the affected running stages will be
cancelled
+ /// 2. A [`QueryStageSchedulerEvent::ExecutorLost`] is posted, which rolls
+ /// back the affected running execution graphs, cancels their running
+ /// tasks, and — when this was the last executor — arms the grace timer
+ /// that fails the jobs left behind on an empty cluster.
+ ///
+ /// Every removal path must go through here, because step 1 also drops the
+ /// executor's heartbeat: once it is gone, nothing else can notice the
+ /// executor is missing and post the event later.
+ /// See <https://github.com/apache/datafusion-ballista/issues/2226>
pub(crate) async fn remove_executor(
&self,
executor_id: &str,
reason: Option<String>,
+ sender: &EventSender<QueryStageSchedulerEvent>,
) {
if let Err(e) = self
.executor_manager
- .remove_executor(executor_id, reason)
+ .remove_executor(executor_id, reason.clone())
.await
{
warn!("Fail to remove executor {executor_id}: {e}");
}
- match self.task_manager.executor_lost(executor_id).await {
- Ok(tasks) => {
- if !tasks.is_empty()
- && let Err(e) =
- self.executor_manager.cancel_running_tasks(tasks).await
- {
- warn!("Fail to cancel running tasks due to {e:?}");
- }
- }
- Err(e) => {
- error!("TaskManager error to handle Executor {executor_id}
lost: {e}");
- }
+ if let Err(e) = sender
+ .post_event(QueryStageSchedulerEvent::ExecutorLost(
+ executor_id.to_owned(),
+ reason,
+ ))
+ .await
+ {
+ error!("Fail to post ExecutorLost for executor {executor_id}:
{e:?}");
}
}
@@ -260,6 +264,7 @@ impl<T: 'static + AsLogicalPlan, U: 'static +
AsExecutionPlan> SchedulerState<T,
async fn launch_tasks(
&self,
bound_tasks: Vec<BoundTask>,
+ sender: &EventSender<QueryStageSchedulerEvent>,
) -> Result<Vec<ExecutorSlot>> {
// Put tasks to the same executor together
// And put tasks belonging to the same stage together for creating
MultiTaskDefinition
@@ -292,6 +297,7 @@ impl<T: 'static + AsLogicalPlan, U: 'static +
AsExecutionPlan> SchedulerState<T,
let n_tasks: usize = tasks.iter().map(|stage_tasks|
stage_tasks.len()).sum();
let state = self.clone();
+ let sender = sender.clone();
let join_handle = tokio::spawn(async move {
let success = match state
.executor_manager
@@ -309,7 +315,9 @@ impl<T: 'static + AsLogicalPlan, U: 'static +
AsExecutionPlan> SchedulerState<T,
// It's OK to remove executor aggressively,
// since if the executor is in healthy state, it
will be registered again.
- state.remove_executor(&executor_id,
Some(err_msg)).await;
+ state
+ .remove_executor(&executor_id, Some(err_msg),
&sender)
+ .await;
false
} else {
diff --git a/ballista/scheduler/src/test_utils.rs
b/ballista/scheduler/src/test_utils.rs
index 496aa8d58..08bc7b9f7 100644
--- a/ballista/scheduler/src/test_utils.rs
+++ b/ballista/scheduler/src/test_utils.rs
@@ -19,7 +19,7 @@ use ballista_core::error::{BallistaError, Result};
use ballista_core::extension::SessionConfigExt;
use ballista_core::{JobId, JobStatusSubscriber};
use datafusion::catalog::Session;
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
use std::future::Future;
use std::sync::Arc;
use std::time::Duration;
@@ -350,6 +350,10 @@ impl TaskLauncher for BlackholeTaskLauncher {
pub struct VirtualTaskLauncher {
sender: Sender<(String, Vec<TaskStatus>)>,
executors: HashMap<String, VirtualExecutor>,
+ /// Executors whose launches must fail, as if the process had died between
+ /// binding and launch. Shared with the owning [`SchedulerTest`], which
adds
+ /// to it through [`SchedulerTest::make_launches_fail`].
+ unreachable: Arc<Mutex<HashSet<String>>>,
}
#[async_trait::async_trait]
@@ -360,6 +364,13 @@ impl TaskLauncher for VirtualTaskLauncher {
tasks: Vec<MultiTaskDefinition>,
_executor_manager: &ExecutorManager,
) -> Result<()> {
+ if self.unreachable.lock().contains(&executor.id) {
+ return Err(BallistaError::Internal(format!(
+ "test: executor {} is unreachable",
+ executor.id
+ )));
+ }
+
let virtual_executor = self.executors.get(&executor.id).ok_or_else(|| {
BallistaError::Internal(format!(
"No virtual executor with ID {} found",
@@ -386,6 +397,7 @@ pub struct SchedulerTest {
scheduler: SchedulerServer<LogicalPlanNode, PhysicalPlanNode>,
session_config: SessionConfig,
status_receiver: Option<Receiver<(String, Vec<TaskStatus>)>>,
+ unreachable_executors: Arc<Mutex<HashSet<String>>>,
}
impl SchedulerTest {
@@ -422,9 +434,12 @@ impl SchedulerTest {
let (status_sender, status_receiver) = channel(1000);
+ let unreachable_executors: Arc<Mutex<HashSet<String>>> =
Arc::default();
+
let launcher = VirtualTaskLauncher {
sender: status_sender,
executors: executors.clone(),
+ unreachable: unreachable_executors.clone(),
};
let mut scheduler: SchedulerServer<LogicalPlanNode, PhysicalPlanNode> =
@@ -466,6 +481,7 @@ impl SchedulerTest {
scheduler,
session_config,
status_receiver: Some(status_receiver),
+ unreachable_executors,
})
}
@@ -548,21 +564,28 @@ impl SchedulerTest {
.await
}
- /// Simulates the loss of an executor: deregisters it from the executor
- /// manager and posts the `ExecutorLost` event. This mirrors the reaper's
+ /// Simulates the loss of an executor. This mirrors the reaper's
/// `remove_executor` path without waiting out the heartbeat timeout.
pub async fn lose_executor(&self, executor_id: &str) -> Result<()> {
- let reason = Some("test: executor lost".to_owned());
self.scheduler
.state
- .executor_manager
- .remove_executor(executor_id, reason.clone())
- .await?;
- self.post_scheduler_event(QueryStageSchedulerEvent::ExecutorLost(
- executor_id.to_owned(),
- reason,
- ))
- .await
+ .remove_executor(
+ executor_id,
+ Some("test: executor lost".to_owned()),
+ &self.scheduler.query_stage_event_loop.get_sender()?,
+ )
+ .await;
+ Ok(())
+ }
+
+ /// Makes every subsequent task launch onto `executor_id` fail, as if the
+ /// process had died after its tasks were bound to it. The scheduler then
+ /// discovers the loss through the failing launch rather than through
+ /// heartbeat expiry.
+ pub fn make_launches_fail(&self, executor_id: &str) {
+ self.unreachable_executors
+ .lock()
+ .insert(executor_id.to_owned());
}
/// Returns the current status of a job, if known.
diff --git a/chaos-testing/README.md b/chaos-testing/README.md
index 35cc551d2..88b198b57 100644
--- a/chaos-testing/README.md
+++ b/chaos-testing/README.md
@@ -340,6 +340,21 @@ fail the job with a clear error. The scenario turns that
grace down via the
cluster builder and asserts the query fails with an error naming the executor
loss.
+**Second path, same hang**
([#2226](https://github.com/apache/datafusion-ballista/issues/2226)).
+Scenario G kept failing intermittently after that fix, because the scheduler
+has _two_ ways of discovering an executor is gone and only one of them armed
+the grace timer. The heartbeat reaper posts `ExecutorLost`, which is where the
+timer lives; a failing task launch instead took
+`SchedulerState::remove_executor`, which rolled the graphs back inline and
+posted nothing. That path also deletes the executor's heartbeat, so the reaper
+could never rediscover it and post the event later — the job was left running
+on an empty cluster with no further executor-loss event to come. Whether a kill
+lands during a task launch or between heartbeats is a timing race, which is why
+the scenario failed only sometimes. The fix routes both removal paths through
+the same `ExecutorLost` event; the unit test
+`test_running_job_fails_when_launch_failure_loses_last_executor` covers the
+launch-failure path deterministically.
+
### For comparison: the heartbeat-expiry path does recover
Ballista's HA recovery is not uniformly broken. Whenever the kill in Scenario
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
