afterincomparableyum commented on code in PR #3731:
URL: https://github.com/apache/celeborn/pull/3731#discussion_r3497193052
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/WorkerSource.scala:
##########
@@ -181,6 +182,7 @@ object WorkerSource {
val FETCH_LOCAL_CHUNK_SUCCESS_COUNT = "FetchLocalChunkSuccessCount"
val FETCH_MEMORY_CHUNK_FAIL_COUNT = "FetchMemoryChunkFailCount"
val FETCH_LOCAL_CHUNK_FAIL_COUNT = "FetchLocalChunkFailCount"
+ val FETCH_CHUNK_UNREGISTERED_STREAM_FAIL_COUNT =
"FetchChunkUnregisteredStreamFailCount"
Review Comment:
CoPilot suggested we track it:
In the streamState == null branch, incrementing FETCH_LOCAL_CHUNK_FAIL_COUNT
will attribute "unregistered stream" failures to local chunk fetches, even
though the storage type is unknown. Consider adding/using a dedicated counter
for this condition (or otherwise avoiding storage-specific metrics here) to
keep monitoring accurate.
I'm fine reverting this change though. I will do that.
##########
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(WorkerSource.FETCH_CHUNK_UNREGISTERED_STREAM_FAIL_COUNT)
Review Comment:
got it.
--
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]
