andygrove opened a new pull request, #1946:
URL: https://github.com/apache/datafusion-ballista/pull/1946
# Which issue does this PR close?
Closes #1943.
# Rationale for this change
At high `target_partitions`, distributed queries intermittently failed
during shuffle
reads with `FetchFailed` — either `EADDRNOTAVAIL` ("Cannot assign requested
address",
with client caching off) or an `h2` `BrokenPipe` ("error reading a body from
connection", with caching on).
Instrumenting the executors with `h2=debug` and reproducing on a 2-executor
× 8-core
cluster (TPC-H SF100 from S3) showed the cause: the shuffle client **opened
and tore
down thousands of connections per query** — ~3,500 client-initiated `GOAWAY`
frames in a
single run, all graceful `NO_ERROR`, with zero `RST_STREAM` (so not h2
rapid-reset). The
failure is an intermittent race: the client's own connection **close racing
an in-flight
body read** produces the broken pipe, and the same churn exhausts ephemeral
ports when
caching is off.
The churn comes from the pool's model: `acquire` checked out an **exclusive
connection
per fetch** and returned it on drop, so a shuffle-heavy stage cycled through
thousands of
short-lived connections.
# What changes are included in this PR?
Replace the exclusive-connection-per-fetch pool with a small, bounded set of
**reused, multiplexed connections per endpoint**.
`DefaultBallistaClientPool` now keeps up
to `CONNECTIONS_PER_ENDPOINT` (8) shared `BallistaClient`s per `(host, port,
config)` and
`acquire` hands out cheap **clones round-robin**. Each underlying
`tonic::Channel`
multiplexes concurrent requests over its connection and reconnects
transparently, so:
- per-query connections collapse from thousands to a handful, removing the
churn-driven
broken-pipe / port-exhaustion races, and
- load is spread across enough connections that no single one stalls under
high
shuffle-fetch concurrency (a single shared connection deadlocks — hence a
small pool
rather than one).
Dropping a handed-out clone is a no-op (the shared clients stay cached); the
background
task still evicts endpoints unused within `idle_timeout`. The change is
contained to
`ballista/executor/src/client_pool.rs`; there are no public API changes.
# Are there any user-facing changes?
No API changes. Fixes intermittent shuffle `FetchFailed` at high
`target_partitions`.
<!-- How tested (verification in progress on a live cluster; will finalize
before marking
ready): 2 executors × 8 cores, TPC-H SF100 from S3/MinIO,
target_partitions=128/256 — the
partition counts that fail ~2 of 3 runs on `main`. With this change the Q3
shuffle stage
(where the failures cluster) completes normally and per-query times are on
par with (a
touch faster than) `main`. An h2=debug run confirms client-initiated GOAWAYs
drop from
~3,500/run to a handful. -->
--
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]
