[CELEBORN-2361] Unblock stage-end when CommitFilesResponse carries an empty map-id bitmap - #3833
Open
strelok89 wants to merge 1 commit into
Open
[CELEBORN-2361] Unblock stage-end when CommitFilesResponse carries an empty map-id bitmap#3833strelok89 wants to merge 1 commit into
strelok89 wants to merge 1 commit into
Conversation
… empty map-id bitmap ### What changes were proposed in this pull request? A worker serializes an empty RoaringBitmap as an empty protobuf bytestring, and Utils.byteStringToRoaringBitmap deserializes empty bytes back to null. That null was stored in CommitFilesResponse.committedMapIdBitMap (a HashMap, which allows null values) and then merged into the ConcurrentHashMap ShuffleCommittedInfo.committedMapIdBitmap via putAll, which rejects null values and throws NPE in CommitHandler.processResponse. Since ReducePartitionCommitHandler.tryFinalCommit had no try/finally, the shuffle was left in inProcessStageEndShuffleSet and setStageEnd was never called, so every GetReducerFileGroup request parked until the executor RPC timeout, hanging the Spark client process. Fixes: - ControlMessages: skip null bitmaps when deserializing CommitFilesResponse so null never reaches ConcurrentHashMap.putAll. - ReducePartitionCommitHandler: wrap the commit in try/finally so the in-process stage-end marker is always cleared, allowing a retry. - CelebornInputStream.skipLocation: treat a null mapId bitmap as "do not skip". This also fixes a latent NPE independent of the empty-bitmap path: CommitHandler already calls setMapIdBitMap with committedMapIdBitmap.get(id), which is null whenever the id has no entry. Note the null-means-unknown semantics are kept deliberately: the wire format collapses null and empty to ByteString.EMPTY, so making byteStringToRoaringBitmap return an empty bitmap would turn "unknown" into "no map ids here" at the PbSerDeUtils call sites and make skipLocation skip locations that must be read. ### Why are the changes needed? Without this, Spark clients can hang for a long time and emit repeated NPEs. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Added a UtilsSuite regression test asserting a CommitFilesResponse with an empty bitmap survives the transport round-trip without null map values; verified it fails without the ControlMessages fix and passes with it. celeborn-client/Test/compile and spotless:check pass locally. Co-Authored-By: afterincomparableyum <224495379+afterincomparableyum@users.noreply.github.com> Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Contributor
|
hey @strelok89 thanks for reviving this pr. i’ve been a bit busy this last month, but was planing on re opening all my closed PRs in a few days. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
This revives #3736 by @afterincomparableyum, which was closed by the stale bot rather than
rejected. Rebased onto current
main, credit preserved viaCo-Authored-By.A worker serializes an empty
RoaringBitmapas an empty protobuf bytestring, andUtils.byteStringToRoaringBitmapdeserializes empty bytes back tonull. Thatnullwasstored in
CommitFilesResponse.committedMapIdBitMap(aHashMap, which allows null values)and then merged into the
ConcurrentHashMapShuffleCommittedInfo.committedMapIdBitmapviaputAll, which rejects null values and throws NPE inCommitHandler.processResponse.Since
ReducePartitionCommitHandler.tryFinalCommithad notry/finally, the shuffle wasleft in
inProcessStageEndShuffleSetandsetStageEndwas never called, so everyGetReducerFileGrouprequest parked until the executor RPC timeout, hanging the Sparkclient process.
Fixes:
ControlMessages: skip null bitmaps when deserializingCommitFilesResponsesonullnever reaches
ConcurrentHashMap.putAll.ReducePartitionCommitHandler: wrap the commit intry/finallyso the in-processstage-end marker is always cleared, allowing a retry.
CelebornInputStream.skipLocation: treat a null mapId bitmap as "do not skip". This alsofixes a latent NPE independent of the empty-bitmap path:
CommitHandleralready callssetMapIdBitMapwithcommittedMapIdBitmap.get(id), which isnullwhenever the id hasno entry.
The null-means-unknown semantics are kept deliberately. The wire format collapses null and
empty to
ByteString.EMPTY, so makingbyteStringToRoaringBitmapreturn an empty bitmapwould turn "unknown" into "no map ids here" at the
PbSerDeUtilscall sites and makeskipLocationskip locations that must be read.Why are the changes needed?
Without this, Spark clients can hang for a long time and emit repeated NPEs.
Does this PR resolve a correctness bug?
Does this PR introduce any user-facing change?
How was this patch tested?
Added a
UtilsSuiteregression test asserting aCommitFilesResponsewith an empty bitmapsurvives the transport round-trip without null map values. Verified it fails without the
ControlMessagesfix and passes with it.celeborn-client/Test/compileandspotless:check -pl common,clientpass locally; the rest is left to CI.🤖 Generated with Claude Code