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

[CELEBORN-555][REFACTOR] Avoid prin noisy blacklist info when record blacklist (#1460)

Browse Source 
* [CELEBORN-555][REFACTOR] Avoid prin noisy blacklist info when record blacklist
This commit is contained in:
Angerszhuuuu 2023-04-26 16:45:44 +08:00 committed by GitHub
parent 13ce04f8a1
commit 4bbc8aec4f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 28 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

44
client/src/main/scala/org/apache/celeborn/client/WorkerStatusTracker.scala
View File

@ -99,25 +99,37 @@ class WorkerStatusTracker(
case _ =>
}
}
if (!failedWorker.isEmpty) {
recordWorkerFailure(failedWorker)
}
recordWorkerFailure(failedWorker)
}
def recordWorkerFailure(failures: ShuffleFailedWorkers): Unit = {
val failedWorker = new ShuffleFailedWorkers(failures)
logInfo(s"Report Worker Failure: ${failedWorker.asScala}, current blacklist $blacklist")
failedWorker.asScala.foreach { case (worker, (statusCode, registerTime)) =>
if (!blacklist.containsKey(worker)) {
blacklist.put(worker, (statusCode, registerTime))
} else {
statusCode match {
case StatusCode.WORKER_SHUTDOWN |
StatusCode.NO_AVAILABLE_WORKING_DIR |
StatusCode.RESERVE_SLOTS_FAILED |
StatusCode.UNKNOWN_WORKER =>
blacklist.put(worker, (statusCode, blacklist.get(worker)._2))
case _ => // Not cover
if (!failures.isEmpty) {
val failedWorker = new ShuffleFailedWorkers(failures)
val failedWorkerMsg = failedWorker.asScala.map { case (worker, (status, time)) =>
s"${worker.readableAddress()} ${status.name()} $time"
}.mkString("\n")
val blacklistMsg = blacklist.asScala.map { case (worker, (status, time)) =>
s"${worker.readableAddress()} ${status.name()} $time"
}.mkString("\n")
logInfo(
s"""
|Reporting Worker Failure:
|$failedWorkerMsg
|Current blacklist:
|$blacklistMsg
""".stripMargin)
failedWorker.asScala.foreach { case (worker, (statusCode, registerTime)) =>
if (!blacklist.containsKey(worker)) {
blacklist.put(worker, (statusCode, registerTime))
} else {
statusCode match {
case StatusCode.WORKER_SHUTDOWN |
StatusCode.NO_AVAILABLE_WORKING_DIR |
StatusCode.RESERVE_SLOTS_FAILED |
StatusCode.UNKNOWN_WORKER =>
blacklist.put(worker, (statusCode, blacklist.get(worker)._2))
case _ => // Not cover
}
}
}
}