[CELEBORN-2434] Clamp endMapIndex in sliceSortedBufferByMapRange

[yew1eb]

What changes were proposed in this pull request?

Add the missing endMapIndex == Integer.MAX_VALUE clamp in ShuffleBlockInfoUtils.sliceSortedBufferByMapRange, mirroring the existing guard in the sibling method getChunkOffsetsFromShuffleBlockInfos. Also add a unit test covering the full re-read (Integer.MAX_VALUE) case.

Why are the changes needed?

When a memory-stored shuffle partition is read once with a map range (e.g. AQE skew join) and then re-read in full (endIndex = Int.MaxValue, e.g. Spark ReusedExchange), FetchHandler.handleReduceOpenStreamInternal passes the raw Int.MaxValue into getSortedFileInfo because FileInfo.addStream returns false for the already-sorted file. The memory branch then loops startMapIndex..Integer.MAX_VALUE in sliceSortedBufferByMapRange — billions of TreeMap.get calls on the Netty fetch event loop, stalling all fetch/openStream traffic on that event loop for minutes to hours.

The disk path is not affected (it goes through the already-clamped getChunkOffsetsFromShuffleBlockInfos).

Symptoms / evidence:

• MemoryStorageReusedExchangeSuite (a single tiny test with 1000 rows) takes 28–54 minutes instead of ~1 minute, both in CI and locally.
• Client side repeatedly logs BatchOpenStream for N cost 240006ms (exactly the 240s fetch timeout), followed by retries that stall again.
• Client logs Connection ... has been quiet for 4.0 m while there are outstanding requests against both workers' fetch ports.
• A thread dump taken while hung shows the fetch event loop spinning:

  "fetch-server-34-2" RUNNABLE
    at java.util.TreeMap.getEntry(TreeMap.java:359)
    at java.util.TreeMap.get(TreeMap.java:278)
    at ShuffleBlockInfoUtils.sliceSortedBufferByMapRange(ShuffleBlockInfoUtils.java:140)
    at PartitionFilesSorter.getSortedFileInfo(PartitionFilesSorter.java:224)
    at FetchHandler.handleReduceOpenStreamInternal(FetchHandler.scala:277)

In production this means: memory storage + a range read followed by a full re-read of the same partition can wedge a worker's fetch event loop for an
unbounded time (the loop count scales with Int.MaxValue, multiplied by the number of files in the batch request).

Does this PR resolve a correctness bug?

• Yes

Does this PR introduce any user-facing change?

• Yes

How was this patch tested?

• Added testSliceSortedBufferByMapRangeWithMaxEndIndex to ShuffleBlockInfoUtilsTest, asserting a full re-read with Integer.MAX_VALUE produces exactly the same result as an explicit full-range read.
• ShuffleBlockInfoUtilsTest: all 6 tests pass.