(datafusion-ballista) branch main updated: fix: enable the executor client pool by default (#2069)

[ag...@apache.org]

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 66af2392a fix: enable the executor client pool by default (#2069)
66af2392a is described below

commit 66af2392a26102dbfa4f147519600fbead6069ab
Author: Andy Grove <ag...@apache.org>
AuthorDate: Thu Jul 16 12:13:57 2026 -0600

    fix: enable the executor client pool by default (#2069)
    
    `client_ttl` defaulted to 0, which disables the client pool entirely. Without a
    pool the shuffle-read path opens and discards a new TCP connection per fetch, so
    a single shuffle-heavy query exhausts the host's ephemeral ports and fails with
    `AddrNotAvailable` — as does every query after it, until the sockets leave
    TIME_WAIT.
    
    A single executor never shows this, because shuffle reads are served from local
    files and no remote fetch happens. It appears as soon as a second executor
    exists, which is every real deployment.
    
    Default `client_ttl` to 30 seconds so the pool is on unless explicitly disabled.
    The pooling machinery already exists and the shuffle reader is written assuming
    it ("multiplexing over few connections"); it was simply off unless the operator
    knew to pass `--client-ttl`.
    
    Measured on the SF1 TPC-DS gate with 1 scheduler + 2 executors, all defaults:
    
                  before   after
      peak TIME_WAIT   16343      12
      gate         11 OK/77 FAIL   88 OK/0 FAIL
    
    The ephemeral range on the test host is 16384 ports, so one query consumed all
    of it.
---
 ballista/executor/src/config.rs           | 23 ++++++++++++++++++++---
 ballista/executor/src/executor_process.rs |  7 +++++--
 2 files changed, 25 insertions(+), 5 deletions(-)

diff --git a/ballista/executor/src/config.rs b/ballista/executor/src/config.rs
index 48b5a28c9..dc6d8fb19 100644
--- a/ballista/executor/src/config.rs
+++ b/ballista/executor/src/config.rs
@@ -180,11 +180,12 @@ pub struct Config {
         help = "Max number of sessions whose shared runtime state (object-store clients, Parquet footer cache) is retained on the executor (LRU). 0 disables caching."
     )]
     pub session_runtime_cache_capacity: usize,
-    /// Number of seconds established client connection should be cached if not used (0 means no cache)
+    /// Number of seconds an established client connection is cached while idle
+    /// (0 disables the cache, so every shuffle fetch opens a new connection)
     #[arg(
         long,
-        default_value_t = 0,
-        help = "Number of seconds established client connection should be cached if not used (0 means no cache, connection will be disposed)."
+        default_value_t = 30,
+        help = "Number of seconds established client connection should be cached if not used (0 means no cache, connection will be disposed; a shuffle-heavy query can then exhaust the host's ephemeral ports)."
     )]
     pub client_ttl: u64,
 }
@@ -256,4 +257,20 @@ mod tests {
         assert!(parse_memory_pool_size("banana").is_err());
         assert!(parse_memory_pool_size("").is_err());
     }
+
+    /// The client pool is what keeps shuffle fetches from opening a fresh TCP
+    /// connection each time; with it off, one shuffle-heavy query can exhaust
+    /// the host's ephemeral ports and take the executor down with it. Pin the
+    /// default on so it cannot regress to the disabled state unnoticed.
+    #[cfg(feature = "build-binary")]
+    #[test]
+    fn client_pool_is_enabled_by_default() {
+        use clap::Parser;
+        let opt = super::Config::parse_from(["ballista-executor"]);
+        assert!(
+            opt.client_ttl > 0,
+            "client_ttl must default to a pooling value, got {}",
+            opt.client_ttl
+        );
+    }
 }
diff --git a/ballista/executor/src/executor_process.rs b/ballista/executor/src/executor_process.rs
index 360a89491..80b4874b3 100644
--- a/ballista/executor/src/executor_process.rs
+++ b/ballista/executor/src/executor_process.rs
@@ -187,7 +187,10 @@ pub struct ExecutorProcessConfig {
     pub override_arrow_flight_service: Option<Arc<ArrowFlightServerProvider>>,
     /// Override function for customizing gRPC client endpoints before they are used
     pub override_create_grpc_client_endpoint: Option<EndpointOverrideFn>,
-    /// Number of seconds established client connection should be cached (0 means no cache)
+    /// Number of seconds an established client connection is cached while idle.
+    /// `0` disables the pool, so every shuffle fetch opens and drops its own
+    /// connection and a shuffle-heavy query can exhaust the host's ephemeral
+    /// ports.
     pub client_ttl: u64,
 }
 
@@ -238,7 +241,7 @@ impl Default for ExecutorProcessConfig {
             override_physical_codec: None,
             override_arrow_flight_service: None,
             override_create_grpc_client_endpoint: None,
-            client_ttl: 0,
+            client_ttl: 30,
         }
     }
 }


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org