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 b3061925ff [CELEBORN-2356] Fix NPE before null-check in
handleChunkFetchRequest
b3061925ff is described below
commit b3061925ffd46854084a6e1117802d209253a562
Author: afterincomparableyum
<[email protected]>
AuthorDate: Mon Oct 12 17:02:32 2026 +0800
[CELEBORN-2356] Fix NPE before null-check in handleChunkFetchRequest
### What changes were proposed in this pull request?
getStreamState returns null for an unknown id. The streamState.buffers
pattern match dereferences it before the null guard runs, throwing
NullPointerException. The intended ChunkFetchFailure reply is unreachable. Here
we should reorder null-check first, then compute storageMetrics.
### Why are the changes needed?
To prevent potential NPEs.
### Does this PR resolve a correctness bug?
- [x] Yes
### Does this PR introduce _any_ user-facing change?
- [ ] Yes
### How was this patch tested?
CI
Closes #3850 from afterincomparableyum/CELEBORN-2356.
Authored-by: afterincomparableyum
<[email protected]>
Signed-off-by: zhengtao <[email protected]>
AI-Contributed/Feature: 0/16
AI-Contributed/UT: 0/33
(cherry picked from commit 29f3bae144de507e23fcb1ae92470e9a0dad5c50)
Signed-off-by: zhengtao <[email protected]>
---
.../service/deploy/worker/FetchHandler.scala | 16 +++++------
.../service/deploy/worker/FetchHandlerSuiteJ.java | 33 ++++++++++++++++++++++
2 files changed, 41 insertions(+), 8 deletions(-)
diff --git
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/FetchHandler.scala
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/FetchHandler.scala
index 7ad990e2bf..31f0d9f226 100644
---
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/FetchHandler.scala
+++
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/FetchHandler.scala
@@ -557,6 +557,14 @@ class FetchHandler(
s" to fetch block $streamChunkSlice")
val streamState =
chunkStreamManager.getStreamState(streamChunkSlice.streamId)
+ if (streamState == null) {
+ val message = s"Stream ${streamChunkSlice.streamId} is not registered
with worker. " +
+ "This can happen if the worker was restarted recently."
+ logError(message)
+ // The storage type is unknown for an unregistered stream, so skip
storage-specific metrics.
+ client.getChannel.writeAndFlush(new ChunkFetchFailure(streamChunkSlice,
message))
+ return
+ }
val storageMetrics = streamState.buffers match {
case _: FileChunkBuffers => (
WorkerSource.FETCH_LOCAL_CHUNK_TIME,
@@ -567,14 +575,6 @@ class FetchHandler(
WorkerSource.FETCH_MEMORY_CHUNK_SUCCESS_COUNT,
WorkerSource.FETCH_MEMORY_CHUNK_FAIL_COUNT)
}
- if (streamState == null) {
- val message = s"Stream ${streamChunkSlice.streamId} is not registered
with worker. " +
- "This can happen if the worker was restart recently."
- logError(message)
- workerSource.incCounter(storageMetrics._3)
- client.getChannel.writeAndFlush(new ChunkFetchFailure(streamChunkSlice,
message))
- return
- }
maxChunkBeingTransferred.foreach { threshold =>
val chunksBeingTransferred = chunkStreamManager.chunksBeingTransferred
// take high cpu usage
diff --git
a/worker/src/test/java/org/apache/celeborn/service/deploy/worker/FetchHandlerSuiteJ.java
b/worker/src/test/java/org/apache/celeborn/service/deploy/worker/FetchHandlerSuiteJ.java
index b54a0b99a5..76fb857e48 100644
---
a/worker/src/test/java/org/apache/celeborn/service/deploy/worker/FetchHandlerSuiteJ.java
+++
b/worker/src/test/java/org/apache/celeborn/service/deploy/worker/FetchHandlerSuiteJ.java
@@ -50,6 +50,7 @@ import
org.apache.celeborn.common.network.buffer.NioManagedBuffer;
import org.apache.celeborn.common.network.client.RpcResponseCallback;
import org.apache.celeborn.common.network.client.TransportClient;
import org.apache.celeborn.common.network.client.TransportResponseHandler;
+import org.apache.celeborn.common.network.protocol.ChunkFetchFailure;
import org.apache.celeborn.common.network.protocol.ChunkFetchSuccess;
import org.apache.celeborn.common.network.protocol.Message;
import org.apache.celeborn.common.network.protocol.OpenStream;
@@ -302,6 +303,38 @@ public class FetchHandlerSuiteJ {
}
}
+ @Test
+ public void testFetchChunkForUnregisteredStream() {
+ EmbeddedChannel channel = new EmbeddedChannel();
+ TransportClient client = new TransportClient(channel,
mock(TransportResponseHandler.class));
+ FetchHandler fetchHandler = mockFetchHandler(null);
+
+ // the stream is never opened, so it is unknown to the ChunkStreamManager
+ long unregisteredStreamId = 12345;
+ fetchHandler.receive(
+ client,
+ new RpcRequest(
+ TransportClient.requestId(),
+ new NioManagedBuffer(
+ new TransportMessage(
+ MessageType.CHUNK_FETCH_REQUEST,
+ PbChunkFetchRequest.newBuilder()
+ .setStreamChunkSlice(
+ PbStreamChunkSlice.newBuilder()
+ .setStreamId(unregisteredStreamId)
+ .setChunkIndex(0)
+ .setOffset(0)
+ .setLen(Integer.MAX_VALUE))
+ .build()
+ .toByteArray())
+ .toByteBuffer())),
+ createRpcResponseCallback(channel));
+
+ ChunkFetchFailure chunkFetchFailure = channel.readOutbound();
+ assertEquals(unregisteredStreamId,
chunkFetchFailure.streamChunkSlice.streamId);
+ assertTrue(chunkFetchFailure.errorString.contains("not registered"));
+ }
+
private FetchHandler mockFetchHandler(FileInfo fileInfo) {
WorkerSource workerSource = mock(WorkerSource.class);
TransportConf transportConf =
