This is an automated email from the ASF dual-hosted git repository.
Dandandan 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 25b330778 fix(scheduler): drop an executor's cached gRPC client when
it is removed (#2310)
25b330778 is described below
commit 25b33077847701c056d36f63e8d714cf2a5ce207
Author: Daniël Heres <[email protected]>
AuthorDate: Sun Aug 16 11:25:49 2026 +0200
fix(scheduler): drop an executor's cached gRPC client when it is removed
(#2310)
* fix(scheduler): drop an executor's cached gRPC client when it is removed
`ExecutorManager` caches an `ExecutorGrpcClient` per executor in
`self.clients`, but nothing ever removes an entry: the map is only
inserted into by `get_client` and read back. `remove_executor` drops the
executor from the cluster state and leaves the client behind.
Because an executor id is a fresh uuid per executor process, a client left
in the map is never looked up again. It is not dropped either, so the
scheduler retains the client and its `Channel` for as long as it runs. A
cluster with executor churn (autoscaling, preemption, restarts) therefore
accumulates one dead client per executor that has ever registered.
Remove the cached client in `remove_executor`, before delegating to the
cluster state so that it goes even if that call fails. Every removal path
already funnels through here, as `SchedulerState::remove_executor`
documents.
Co-Authored-By: Claude <[email protected]>
* Clean up comments in executor manager
Removed unnecessary comments about cached client removal.
---------
Co-authored-by: Claude <[email protected]>
---
ballista/scheduler/src/state/executor_manager.rs | 27 ++++++++++++++++++++++++
1 file changed, 27 insertions(+)
diff --git a/ballista/scheduler/src/state/executor_manager.rs
b/ballista/scheduler/src/state/executor_manager.rs
index 1da756566..da8d7470b 100644
--- a/ballista/scheduler/src/state/executor_manager.rs
+++ b/ballista/scheduler/src/state/executor_manager.rs
@@ -381,6 +381,8 @@ impl ExecutorManager {
reason: Option<String>,
) -> Result<()> {
info!("Removing executor {executor_id}: {reason:?}");
+ // Drop the cached client
+ self.clients.remove(executor_id);
self.cluster_state.remove_executor(executor_id).await
}
@@ -608,6 +610,7 @@ mod tests {
use crate::test_utils::test_cluster_context;
use ballista_core::extension::SessionConfigExt;
use datafusion::prelude::SessionConfig;
+ use tonic::transport::Endpoint;
#[test]
fn grpc_client_max_message_size_flag_reaches_client_config() {
@@ -646,4 +649,28 @@ mod tests {
32 * 1024 * 1024
);
}
+
+ #[tokio::test]
+ async fn removing_an_executor_drops_its_cached_client() {
+ let manager = ExecutorManager::new(
+ test_cluster_context().cluster_state(),
+ Arc::new(SchedulerConfig::default()),
+ );
+
+ // `get_client` needs an executor to connect to, so cache a client
+ // directly. `connect_lazy` gives a `Channel` without a server behind
it.
+ let channel =
Endpoint::from_static("http://localhost:1";).connect_lazy();
+ manager
+ .clients
+ .insert("executor-1".to_owned(), ExecutorGrpcClient::new(channel));
+
+ manager
+ .remove_executor("executor-1", None)
+ .await
+ .expect("executor removed");
+
+ // An executor id is a fresh uuid per executor process, so a client
left
+ // behind here would never be reused, and never dropped either.
+ assert!(manager.clients.is_empty());
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
