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:
@shlomitubul, 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.
--
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]
