shlomitubul opened a new pull request, #3845:
URL: https://github.com/apache/celeborn/pull/3845
### What changes were proposed in this pull request?
`LifecycleManager` now tracks the Celeborn shuffle ids released by
`unregisterShuffle` and replies a
new `StatusCode.SHUFFLE_EXPIRED` for them, instead of falling through to the
`SHUFFLE_UNREGISTERED`
reply that the client turns into an empty file group.
- `expiredShuffleIds` is populated in `unregisterShuffle`, next to the
existing
`unregisterShuffleTime.put`, so the id is recorded at least one
`celeborn.client.shuffle.expired.checkInterval` before
`removeExpiredShuffle` erases any state.
- `handleGetReducerFileGroup` checks it before the `registeredShuffle`
check, since a released id is
by then absent from `registeredShuffle`.
- The entry is dropped in `unregisterAppShuffle`, when Spark has
unregistered the app shuffle and no
reader can exist.
The existing `SHUFFLE_UNREGISTERED` reply and its empty file group are
unchanged, so a map stage
that wrote no data still reads as empty. No client-side change is needed:
`SHUFFLE_EXPIRED` falls
through the existing `default` case in `loadFileGroupInternal`, which
returns a null file group and
makes `updateFileGroup` throw `CelebornIOException`.
### Why are the changes needed?
`SHUFFLE_UNREGISTERED` currently means two different things and the
LifecycleManager cannot tell
them apart once a shuffle has been swept:
- the shuffle was never registered because its map stage wrote no data,
which is expected because
registration is lazy, on the first push, so an empty file group is correct;
- the shuffle was registered and has since been released, so its data is
gone and must be
recomputed.
The second case is reachable while reduce tasks are still running. A task
resolves a valid shuffle
id at the top of `CelebornShuffleReader.read`; a fetch failure elsewhere in
the stage invalidates
that id and Spark reruns the map stage under a new id; with
`celeborn.client.spark.fetch.cleanFailedShuffle` enabled (CELEBORN-1896)
`FailedShuffleCleaner`
releases the old id and `removeExpiredShuffle` erases it. Spark does not
cancel the tasks of a
failed stage attempt (CELEBORN-1904), so a task from the first step can call
`updateFileGroup`
afterwards, receive an empty file group, read zero records and **report
success**.
Spark keeps the first successful result per partition even when it comes
from a zombie stage
attempt, so that empty output becomes the partition's final output and the
retry attempt skips it.
The application succeeds with missing rows and no failure appears in the
logs. This was observed in
production on a large reduce stage after a single fetch failure.
Making the read path fail closed is complementary to CELEBORN-1904:
cancelling the tasks of a failed
stage attempt narrows the window but cannot close it.
### Does this PR resolve a correctness bug?
- [x] Yes
### Does this PR introduce _any_ user-facing change?
- [ ] Yes
A reduce task that reads a released shuffle now fails instead of silently
returning no rows. The
failure is confined to tasks of an abandoned stage attempt, whose failures
Spark does not count
(`TaskSetManager` skips accounting when `isZombie`), and a live task that
hit this would retry,
resolve the current shuffle id and read the correct data. No configuration
is added or changed.
### How was this patch tested?
New case in `LifecycleManagerUnregisterShuffleSuite` asserting that, for the
same
`GetReducerFileGroup` request, a registered shuffle replies `SUCCESS`, the
same id after
`unregisterShuffle` plus the `removeExpiredShuffle` sweep replies
`SHUFFLE_EXPIRED`, and an id that
was never registered still replies `SHUFFLE_UNREGISTERED` with an empty file
group.
`LifecycleManagerUnregisterShuffleSuite` passes in full (7 tests) with
`-Pspark-3.5`.
--
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]
