andygrove opened a new pull request, #1945:
URL: https://github.com/apache/datafusion-ballista/pull/1945
# Which issue does this PR close?
Closes #1943.
# Rationale for this change
At high `target_partitions`, distributed TPC-H queries intermittently failed
with
`FetchFailed` during shuffle reads. Two symptoms, one root:
- with client connection caching **off** (default, `--client-ttl 0`):
`EADDRNOTAVAIL` / "Cannot assign requested address" — ephemeral-port
exhaustion;
- with caching **on** (`--client-ttl > 0`): an `h2` `BrokenPipe` — "error
reading a
body from connection", i.e. the connection broke **mid-stream while
reading data
batches**.
Instrumented `h2=debug` runs (2 executors × 8 cores, TPC-H SF100 from S3)
showed the
shuffle client opening and tearing down **thousands of connections per
query**
(client-initiated `GOAWAY`s; zero `RST_STREAM`, so not h2 rapid-reset), and
the failure
being an intermittent **race**: the same `target_partitions=128` run failed
on some
attempts and passed on others.
Root cause for the broken-pipe case: `fetch_partition_remote` returned the
pooled
connection to the pool **as soon as `fetch_partition` returned the response
stream**.
The `PooledClient` guard was a local that dropped at that point, returning
(a clone of)
the connection to the idle pool while the response **body was still
streaming**. That
marks the connection idle/reusable/evictable mid-fetch, racing connection
teardown/reuse against the in-flight body read. And because
`execute_do_get`'s retry
only covers the request + first (schema) message — not the body phase — the
break
surfaces as a hard failure rather than a retry.
# What changes are included in this PR?
Keep the `PooledClient` checked out for the **lifetime of the response
stream**.
`fetch_partition_remote` now wraps the returned stream in a small
`PooledPartitionStream` that owns the `PooledClient`, so the connection is
only
returned to the pool — or `discard()`ed, if the stream yields a transport
error — once
the stream has been fully consumed. This prevents a connection from being
reused,
evicted, or closed while its body is still being read.
Scoped, ~60-line change to
`ballista/core/src/execution_plans/shuffle_reader.rs`; no
public API changes.
# Are there any user-facing changes?
No API changes. Fixes intermittent shuffle `FetchFailed` at high
`target_partitions`.
<!-- Verification (in progress, this is a draft): reproduced on a 2-executor
× 8-core
cluster reading TPC-H SF100 from S3/MinIO. Without this change,
target_partitions=128
fails roughly 2 of 3 runs; with it, repeated 128/256 runs are being
confirmed to pass.
Will post the final tally and mark ready for review. -->
--
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]
