@@ -73,9 +73,8 @@ class BlockManagerMasterActor(val isLocal: Boolean, conf: SparkConf, listenerBus
73
73
74
74
case UpdateBlockInfo (
75
75
blockManagerId, blockId, storageLevel, deserializedSize, size, tachyonSize) =>
76
- // TODO: Ideally we want to handle all the message replies in receive instead of in the
77
- // individual private methods.
78
- updateBlockInfo(blockManagerId, blockId, storageLevel, deserializedSize, size, tachyonSize)
76
+ sender ! updateBlockInfo(
77
+ blockManagerId, blockId, storageLevel, deserializedSize, size, tachyonSize)
79
78
80
79
case GetLocations (blockId) =>
81
80
sender ! getLocations(blockId)
@@ -351,23 +350,23 @@ class BlockManagerMasterActor(val isLocal: Boolean, conf: SparkConf, listenerBus
351
350
storageLevel : StorageLevel ,
352
351
memSize : Long ,
353
352
diskSize : Long ,
354
- tachyonSize : Long ) {
353
+ tachyonSize : Long ): Boolean = {
355
354
355
+ var updated = true
356
356
if (! blockManagerInfo.contains(blockManagerId)) {
357
357
if (blockManagerId.isDriver && ! isLocal) {
358
358
// We intentionally do not register the master (except in local mode),
359
359
// so we should not indicate failure.
360
- sender ! true
360
+ // do nothing here, "updated == true".
361
361
} else {
362
- sender ! false
362
+ updated = false
363
363
}
364
- return
364
+ return updated
365
365
}
366
366
367
367
if (blockId == null ) {
368
368
blockManagerInfo(blockManagerId).updateLastSeenMs()
369
- sender ! true
370
- return
369
+ return updated
371
370
}
372
371
373
372
blockManagerInfo(blockManagerId).updateBlockInfo(
@@ -391,7 +390,7 @@ class BlockManagerMasterActor(val isLocal: Boolean, conf: SparkConf, listenerBus
391
390
if (locations.size == 0 ) {
392
391
blockLocations.remove(blockId)
393
392
}
394
- sender ! true
393
+ updated
395
394
}
396
395
397
396
private def getLocations (blockId : BlockId ): Seq [BlockManagerId ] = {
0 commit comments