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

[CELEBORN-645][REFACTOR] Refine logic about handle HeartbeatFromWorkerResponse

Browse Source 
### What changes were proposed in this pull request?
Refine the logic here to make it easier understand.

### Why are the changes needed?

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

### How was this patch tested?

Closes #1555 from AngersZhuuuu/CELEBORN-645.

Authored-by: Angerszhuuuu <angers.zhu@gmail.com>
Signed-off-by: Angerszhuuuu <angers.zhu@gmail.com>
This commit is contained in:
Angerszhuuuu 2023-06-07 16:34:44 +08:00
parent 9502b1f26d
commit d4cb6dd8ab
1 changed files with 3 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala
View File

@ -304,13 +304,10 @@ private[celeborn] class Worker(
activeShuffleKeys,
estimatedAppDiskUsage),
classOf[HeartbeatResponse])
if (response.registered) {
response.expiredShuffleKeys.asScala.foreach(shuffleKey => workerInfo.releaseSlots(shuffleKey))
cleanTaskQueue.put(response.expiredShuffleKeys)
} else {
response.expiredShuffleKeys.asScala.foreach(shuffleKey => workerInfo.releaseSlots(shuffleKey))
cleanTaskQueue.put(response.expiredShuffleKeys)
if (!response.registered) {
logError("Worker not registered in master, clean expired shuffle data and register again.")
// Clean expired shuffle.
cleanup(response.expiredShuffleKeys)
try {
registerWithMaster()
} catch {