sunchao commented on code in PR #3746:
URL: https://github.com/apache/celeborn/pull/3746#discussion_r3606905363
##########
client-spark/spark-3/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleReader.scala:
##########
@@ -600,6 +602,26 @@ class CelebornShuffleReader[K, C](
object CelebornShuffleReader {
var streamCreatorPool: ThreadPoolExecutor = null
+ @VisibleForTesting
+ private[celeborn] def tryCreateClient(
+ location: PartitionLocation,
+ createClient: PartitionLocation => TransportClient,
+ onClientCreateFailure: Exception => Unit): Option[TransportClient] = {
+ if (Thread.currentThread().isInterrupted) {
+ throw new InterruptedException("Client creation interrupted")
+ }
+ try {
+ Some(createClient(location))
+ } catch {
+ case ex: InterruptedException =>
Review Comment:
Addressed in `03a61748b`. I extracted the cause-chain lookup into
`ExceptionUtils.findInterruptedException()` and reused it from
`TransportClientFactory`, `CelebornInputStream`, and `CelebornShuffleReader`.
`tryCreateClient` now unwraps a nested interruption, restores the worker thread
interrupt flag, and exits without invoking the ordinary failure callback.
`createClientsInParallel` also handles `ExecutionException` from
`Future.get()`: it unwraps the nested interruption, restores the caller flag,
rethrows it, and the existing `finally` cancels outstanding futures. New
regressions cover both wrapped client creation and the parallel future
boundary; the Spark 3.5 reader suite passes 9/9.
--
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]
