sunchao commented on code in PR #3746:
URL: https://github.com/apache/celeborn/pull/3746#discussion_r3606904092
##########
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:
Addressed in `03a61748b`. The sequential path now owns one
`attemptedClientHostPorts` set for the full reader setup and records each
`hostAndFetchPort` before client creation. If that attempt fails, later
locations for the same endpoint short-circuit, while a different endpoint
remains eligible. I added the `attempt sequential batch open stream client
creation once per worker endpoint` regression; the Spark 3.5 reader suite
passes 9/9.
##########
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:
I traced the concrete producers of this exception shape.
`TransportClient.sendRpcSync()` wraps `Future.get()` failures; an
`InterruptedException` there means the waiting caller thread was interrupted,
and the wait clears its flag. Netty connect/TLS `Future.await()` has the same
semantics: ordinary connect or handshake failures complete the future with
their own non-interruption cause, and remote RPC failures are reconstructed as
plain `IOException`. The only cross-thread reader case is the DFS fetch worker,
whose interrupt comes from `close()` calling `shutdownNow()`. Therefore a
nested `InterruptedException` on these paths is a hidden cancellation or
closure signal, not a recoverable worker failure; retrying it would swallow
that signal and incorrectly exclude or fail over from a healthy worker. I kept
the classification unchanged. Existing tests cover ordinary `IOException` peer
failover, while the wrapped-interruption tests verify no exclusion or failover;
the focused input-str
eam suite passes 6/6.
--
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]
