Shuffle fetch exhausts ephemeral ports at high target_partitions (client connection caching off by default)

[andygrove]

Describe the bug

With connection caching disabled — which is the default (--client-ttl 0) — Ballista's
shuffle reader opens and then disposes a new TCP connection per remote partition fetch.
As target_partitions grows, the shuffle fan-out produces enough short-lived connections to
exhaust the executor's ephemeral ports, and queries fail with EADDRNOTAVAIL:

Job failed due to stage 3 failed: Task failed due to runtime execution error:
FetchFailed(..., "Error connecting to Ballista scheduler or executor at http://10.42.0.20:50051:
tonic::transport::Error(Transport, ConnectError(ConnectError(\"tcp connect error\",
10.42.0.20:50051, Os { code: 99, kind: AddrNotAvailable,
message: \"Cannot assign requested address\" })))")

To Reproduce

• Cluster: 2 executors × 8 cores (16 task slots), executors started with defaults
  (i.e. --client-ttl 0).
• TPC-H SF100, Parquet, read from S3 (MinIO).
• Run the tpch benchmark against the cluster at increasing --partitions
  (target_partitions):

target_partitions	result
16 / 32 / 64	complete
128	fails at stage 3 with the EADDRNOTAVAIL FetchFailed above
256	fails at stage 18, same error

Root cause

In ballista/core/src/execution_plans/shuffle_reader.rs, fetch_partition_remote has two
paths:

• With a client_pool (Some) → reuses pooled connections.

• Without (None) → creates a fresh connection per fetch. There is already an explicit
  TODO here acknowledging the problem:

  // TODO for shuffle client connections, we should avoid creating new connections again and again.
  // And we should also avoid to keep alive too many connections for long time.

The pool (DefaultBallistaClientPool, added in #1578) is only wired into the execution
engine when the executor's --client-ttl > 0 (executor_process.rs), and client_ttl
defaults to 0 — so out of the box there is no pool and every shuffle fetch opens and
disposes its own connection. High partition counts then exhaust local ports.

Expected behavior

Shuffle fetches should not exhaust local ephemeral ports as target_partitions scales.
Connection reuse (already implemented) should apply by default, or shuffle connections should
otherwise be bounded/reused independent of the caching TTL.

Possible mitigations

• Workaround: start executors with --client-ttl > 0 to enable the existing
  DefaultBallistaClientPool. (Inferred from the code path; I have not yet re-run to confirm
  it resolves the failure in this environment.)
• Fix ideas: default --client-ttl to a non-zero value, and/or always pool/reuse shuffle
  connections regardless of the caching TTL, so the default configuration does not fail at
  higher partition counts.

Additional context

Found while sweeping target_partitions for a TPC-H SF100 comparison; the sweet spot on this
2×8 cluster was target_partitions=32, and 128/256 hit this failure. Ballista is main
(DataFusion 54); executors 8 cores each.

[andygrove]

Update: verified the --client-ttl workaround — it does not resolve the failure; it changes the failure mode.

I re-ran with executors started with --client-ttl=60 (connection caching enabled), same
cluster (2 executors × 8 cores), same TPC-H SF100, target_partitions=128.

Result:

• The EADDRNOTAVAIL / "Cannot assign requested address" port-exhaustion error is gone —
  so caching does prevent the new-connection-per-fetch storm.

• But the query still fails at the same point (~stage 3, 2/22 queries), now with a
  different error — a reused pooled connection breaking mid-fetch:

  h2 protocol error: error reading a body from connection
  hyper::Error(Body, Io(Custom { kind: BrokenPipe, error: "stream closed because of a broken pipe" }))
  

Both executors stayed healthy throughout (0 restarts, no OOM / termination), so this is not
an executor crash — the pooled connection itself is breaking under the high-partition shuffle
load, and the fetch is not retried on a fresh connection.

So high-target_partitions shuffle is unstable in both modes:

shuffle client caching	failure at target_partitions=128
off (default, --client-ttl 0)	ephemeral-port exhaustion (EADDRNOTAVAIL)
on (--client-ttl 60)	h2 broken-pipe on connection reuse

(target_partitions 16 / 32 / 64 still complete fine in both modes.)

Refined take / possible fixes:

• Enabling the client pool by default is necessary but not sufficient on its own.
• In fetch_partition_remote (shuffle_reader.rs), a failed pooled fetch calls
  pooled.discard() but does not retry on a fresh connection — so a single stale/broken
  pooled connection fails the whole query. Retrying the fetch with a new connection when the
  pooled one is broken would likely make this robust.
• It's also worth understanding why the connection is closed under load as partition count
  scales (h2 flow-control / max-concurrent-streams, server-side idle close, etc.).

Environment: Ballista main (DataFusion 54), executors 8 cores each.

[andygrove]

Update: instrumented the repro (executor-side h2=debug) and root-caused it. The failure is intermittent, and it's client-driven connection churn — not a server-side reset.

Re-ran target_partitions=128 with RUST_LOG=info,h2=debug on both executors (shuffle
fetches are executor→executor, so both sides matter). Notably, this run of 128 passed
(22/22) — the failure is a timing race, which is why the same config fails some runs
and not others.

Frame inventory for one run:

	count
GOAWAY sent by client	3,292
GOAWAY sent by server	348
RST_STREAM (sent/recv)	0
SETTINGS frames	18,660
client-side "Remote shuffle read fail, retry" (all recovered)	93

So under high target_partitions the client opens and tears down thousands of connections
per run (the exclusive one-connection-per-fetch model), and RST_STREAM: 0 rules out h2
rapid-reset. All the GOAWAYs are NO_ERROR (graceful). This churn is what produces both
symptoms: with caching off it exhausts ephemeral ports (EADDRNOTAVAIL); with caching on it
occasionally catches a fetch mid-stream (BrokenPipe).

Two coupled defects in the shuffle client:

1. Connection returned to the pool before its stream is drained. In
   fetch_partition_remote (shuffle_reader.rs), the PooledClient guard is a local that
   drops the instant the function returns the stream — and PooledClient::drop
   (client_pool.rs) returns self.client.clone() to the pool. So a connection is marked
   idle/reusable/evictable while its response body is still streaming, opening a
   teardown-vs-read race (and connection aliasing via the clone).

2. Retry doesn't cover the body phase. execute_do_get (client.rs) retries only the
   do_get request and the first (schema) message, and retries on the same connection. A
   connection error while reading data batches (after the schema) is outside the retry loop
   entirely → hard FetchFailed. (The recoverable case is what produced the 93 successful
   retries above; the mid-body case is the intermittent failure.)

Fix direction (no buffering):

• Tie the connection's lifetime to the stream: hold the PooledClient until the returned
  stream is fully consumed, so the pool can't reuse/evict/close a connection mid-fetch.
• Reduce churn by reusing a few long-lived, multiplexed connections per peer (h2 multiplexes
  streams and auto-reconnects) instead of one connection per fetch — this removes the
  port-exhaustion mode outright.

I'm going to try a fix along these lines and verify it against the live 128/256 repro.

[andygrove]

Investigation summary: root cause + six attempts + the design tension

I spent a good while reproducing and instrumenting this on a 2-executor × 8-core
cluster (TPC-H SF100 from S3/MinIO, target_partitions 128/256). Posting the full
findings so the fix can be decided as a design question rather than guessed at — I
tried six approaches and each hit a different corner of what looks like a fundamental
tension.

Confirmed root cause: shuffle-connection churn

With executor-side RUST_LOG=info,h2=debug, a single failing target_partitions=128
run shows the shuffle client opening and tearing down ~3,500 connections (client-
initiated GOAWAY frames, all NO_ERROR; 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 h2 … BrokenPipe ("error reading a
body from connection"); with client caching off the same churn exhausts ephemeral
ports (EADDRNOTAVAIL). The churn comes from the pool checking out an exclusive
connection per fetch and returning it on drop.

The tension (why it's not a one-line fix)

A shuffle read opens many concurrent partition streams. Each open stream needs
either its own connection or to share one — and every option hits a wall:

Connection model	Failure
One connection per fetch (today)	churn → broken-pipe race / ephemeral-port exhaustion
Hold-connection-for-stream + keepalive	still broken pipe — the churn is unchanged
Multiplex, 1 connection/peer	hang — the shared 64 KB h2 connection-level flow-control window starves under many concurrent fetch streams; executors go fully idle
Multiplex, small pool (N/peer)	same hang, just on heavier queries (raises the threshold, doesn't remove it)
Bounded exclusive pool (semaphore)	hang — permit starvation: a stage opens more concurrent partition streams than the pool cap; the streams holding connections can't release (they're waiting on the consumer, which is waiting on a stream that can't get a connection)
Unbounded exclusive pool	back to churn / port exhaustion

So: exclusive-per-open-stream → either unbounded (churn/ports) or bounded (starvation
deadlock); shared/multiplexed → h2 flow-control deadlock. Each of the six attempts
landed on one of these.

What each attempt showed (evidence)

1. Hold the pooled connection for the response-stream lifetime (so it isn't
   reused/evicted mid-stream). Reduced it but did not fix it — still broke, because
   the churn itself remained.
2. HTTP/2 keepalive on shuffle connections (they were created with a None config,
   i.e. no keepalive at all). Correct on its own merits, but didn't fix the broken
   pipe — the connection dies while actively streaming, not while idle.
3. Multiplex over one connection per peer. Removed the churn entirely, but hung:
   the executors sit at ~0 CPU on the first shuffle-heavy query. The tiny default
   connection-level flow-control window is the suspect.
4. Multiplex over a small round-robin pool (N=8 per peer). Cleared the moderate
   shuffle stage but hung on a heavier join (more concurrent streams/connection).
5. Bounded exclusive pool (semaphore, ≤64/endpoint, reuse, one fetch per
   connection). Also hung, and earlier — permit starvation as above.

(Per-query timings, where runs completed, were on par with main — so none of this is
a throughput problem; it's a liveness one.)

Options that remain (haven't tried / need design)

1. Multiplex + enlarge the h2 flow-control windows. Bump
   initial_connection_window_size / initial_stream_window_size from the 64 KB
   default to several MB on the shuffle client (and matching server settings). This is
   the one principled lever I did not pull, and it directly targets the multiplex
   deadlock cause. If it works, multiplexing becomes viable and removes the churn.
2. Fetch each partition to completion before releasing the connection (buffer to
   memory or spill to disk). Decouples connection-hold-time from consume-time, so a
   bounded exclusive pool no longer starves. Costs a per-partition buffer.
3. A dedicated shuffle transport not built on per-partition Flight do_get.

Environment / repro

• Ballista main (DataFusion 54), executors 8 cores each, --client-ttl 60.
• TPC-H SF100 from S3; target_partitions 128 fails ~2 of 3 runs on main; 256 fails
  more often. The shuffle-heavy queries (Q7/Q9/Q17/Q21) are where failures cluster.

Happy to run any of the above against this cluster if a direction is chosen. My instinct
is option 1 (flow-control windows) is the smallest principled next step, but this is a
core-transport decision so I wanted the full picture on the record first.

[andygrove]

Reference: how Spark's shuffle fetch avoids all three corners

For whoever picks up the fix — Spark solves this exact problem, and its design points
at what Ballista is actually missing. Verified against the Spark 3.5.8 source
(org.apache.spark.storage.ShuffleBlockFetcherIterator,
org.apache.spark.network.util.TransportConf).

Transport. Not gRPC/HTTP-2 — a custom length-delimited Netty TCP protocol
(NettyBlockTransferService / TransportClientFactory). There is no protocol-level
64 KB connection-level flow-control window to starve, which is what deadlocked the
multiplexing attempts here.

Connections. spark.shuffle.io.numConnectionsPerPeer defaults to 1
(TransportConf.numConnectionsPerPeer() → 1). All block fetches to a peer are
multiplexed over that one long-lived channel — i.e. Spark runs the exact model that
hung Ballista at 1 connection/peer, and it is fine.

Why it doesn't deadlock — a reduce-side in-flight governor. This is the piece
Ballista lacks. ShuffleBlockFetcherIterator only issues a fetch when
(isRemoteBlockFetchable, SBFI.scala:1215):

bytesInFlight == 0 ||
  (reqsInFlight + 1 <= maxReqsInFlight &&
   bytesInFlight + fetchReqQueue.front.size <= maxBytesInFlight)

with, by default:

• spark.reducer.maxSizeInFlight = 48m (maxBytesInFlight),
• targetRemoteRequestSize = max(maxBytesInFlight / 5, 1) — requests are pre-split so up
  to ~5 peers fetch in parallel (SBFI.scala:112),
• spark.reducer.maxReqsInFlight (default Int.MaxValue) and
  spark.reducer.maxBlocksInFlightPerAddress — caps on outstanding requests / blocks
  per address.

So the amount of concurrent in-flight data on any connection is bounded at the
application layer. The transport is never flooded, so one multiplexed connection is
enough and never starves. Ballista opens a do_get stream per partition with no global
in-flight-bytes/requests governor, so it must either flood one h2 connection (→ window
deadlock) or open a connection per fetch (→ churn).

Resilience. Shuffle fetches are discrete, idempotent blocks, and Spark retries
them: spark.shuffle.io.maxRetries = 3 (TransportConf.maxIORetries()). A mid-fetch
connection break just refetches the block. Large blocks stream to disk
(spark.maxRemoteBlockSizeFetchToMem) instead of pinning memory/the connection. Contrast
Ballista, where execute_do_get's retry covers only the request + schema message, so a
broken pipe during the body is fatal, and a partition is an open-ended stream rather than
a bounded, retriable unit.

Mapping to the three-corner tension above

Ballista failure	How Spark avoids it
Multiplex → h2 64 KB window deadlock	Netty transport, no shared protocol window; plus reduce-side in-flight cap so a connection is never overloaded
Exclusive-per-stream → churn / pool starvation	1 reused connection/peer, many blocks multiplexed, throttled by in-flight bytes/requests
Broken pipe mid-stream fatal	Blocks are idempotent + retried (maxRetries=3); big blocks spill to disk

Design takeaway

The decisive thing Spark has that Ballista doesn't is a reduce-side in-flight governor
(maxBytesInFlight / maxReqsInFlight / maxBlocksInFlightPerAddress). That governor is
what makes multiplexing over very few connections safe. So the faithful fix is not a
pool-shape tweak or just larger h2 windows in isolation, but roughly:

1. a small, reused set of connections per peer (even 1–N), plus
2. a bounded in-flight-bytes / in-flight-requests governor on the shuffle reader, plus
3. retriable, bounded block fetches (retry the body phase; optionally spill large
   partitions to disk).

That combination dissolves all three corners at once, and it's a direct port of a model
that's been load-bearing in Spark for years.