SteNicholas commented on code in PR #3746:
URL: https://github.com/apache/celeborn/pull/3746#discussion_r3487543186
##########
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:
All locations grouped under one `hostPort` share the same `host:fetchPort`
(`groupOpenStreamLocations` keys by `hostAndFetchPort`, and the `createClient`
lambda at ~304 uses only `getHost`/`getFetchPort`). So the "second" attempt
re-targets the **same** worker and re-runs `TransportClientFactory`'s full
`maxIORetries` connect budget against an already-dead host:port — there is no
real same-worker fallback within a group (the replica lives on a different host
= a separate group/task). `MAX_CLIENT_CREATION_ATTEMPTS_PER_HOST = 1` would
yield the same set of clients with half the connect-retry latency per dead
worker; if you intentionally keep 2, the comment about "same-worker fallback"
is misleading.
##########
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:
Unwrapping a wrapped `InterruptedException` to stop retries is reasonable,
but note this also calls `Thread.currentThread().interrupt()` and throws a bare
`InterruptedException` from shared infra used by ~6 `createClient` callers,
while only the parallel reader (`createClientsInParallel`) added a matching
`case ex: InterruptedException`. The others —
`CelebornInputStream.createReaderWithRetry` (`catch (Exception)` →
`excludeFailedFetchLocation` + `Uninterruptibles.sleepUninterruptibly`, which
re-asserts the flag), the non-parallel `makeOpenStreamList`, and the
`ShuffleClientImpl`/`PushDataHandler` push paths — swallow it generically.
Because Netty `await()` throws immediately when the interrupt flag is preset,
the leftover flag makes each subsequent `createClient` fast-fail in a cascade,
and the cancellation is recorded as a worker failure (shared, cross-task
exclusion state on replicated clusters). Consider handling
`InterruptedException` consistently across these call site
s, or documenting the trade-off.
##########
common/src/main/java/org/apache/celeborn/common/network/client/TransportClientFactory.java:
##########
@@ -315,14 +328,19 @@ public void initChannel(SocketChannel ch) {
long preConnect = System.nanoTime();
ChannelFuture cf = bootstrap.connect(address);
if (connectTimeoutMs <= 0) {
- cf.await();
+ awaitWithChannelCleanup(
+ () -> {
+ cf.await();
+ return true;
+ },
+ cf);
assert cf.isDone();
if (cf.isCancelled()) {
Review Comment:
Minor / pre-existing but now inconsistent with the new cleanup paths: in
this `connectTimeoutMs <= 0` branch, the `isCancelled()` (339-341) and
`!isSuccess()` (340-342) failures throw without `closeChannel(cf)`, leaking the
half-open channel — whereas the `connectTimeoutMs > 0` branch below closes on
both timeout and `cause != null`, and the new interrupt path closes too. Worth
closing the channel here as well.
--
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]
