[CELEBORN-2361] Unblock stage-end when CommitFilesResponse carries an empty map-id bitmap

[afterincomparableyum]

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 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 the ConcurrentHashMap.putAll.
• ReducePartitionCommitHandler: wrap the commit in try/finally so the in-process stage-end marker is always cleared, allowing retry.
• CelebornInputStream.skipLocation: treat a null mapId bitmap as "do not skip", guarding against the same empty->null round-trip in the read path.

I added a UtilsSuite regression test asserting a CommitFilesResponse with an empty bitmap survives the transport round-trip without null map values.

Why are the changes needed?

Without this, spark clients can hang for a really long time and see lots of NPE errors.

Does this PR resolve a correctness bug?

• Yes

Does this PR introduce any user-facing change?

• Yes

How was this patch tested?

CI/CD

[SteNicholas]

@afterincomparableyum, please resolve the conflicts firstly.

[Copilot]

Pull request overview

Fixes a stage-end hang/NPE path triggered when an empty RoaringBitmap is serialized as ByteString.EMPTY and then deserialized to null, which previously propagated into a ConcurrentHashMap.putAll and could prevent stage-end cleanup.

Changes:

• Drop null map-id bitmaps while deserializing CommitFilesResponse to prevent ConcurrentHashMap.putAll NPEs.
• Ensure ReducePartitionCommitHandler.tryFinalCommit always clears the in-process stage-end marker via try/finally, enabling retries after failures.
• Guard the read path so CelebornInputStream treats a null map-id bitmap as “do not skip”.
• Add a regression test covering the transport round-trip behavior for empty bitmaps in CommitFilesResponse.

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated no comments.

File	Description
common/src/test/scala/org/apache/celeborn/common/util/UtilsSuite.scala	Adds a regression test ensuring empty bitmap entries don’t round-trip into null map values and don’t break ConcurrentHashMap.putAll.
common/src/main/scala/org/apache/celeborn/common/protocol/message/ControlMessages.scala	Skips deserialized null bitmaps when building CommitFilesResponse from protobuf.
client/src/main/scala/org/apache/celeborn/client/commit/ReducePartitionCommitHandler.scala	Wraps final commit in try/finally to always clear the in-process stage-end marker.
client/src/main/java/org/apache/celeborn/client/read/CelebornInputStream.java	Prevents NPE and preserves correctness by treating null bitmap hints as “don’t skip”.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[afterincomparableyum]

@SteNicholas I resolved the conflicts