RexXiong commented on PR #3692:
URL: https://github.com/apache/celeborn/pull/3692#issuecomment-4534592324
**Overall**: Clean refactoring that parallelizes data-client creation across
workers. The separation into group → parallel create → build request is
well-structured, and the test coverage is thorough (parallel execution, failure
isolation, retry, interruption).
**1. Redundant retries with identical connection parameters**
`createClientsInParallel` iterates through all locations for a given
`hostPort` until `createClient` succeeds. However, all locations at the same
`hostPort` share identical `host` and `fetchPort`, so each retry is an
identical connection attempt:
```scala
while (!clientCreated && locationsIterator.hasNext) {
val location = locationsIterator.next()
try {
clientsByHostPort.put(hostPort, createClient(location)) // same
host:port every time
clientCreated = true
} catch { ... }
}
```
For a reducer reading 1000 partitions from a failing worker, this means up
to 1000 identical connection attempts (each paying the full connection
timeout). Since `futures.foreach(_.get())` blocks until ALL futures complete,
one slow-failing worker would dominate the entire parallel creation phase —
partially defeating the purpose of parallelization.
Consider either:
- Taking only the first location per hostPort for client creation (the
original behavior was also 1 attempt)
- Adding a configurable max retry count (e.g., 2-3 attempts per hostPort)
**2. +1 to SteNicholas's suggestion for a config switch**
A `celeborn.client.spark.batch.openStream.parallelClientCreation.enabled`
(or similar) flag would be prudent for a new parallelization path. This allows
users to fall back to serial behavior if unexpected issues arise in production.
Could default to `true`.
**3. Minor: `onClientCreateFailure` invoked per failed location**
In the original code, `excludeFailedFetchLocation` was called at most once
per failing worker. Now it can be called N times (once per location at that
hostPort). While `excludeFailedFetchLocation` is idempotent
(ConcurrentHashMap), the associated `logWarning` will emit N log lines for the
same worker, which could be noisy for large reducers. This ties back to point 1
— bounding retries would also bound log spam.
*Reviewed with Claude Code*
--
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]
