andygrove opened a new issue, #1943: URL: https://github.com/apache/datafusion-ballista/issues/1943
## 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: ```rust // 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. -- 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]
