sunchao commented on code in PR #3692:
URL: https://github.com/apache/celeborn/pull/3692#discussion_r3295083904
##########
client-spark/spark-3/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleReader.scala:
##########
@@ -548,6 +556,35 @@ class CelebornShuffleReader[K, C](
object CelebornShuffleReader {
var streamCreatorPool: ThreadPoolExecutor = null
+
+ @VisibleForTesting
+ private[celeborn] def createClientsInParallel(
+ locationsByHostPort: Seq[(String, Seq[PartitionLocation])],
+ streamCreatorPool: ThreadPoolExecutor,
+ createClient: PartitionLocation => TransportClient,
+ onClientCreateFailure: (String, PartitionLocation, Exception) => Unit)
+ : Map[String, TransportClient] = {
+ val clientsByHostPort = JavaUtils.newConcurrentHashMap[String,
TransportClient]()
+ val futures = locationsByHostPort.map { case (hostPort, locations) =>
+ streamCreatorPool.submit(new Runnable {
+ override def run(): Unit = {
+ locations.find { location =>
+ try {
+ clientsByHostPort.put(hostPort, createClient(location))
+ true
+ } catch {
+ case ex: Exception =>
+ onClientCreateFailure(hostPort, location, ex)
+ false
+ }
+ }
+ }
+ })
+ }
+ futures.foreach(_.get())
Review Comment:
Fixed in 17d5afaa4. The wait now restores the task thread interrupt status
and cancels unfinished client-creation futures when it exits exceptionally. The
client-creation worker also propagates InterruptedException instead of treating
cancellation as a retryable location failure. Added a regression test that
interrupts the waiting caller and verifies blocked creation is cancelled
without retrying.
--
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]
