You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by za...@apache.org on 2026/10/12 10:05:13 UTC
(celeborn) branch branch-0.7 updated: [CELEBORN-2361] Unblock stage-end when CommitFilesResponse carries an empty map-id bitmap
This is an automated email from the ASF dual-hosted git repository.
zaynt4606 pushed a commit to branch branch-0.7
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/branch-0.7 by this push:
new a21d88acd9 [CELEBORN-2361] Unblock stage-end when CommitFilesResponse carries an empty map-id bitmap
a21d88acd9 is described below
commit a21d88acd9e61af4cb6e8827a5bc9a0223b00538
Author: strelok89 <al...@outlook.co.il>
AuthorDate: Mon Oct 12 17:47:47 2026 +0800
[CELEBORN-2361] Unblock stage-end when CommitFilesResponse carries an empty map-id bitmap
This revives #3736 by afterincomparableyum, which was closed by the stale bot rather than
rejected. Rebased onto current `main`, credit preserved via `Co-Authored-By`.
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.
The original revision of this PR also made `CelebornInputStream.skipLocation` treat a null
mapId bitmap as "do not skip". That has since landed on `main` via CELEBORN-2442
(`adadbd924`), which moved the logic into `CelebornInputStream.shouldSkipLocation` with the
same null-safe behavior, so it is dropped here.
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
`shouldSkipLocation` skip locations that must be read.
Without this, Spark clients can hang for a long time and emit repeated NPEs.
- [x] Yes
- [ ] Yes
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 -pl common,client` pass locally; the rest is left to CI.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Backport to branch-0.7: restore the null-bitmap guard in the legacy
CelebornInputStream.skipLocation path and retain the branch-specific UtilsSuite
imports. Range-read tests cover missing metadata with and without a peer,
and a known bitmap with no overlap.
Backport validation: 9 CelebornInputStreamPeerFailoverTest tests and 31 UtilsSuite
tests passed, together with spotless:check for common and client. Both unknown
bitmap tests reproduce the NPE when the read-side guard is removed.
Closes #3833 from strelok89/CELEBORN-2361.
Authored-by: strelok89 <al...@outlook.co.il>
Signed-off-by: zhengtao <sh...@alibaba-inc.com>
Co-Authored-By: afterincomparableyum <22...@users.noreply.github.com>
Co-Authored-By: Claude Opus 5 (1M context) <no...@anthropic.com>
Co-Authored-By: Codex <no...@openai.com>
(cherry picked from commit 71724db23980cc80cb920fee2a057b96f26893f7)
Signed-off-by: zhengtao <sh...@alibaba-inc.com>
AI-Model: gpt-6
AI-Contributed/Feature: 4/48
AI-Contributed/UT: 60/90
---
.../celeborn/client/read/CelebornInputStream.java | 4 ++
.../commit/ReducePartitionCommitHandler.scala | 37 +++++++------
.../read/CelebornInputStreamPeerFailoverTest.java | 60 ++++++++++++++++++++--
.../common/protocol/message/ControlMessages.scala | 7 ++-
.../apache/celeborn/common/util/UtilsSuite.scala | 30 ++++++++++-
5 files changed, 117 insertions(+), 21 deletions(-)
diff --git a/client/src/main/java/org/apache/celeborn/client/read/CelebornInputStream.java b/client/src/main/java/org/apache/celeborn/client/read/CelebornInputStream.java
index e9dc4b9625..cc873ec508 100644
--- a/client/src/main/java/org/apache/celeborn/client/read/CelebornInputStream.java
+++ b/client/src/main/java/org/apache/celeborn/client/read/CelebornInputStream.java
@@ -372,6 +372,10 @@ public abstract class CelebornInputStream extends InputStream {
if (bitmap == null && location.hasPeer()) {
bitmap = location.getPeer().getMapIdBitMap();
}
+ if (bitmap == null) {
+ // Missing filter metadata cannot prove that this location is irrelevant.
+ return false;
+ }
for (int i = startMapIndex; i < endMapIndex; i++) {
if (bitmap.contains(i)) {
return false;
diff --git a/client/src/main/scala/org/apache/celeborn/client/commit/ReducePartitionCommitHandler.scala b/client/src/main/scala/org/apache/celeborn/client/commit/ReducePartitionCommitHandler.scala
index b3e7aa90ab..febee90f8d 100644
--- a/client/src/main/scala/org/apache/celeborn/client/commit/ReducePartitionCommitHandler.scala
+++ b/client/src/main/scala/org/apache/celeborn/client/commit/ReducePartitionCommitHandler.scala
@@ -210,23 +210,28 @@ class ReducePartitionCommitHandler(
}
}
- // ask allLocations workers holding partitions to commit files
- val allocatedWorkers = shuffleAllocatedWorkers.get(shuffleId)
- val (dataLost, commitFailedWorkers) = handleFinalCommitFiles(shuffleId, allocatedWorkers)
- recordWorkerFailure(commitFailedWorkers)
- // reply
- if (!dataLost) {
- logInfo(s"Succeed to handle stageEnd for $shuffleId.")
- // record in stageEndShuffleSet
- setStageEnd(shuffleId)
- } else {
- logError(s"Failed to handle stageEnd for $shuffleId, lost file!")
- dataLostShuffleSet.add(shuffleId)
- // record in stageEndShuffleSet
- setStageEnd(shuffleId)
+ try {
+ // ask allLocations workers holding partitions to commit files
+ val allocatedWorkers = shuffleAllocatedWorkers.get(shuffleId)
+ val (dataLost, commitFailedWorkers) = handleFinalCommitFiles(shuffleId, allocatedWorkers)
+ recordWorkerFailure(commitFailedWorkers)
+ // reply
+ if (!dataLost) {
+ logInfo(s"Succeed to handle stageEnd for $shuffleId.")
+ // record in stageEndShuffleSet
+ setStageEnd(shuffleId)
+ } else {
+ logError(s"Failed to handle stageEnd for $shuffleId, lost file!")
+ dataLostShuffleSet.add(shuffleId)
+ // record in stageEndShuffleSet
+ setStageEnd(shuffleId)
+ }
+ true
+ } finally {
+ // Clear the in-process marker even if the commit threw, so stage-end can be retried
+ // instead of parking every GetReducerFileGroup until the RPC timeout.
+ inProcessStageEndShuffleSet.remove(shuffleId)
}
- inProcessStageEndShuffleSet.remove(shuffleId)
- true
}
private def handleFinalCommitFiles(
diff --git a/client/src/test/java/org/apache/celeborn/client/read/CelebornInputStreamPeerFailoverTest.java b/client/src/test/java/org/apache/celeborn/client/read/CelebornInputStreamPeerFailoverTest.java
index ec1003b7d0..215482b259 100644
--- a/client/src/test/java/org/apache/celeborn/client/read/CelebornInputStreamPeerFailoverTest.java
+++ b/client/src/test/java/org/apache/celeborn/client/read/CelebornInputStreamPeerFailoverTest.java
@@ -45,6 +45,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Before;
import org.junit.Test;
+import org.roaringbitmap.RoaringBitmap;
import org.apache.celeborn.client.ShuffleClient;
import org.apache.celeborn.client.security.CryptoHandler;
@@ -285,6 +286,52 @@ public class CelebornInputStreamPeerFailoverTest {
Optional.<CryptoHandler>empty());
}
+ @Test
+ public void testRangeReadWithUnknownBitmapWithoutPeerOpensReader() throws Exception {
+ PartitionLocation primary = createPartitionLocation(PRIMARY_HOST);
+ primary.setMapIdBitMap(null);
+ TransportClient client = mockReplicaClient();
+ when(clientFactory.createClient(PRIMARY_HOST, PORT + 2)).thenReturn(client);
+
+ try (CelebornInputStream inputStream = createRangeReadInputStream(primary)) {
+ verify(clientFactory).createClient(PRIMARY_HOST, PORT + 2);
+ }
+ }
+
+ @Test
+ public void testRangeReadWithBothBitmapsUnknownOpensReader() throws Exception {
+ PartitionLocation primary = createPartitionLocation(PRIMARY_HOST);
+ PartitionLocation replica = createPartitionLocation(REPLICA_HOST);
+ primary.setMapIdBitMap(null);
+ replica.setMapIdBitMap(null);
+ primary.setPeer(replica);
+ replica.setPeer(primary);
+ TransportClient client = mockReplicaClient();
+ when(clientFactory.createClient(PRIMARY_HOST, PORT + 2)).thenReturn(client);
+
+ try (CelebornInputStream inputStream = createRangeReadInputStream(primary)) {
+ verify(clientFactory).createClient(PRIMARY_HOST, PORT + 2);
+ }
+ }
+
+ @Test
+ public void testRangeReadWithNonoverlappingBitmapSkipsReader() throws Exception {
+ PartitionLocation primary = createPartitionLocation(PRIMARY_HOST);
+ primary.setMapIdBitMap(RoaringBitmap.bitmapOf(20));
+
+ try (CelebornInputStream inputStream = createRangeReadInputStream(primary)) {
+ assertEquals(-1, inputStream.read());
+ verify(clientFactory, never()).createClient(anyString(), anyInt());
+ }
+ }
+
+ private CelebornInputStream createRangeReadInputStream(PartitionLocation location)
+ throws IOException {
+ conf.set(CelebornConf.SHUFFLE_RANGE_READ_FILTER_ENABLED().key(), "true");
+ conf.set(CelebornConf.CLIENT_CHUNK_PREFETCH_ENABLED().key(), "false");
+ return createInputStream(location, null, 0, 10);
+ }
+
private CelebornInputStream createInputStream(String primaryHost, String replicaHost)
throws IOException {
return createInputStream(primaryHost, replicaHost, null);
@@ -297,8 +344,15 @@ public class CelebornInputStreamPeerFailoverTest {
primary.setPeer(replica);
replica.setPeer(primary);
+ return createInputStream(primary, exceptionMaker, 0, 100);
+ }
+
+ private CelebornInputStream createInputStream(
+ PartitionLocation location, ExceptionMaker exceptionMaker, int startMapIndex, int endMapIndex)
+ throws IOException {
+
ArrayList<PartitionLocation> locations = new ArrayList<>();
- locations.add(primary);
+ locations.add(location);
ArrayList<PbStreamHandler> streamHandlers = new ArrayList<>();
streamHandlers.add(PbStreamHandler.newBuilder().setStreamId(123L).setNumChunks(10).build());
@@ -314,8 +368,8 @@ public class CelebornInputStreamPeerFailoverTest {
new HashMap<>(),
0,
1L,
- 0,
- 100,
+ startMapIndex,
+ endMapIndex,
fetchExcludedWorkers,
shuffleClient,
1,
diff --git a/common/src/main/scala/org/apache/celeborn/common/protocol/message/ControlMessages.scala b/common/src/main/scala/org/apache/celeborn/common/protocol/message/ControlMessages.scala
index f1e34aa54f..59b1968ea7 100644
--- a/common/src/main/scala/org/apache/celeborn/common/protocol/message/ControlMessages.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/protocol/message/ControlMessages.scala
@@ -1480,7 +1480,12 @@ object ControlMessages extends Logging {
pbCommitFilesResponse.getCommittedReplicaStorageInfosMap.asScala.foreach(entry =>
committedReplicaStorageInfos.put(entry._1, StorageInfo.fromPb(entry._2)))
pbCommitFilesResponse.getMapIdBitmapMap.asScala.foreach { entry =>
- committedBitMap.put(entry._1, Utils.byteStringToRoaringBitmap(entry._2))
+ // An empty bitmap is serialized as ByteString.EMPTY and deserializes back to null.
+ // Skip it: null values break the ConcurrentHashMap.putAll in processResponse.
+ val bitmap = Utils.byteStringToRoaringBitmap(entry._2)
+ if (bitmap != null) {
+ committedBitMap.put(entry._1, bitmap)
+ }
}
CommitFilesResponse(
StatusCode.fromValue(pbCommitFilesResponse.getStatus),
diff --git a/common/src/test/scala/org/apache/celeborn/common/util/UtilsSuite.scala b/common/src/test/scala/org/apache/celeborn/common/util/UtilsSuite.scala
index c98f3bd498..4450489087 100644
--- a/common/src/test/scala/org/apache/celeborn/common/util/UtilsSuite.scala
+++ b/common/src/test/scala/org/apache/celeborn/common/util/UtilsSuite.scala
@@ -22,6 +22,7 @@ import java.util
import java.util.Collections
import java.util.concurrent.atomic.AtomicInteger
+import org.roaringbitmap.RoaringBitmap
import org.scalatest.matchers.must.Matchers.contain
import org.scalatest.matchers.should.Matchers.convertToAnyShouldWrapper
@@ -32,7 +33,7 @@ import org.apache.celeborn.common.exception.CelebornException
import org.apache.celeborn.common.identity.DefaultIdentityProvider
import org.apache.celeborn.common.network.protocol.SerdeVersion
import org.apache.celeborn.common.protocol.{PartitionLocation, PbReviseLostShuffles, PbReviseLostShufflesResponse, TransportModuleConstants}
-import org.apache.celeborn.common.protocol.message.ControlMessages.{GetReducerFileGroupResponse, MapperEnd, ReviseLostShuffles, ReviseLostShufflesResponse}
+import org.apache.celeborn.common.protocol.message.ControlMessages.{CommitFilesResponse, GetReducerFileGroupResponse, MapperEnd, ReviseLostShuffles, ReviseLostShufflesResponse}
import org.apache.celeborn.common.protocol.message.StatusCode
class UtilsSuite extends CelebornFunSuite {
@@ -282,6 +283,33 @@ class UtilsSuite extends CelebornFunSuite {
assert(set.size == 0)
}
+ test("CommitFilesResponse with empty map-id bitmap survives transport round-trip") {
+ // An empty RoaringBitmap serializes to ByteString.EMPTY and deserializes back to null,
+ // which used to NPE when merged into the ConcurrentHashMap in CommitHandler.processResponse.
+ val committedMapIdBitMap = new util.HashMap[String, RoaringBitmap]()
+ committedMapIdBitMap.put("0-0", new RoaringBitmap())
+ val nonEmpty = RoaringBitmap.bitmapOf(1, 2, 3)
+ committedMapIdBitMap.put("0-1", nonEmpty)
+
+ val response = CommitFilesResponse(
+ StatusCode.SUCCESS,
+ Collections.emptyList[String](),
+ Collections.emptyList[String](),
+ Collections.emptyList[String](),
+ Collections.emptyList[String](),
+ committedMapIdBitMap = committedMapIdBitMap)
+ val responseTrans = Utils.fromTransportMessage(Utils.toTransportMessage(response))
+ .asInstanceOf[CommitFilesResponse]
+
+ assert(!responseTrans.committedMapIdBitMap.containsValue(null))
+ assert(!responseTrans.committedMapIdBitMap.containsKey("0-0"))
+ assert(responseTrans.committedMapIdBitMap.get("0-1") == nonEmpty)
+
+ val concurrent = JavaUtils.newConcurrentHashMap[String, RoaringBitmap]()
+ concurrent.putAll(responseTrans.committedMapIdBitMap)
+ assert(concurrent.size() == 1)
+ }
+
test("validate number of client/server netty threads") {
val celebornConf = new CelebornConf()
celebornConf.set("celeborn.io.maxDefaultNettyThreads", "100")