sunchao commented on code in PR #3746:
URL: https://github.com/apache/celeborn/pull/3746#discussion_r3540633886
##########
client-spark/spark-3/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleReader.scala:
##########
@@ -600,6 +600,9 @@ class CelebornShuffleReader[K, C](
object CelebornShuffleReader {
var streamCreatorPool: ThreadPoolExecutor = null
+ // TransportClientFactory already retries each attempt; allow one extra
pooled-client attempt.
+ private val MAX_CLIENT_CREATION_ATTEMPTS_PER_HOST = 2
Review Comment:
Thanks. I removed the outer per-host retry entirely rather than adding
another setting. `celeborn.data.io.maxRetries` remains the single retry control
inside `TransportClientFactory`; adding a second outer setting would multiply
that complete retry budget.
##########
client-spark/spark-3/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleReader.scala:
##########
@@ -600,6 +600,9 @@ class CelebornShuffleReader[K, C](
object CelebornShuffleReader {
var streamCreatorPool: ThreadPoolExecutor = null
+ // TransportClientFactory already retries each attempt; allow one extra
pooled-client attempt.
+ private val MAX_CLIENT_CREATION_ATTEMPTS_PER_HOST = 2
Review Comment:
Agreed. Every location in a group shares the same `host:fetchPort`, so
another outer attempt targets the same endpoint. In `311bc8c1c` I removed the
outer retry and now use one representative location per group;
`TransportClientFactory` owns retries through `celeborn.data.io.maxRetries`. I
also removed the synthetic test that made success depend on `PartitionLocation`
rather than the endpoint.
##########
common/src/main/java/org/apache/celeborn/common/network/client/TransportClientFactory.java:
##########
@@ -160,9 +161,10 @@ public TransportClient retryCreateClient(
try {
return createClient(remoteHost, remotePort, partitionId,
supplier.get());
} catch (Exception e) {
- if (e instanceof InterruptedException) {
+ InterruptedException interruptedException =
findInterruptedException(e);
Review Comment:
Thanks, addressed in `311bc8c1c`. The shared `tryCreateClient` helper now
covers both sequential and parallel batch-open, and `CelebornInputStream`
detects direct or wrapped interruption at reader creation, failed-stream
cleanup, reconnect, and buffer-fill boundaries. Those paths restore the flag
and exit before exclusion, retry, peer failover, or
`reportShuffleFetchFailure`. The push paths already distinguish
`InterruptedException` and pass it unchanged to their callbacks, so I left them
unchanged. Tests cover initial creation, cleanup, and reconnect; the reconnect
test also verifies that a cleanup failure cannot mask cancellation.
--
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]
