Copilot commented on code in PR #3731:
URL: https://github.com/apache/celeborn/pull/3731#discussion_r3586275594
##########
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)
+ workerSource.incCounter(storageMetrics._3)
+ client.getChannel.writeAndFlush(new ChunkFetchFailure(streamChunkSlice,
message))
+ return
+ }
val storageMetrics = streamState.buffers match {
Review Comment:
`storageMetrics` is referenced inside the new `streamState == null` branch
before it is defined (and it can’t be computed when `streamState` is null).
This will not compile and also defeats the intent of moving the null-guard
earlier. Recommend removing that reference and incrementing a dedicated/neutral
failure counter for 'unknown stream' (or skipping metrics increment here), and
only using `storageMetrics` after it’s computed from a non-null `streamState`.
##########
worker/src/test/java/org/apache/celeborn/service/deploy/worker/FetchHandlerSuiteJ.java:
##########
@@ -302,6 +303,38 @@ public void
testDoNotDeleteOriginalFileWhenNonRangeLocalReadWorkInProgress() thr
}
}
+ @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();
Review Comment:
`EmbeddedChannel.readOutbound()` returns `Object`, so this assignment won’t
compile without an explicit cast. Update to cast the result (and consider
asserting the outbound message type before casting if you want clearer
failures).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
