Skip to content

[CELEBORN-2460] Reply SHUFFLE_EXPIRED for a released shuffle instead of an empty file group - #3845

Open
shlomitubul wants to merge 1 commit into
apache:mainfrom
shlomitubul:celeborn-expired-shuffle-file-group
Open

[CELEBORN-2460] Reply SHUFFLE_EXPIRED for a released shuffle instead of an empty file group#3845
shlomitubul wants to merge 1 commit into
apache:mainfrom
shlomitubul:celeborn-expired-shuffle-file-group

Conversation

@shlomitubul

Copy link
Copy Markdown
Contributor

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?

  • 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.

…of an empty file group

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
@Kalvin2077

Copy link
Copy Markdown
Contributor
  • Final-unregister paths can leave tombstones behind. unregisterShuffle adds them, but the stage-rerun-disabled branch and Flink job cleanup call it without removing them.

  • Delayed cleanup can re-add a tombstone after final unregistration. FailedShuffleCleaner can still process a queued or already-drained ID. The cleanup callback only removes the deduplication entry; it does not cancel that work.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants