SteNicholas commented on code in PR #3746:
URL: https://github.com/apache/celeborn/pull/3746#discussion_r3541918004
##########
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:
Consistency: this catches only a **bare** `InterruptedException`. A wrapped
one (`IOException(InterruptedException)`) falls through to `case ex: Exception`
below and is reported as a worker failure via `excludeFailedFetchLocation` —
the exact misclassification that
`TransportClientFactory.findInterruptedException` and
`CelebornInputStream.isInterruption` (both added in this PR) walk the cause
chain to prevent. It's masked today only because the real `createClient` path
goes through `retryCreateClient`, which already unwraps to a bare
`InterruptedException` — a fragile coupling.
Note the three interrupt-detection helpers this PR introduces already
disagree: `findInterruptedException` ignores the thread flag, `isInterruption`
checks it first, and `tryCreateClient` checks neither the chain nor the flag
mid-`createClient`. Consider extracting one shared helper (e.g. in `common`
`ExceptionUtils`) and using it in all three sites so they can't drift.
Related latent gap in `createClientsInParallel` below: a pool task that
rethrows `InterruptedException` surfaces at `futures.foreach(_.get())` as a
`java.util.concurrent.ExecutionException`, which the `catch { case ex:
InterruptedException }` there won't match — so the re-interrupt is skipped and
a raw `ExecutionException` escapes `read()`. It's hard to hit today (the
waiting thread's own `.get()` usually throws `InterruptedException` first, and
pool-thread flags are cleared between tasks), but unwrapping
`ExecutionException(InterruptedException)` there would close it.
##########
client-spark/spark-3/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleReader.scala:
##########
@@ -215,21 +215,23 @@ class CelebornShuffleReader[K, C](
partCnt += 1
val hostPort = location.hostAndFetchPort
if (!workerRequestMap.containsKey(hostPort)) {
- try {
- val client = shuffleClient.getDataClientFactory().createClient(
- location.getHost,
- location.getFetchPort)
+ CelebornShuffleReader.tryCreateClient(
Review Comment:
The retry-bounding fix (and the `headOption` change in
`createClientsInParallel`) only covers the **parallel** client-creation path.
The sequential `makeOpenStreamList` still attempts creation per
partition-location: when `tryCreateClient` returns `None`,
`onClientCreateFailure` only excludes + warns and never populates
`workerRequestMap`, so the next location sharing the same `hostPort` re-enters
`if (!workerRequestMap.containsKey(hostPort))` (line 217) and calls
`createClient` against the same dead endpoint again — the `N × maxRetries`
blow-up this PR sets out to remove.
So with
`celeborn.client.spark.batch.openStream.parallelClientCreation.enabled=false`,
the stated goal ("Bound … client creation retries") isn't achieved. This isn't
a regression (it matches the old behavior), but since the parallel path is now
bounded it's an inconsistency worth either closing (e.g. record the failed
`hostPort` so subsequent same-host locations short-circuit) or calling out as a
known limitation.
##########
client/src/main/java/org/apache/celeborn/client/read/CelebornInputStream.java:
##########
@@ -438,6 +440,23 @@ private boolean isExcluded(PartitionLocation location) {
}
}
+ private static boolean isInterruption(Throwable throwable) {
+ if (Thread.currentThread().isInterrupted()) {
Review Comment:
`isInterruption` returns true when either (a) the current thread's interrupt
flag is set, or (b) **any** exception in `throwable`'s cause chain is an
`InterruptedException`. Both are broader than "this failure is a task
cancellation."
Concern: a genuine, retryable fetch failure whose cause chain happens to
wrap an `InterruptedException` (e.g. a transient Netty blocking-await
interruption that got wrapped, without the task actually being cancelled) is
now treated as an interruption in `getNextChunk` / `createReaderWithRetry` —
the code closes the reader and throws **without** attempting the peer/replica,
whereas the pre-PR path would have failed over. It also flips the live thread's
interrupt flag via `interruptedIOException`. In the common cancellation case
that's exactly right; the risk is the false-positive where an
`InterruptedException` in the chain doesn't represent a real cancellation,
which would strand a recoverable read on the primary and turn it into a task
failure. Worth confirming that scenario can't arise from a non-cancellation
source.
--
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]
