Copilot commented on code in PR #3692:
URL: https://github.com/apache/celeborn/pull/3692#discussion_r3278307227
##########
client-spark/spark-3/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleReader.scala:
##########
@@ -272,10 +250,40 @@ class CelebornShuffleReader[K, C](
locations = filterLocations.asJava
}
partitionId2PartitionLocations.put(partitionId, locations)
- makeOpenStreamList(locations)
+ groupOpenStreamLocations(locations)
}
}
+ val clientsByHostPort = CelebornShuffleReader.createClientsInParallel(
+ locationsByHostPort.map { case (hostPort, locations) =>
+ (hostPort, locations.get(0))
+ }.toSeq,
+ streamCreatorPool,
+ location =>
+ shuffleClient.getDataClientFactory().createClient(
+ location.getHost,
+ location.getFetchPort),
+ (hostPort, location, ex) => {
+ shuffleClient.excludeFailedFetchLocation(hostPort, ex)
+ logWarning(
+ s"Failed to create client for $shuffleKey-${location.getId} from
host: ${hostPort}. " +
+ s"Shuffle reader will try its replica if exists.")
+ })
Review Comment:
This refactor changes the retry behavior for data-client creation per
worker. Previously, if `createClient` failed for a `hostPort`, later locations
on the same `hostPort` would re-attempt client creation because
`workerRequestMap` remained empty; now `createClient` is attempted only once
per worker (using `locations.get(0)`), and the worker is skipped for batch-open
even if a subsequent attempt might succeed. If the old implicit retry was
intentional, consider adding an explicit bounded retry (with backoff) around
`createClient`, or documenting that batch-open will now do at most one
client-creation attempt per worker.
--
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]
