SteNicholas commented on code in PR #3845:
URL: https://github.com/apache/celeborn/pull/3845#discussion_r4258722457
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -971,6 +972,15 @@ class LifecycleManager(val appUniqueId: String, val conf:
CelebornConf) extends
// If isSegmentGranularityVisible is set to true, the downstream reduce
task may start early than upstream map task, e.g. flink hybrid shuffle.
// Under these circumstances, there's a possibility that the shuffle might
not yet be registered when the downstream reduce task send GetReduceFileGroup
request,
// so we shouldn't send a SHUFFLE_NOT_REGISTERED response directly, should
enqueue this request to pending list, and response to the downstream reduce
task the ReduceFileGroup when the upstream map task register shuffle done
+ if (expiredShuffleIds.contains(shuffleId)) {
Review Comment:
[P1] Re-check expiration before returning an empty file group
These two set lookups are not atomic with cleanup. A request can observe
`expiredShuffleIds.contains(shuffleId) == false`, pause while
`unregisterShuffle` adds the marker and the expiry sweep removes the registered
entry, then resume at the `registeredShuffle` check and return
`SHUFFLE_UNREGISTERED` with an empty group. `LifecycleManager` permits
concurrent message processing, so the expiry delay does not exclude this
interleaving. I reproduced it by pausing the first lookup across the normal
cleanup timer: the response was `SHUFFLE_UNREGISTERED`, `fileGroup.isEmpty ==
true`. The Java client treats that as a successful empty read, preserving the
silent data-loss path this PR addresses. Please re-check the marker after
observing that the shuffle is unregistered, before replying with an empty
group, or synchronize this decision with cleanup.
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -1279,6 +1289,12 @@ class LifecycleManager(val appUniqueId: String, val
conf: CelebornConf) extends
}
}
}
+ if (celebornShuffleIdToAppShuffleIdMap.containsKey(shuffleId)) {
+ val appShuffleId = celebornShuffleIdToAppShuffleIdMap.get(shuffleId)
+ if (shuffleIdMapping.containsKey(appShuffleId) &&
registeredShuffle.contains(shuffleId)) {
+ expiredShuffleIds.add(shuffleId)
Review Comment:
[P2] Retire existing expired markers during final shuffle cleanup
The new guards prevent final unregistration from adding a marker, but they
do not remove markers already added by an earlier failed-copy cleanup. There is
no `remove` or `clear` for `expiredShuffleIds`, including in
`unregisterAppShuffle`, so a long-lived SparkContext retains one entry per
released copy across completed jobs indefinitely. A probe that allocates and
registers a mapped shuffle, calls `unregisterShuffle`, then calls
`unregisterAppShuffle(appShuffleId, true)` still finds the expired marker.
Please tie marker retention to the app-shuffle lifecycle and retire those
existing entries when they are no longer needed, coordinating with the
failed-shuffle cleaner so delayed cleanup cannot re-add them.
--
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]
