andygrove opened a new issue, #2068:
URL: https://github.com/apache/datafusion-ballista/issues/2068
## Describe the bug
The executor's Ballista client pool is **disabled by default**
(`--client-ttl`
defaults to `0`, "no cache, connection will be disposed"). With no pool, the
shuffle-read path opens and discards a **new TCP connection per fetch**. On
any
multi-executor cluster a single query exhausts the machine's entire ephemeral
port range, and the query — plus every query after it — fails with
`AddrNotAvailable`.
With one executor this is invisible: shuffle reads are served from local
files,
so there are no remote fetches. It appears as soon as a second executor
exists,
which is every real deployment.
## To Reproduce
SF1 TPC-DS parquet, 1 scheduler + **2** executors (defaults, i.e. no
`--client-ttl`):
```
ballista-executor -c 4 -p 50052 --bind-grpc-port 50062 --work-dir /tmp/wd1
ballista-executor -c 4 -p 50053 --bind-grpc-port 50063 --work-dir /tmp/wd2
tpcds --host 127.0.0.1 --port 50050 --path <data> --partitions 16 --verify \
-c datafusion.optimizer.prefer_hash_join=false
```
Result — the run collapses at q14 and never recovers:
```
Query 14 FAILED to run in 10.463s: ... FetchFailed(..., "Error connecting to
Ballista scheduler or executor at http://127.0.0.1:50052: ...
ConnectError("tcp
connect error", 127.0.0.1:50052, Os { code: 49, kind: AddrNotAvailable,
message: "Can't assign requested address" })")
Query 15 FAILED ...
Query 16 FAILED ... (every subsequent query cannot connect)
```
Final tally: **11 OK / 77 FAILED**. The scheduler and both executors stay
alive
the whole time — nothing crashes, the host is simply out of ephemeral ports.
### Measurements
Sockets sampled during a single q14 run (macOS ephemeral range 49152–65535 =
**16384 ports**):
| | default (`--client-ttl 0`) | `--client-ttl 30` |
|---|---:|---:|
| peak TCP sockets | 16563 | 1618 |
| peak TIME_WAIT | 16343 | 11 |
| q14 | FAILED (`AddrNotAvailable`) | verified OK |
| full TPC-DS gate (2 executors) | 11 OK / 77 FAILED | **88 OK / 0 FAILED** |
One query consumes the entire ephemeral port range. TIME_WAIT dominating the
count is the tell: connections are opened and immediately dropped rather than
reused.
The TIME_WAIT sockets are overwhelmingly against the **executor flight
ports**,
which pins it to the shuffle-read path rather than scheduler traffic:
```
TIME_WAIT by destination port:
8315 50052 <- executor 1 flight
8023 50053 <- executor 2 flight
1 50050 <- scheduler grpc
```
## Cause
`ballista/executor/src/executor_process.rs`:
```rust
if opt.client_ttl > 0 {
let client_pool =
Arc::new(DefaultBallistaClientPool::with_eviction_thread(
Duration::from_secs(opt.client_ttl),
));
Arc::new(DefaultExecutionEngine::with_client_pool(client_pool))
} else {
Arc::new(DefaultExecutionEngine::new()) // no pool
}
```
and `ballista/executor/src/config.rs`:
```rust
#[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).")]
pub client_ttl: u64,
```
Without a pool, `DefaultExecutionEngine::new()` leaves `client_pool: None`,
so
`ShuffleReaderExec` never gets one attached and each remote fetch builds a
fresh
client. This is at odds with the shuffle reader's own design intent, which
assumes connection reuse:
> Because total in-flight bytes stay <= the sized h2 window, the governor —
not
> the 64 KB transport window — is the binding backpressure, **which is what
makes
> multiplexing over few connections safe.**
The pooling machinery is all present and works; it is simply off unless the
operator knows to pass `--client-ttl`.
## Expected behavior
Connection pooling should be **on by default**. The default should not be a
configuration that cannot survive a two-executor cluster. Opting out (via
`--client-ttl 0`) can remain available.
## Additional context
- Reproduced on `upstream/main` (e843c8e8), SF1, 1 scheduler + 2 executors
(`-c 4` each), default static planner, `prefer_hash_join=false`.
- Workaround today: start executors with `--client-ttl <seconds>` (e.g.
`30`).
- Not scale-dependent — SF1 is enough. It is driven by fetch count
(stages x partitions), so larger scale factors and partition counts make it
worse.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
