apache/celeborn
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

[MINOR] Refine the log for fetch failure and rpc metrics dump

Browse Source 
### What changes were proposed in this pull request?
Minor change for the log, to make the fetch failure message and rpc metrics dump much clear.

### Why are the changes needed?
As title.

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

GA.

Closes #3136 from turboFei/log_shuffle.

Authored-by: Wang, Fei <fwang12@ebay.com>
Signed-off-by: SteNicholas <programgeek@163.com>
This commit is contained in:
Wang, Fei 2025-03-10 10:56:53 +08:00 committed by SteNicholas
parent 196ad607cd
commit 6f5ad2dde8
2 changed files with 3 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
client-spark/spark-2/src/main/scala/org/apache/spark/celeborn/ExceptionMakerHelper.scala
View File

@ -23,7 +23,7 @@ import org.apache.celeborn.common.util.ExceptionMaker
object ExceptionMakerHelper {
val FETCH_FAILURE_ERROR_MSG = "Celeborn FetchFailure with shuffle id "
val FETCH_FAILURE_ERROR_MSG = "Celeborn FetchFailure with appShuffleId/shuffleId: "
val SHUFFLE_FETCH_FAILURE_EXCEPTION_MAKER = new ExceptionMaker() {
override def makeFetchFailureException(

4
common/src/main/scala/org/apache/celeborn/common/rpc/RpcMetricsTracker.scala
View File

@ -86,7 +86,7 @@ private[celeborn] class RpcMetricsTracker(
if (slowRpcInterval < 0 || System.currentTimeMillis() - lastLogTime > slowRpcInterval &&
lastSlowLogTime.compareAndSet(lastLogTime, System.currentTimeMillis())) {
logWarning(
s"slow rpc detected: currentQueueSize = ${queueLengthFunc()}, queueTime=$queueTime processTime=$processTime message=$message")
s"slow rpc detected: currentQueueSize = ${queueLengthFunc()}, queueTime=$queueTime(ns) processTime=$processTime(ns) message=$message")
}
val lastTime = lastDumpTime.get
@ -144,7 +144,7 @@ private[celeborn] class RpcMetricsTracker(
return
val builder = new StringBuilder();
builder.append(s"RPC statistics for $name").append("\n")
builder.append(s"RPC statistics for $name (time unit: ns)").append("\n")
builder.append(s"current queue size = ${queueLengthFunc()}").append("\n")
builder.append(s"max queue length = ${maxQueueLength.get()}").append("\n")
histogramMap.entrySet.asScala.foreach(entry => {