mridulm commented on code in PR #50439:
URL: https://github.com/apache/spark/pull/50439#discussion_r2040989059
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -863,36 +863,36 @@ class BlockManagerMasterEndpoint(
blockId: BlockId,
requesterHost: String): Option[BlockLocationsAndStatus] = {
val allLocations =
Option(blockLocations.get(blockId)).map(_.toSeq).getOrElse(Seq.empty)
- val hostLocalLocations = allLocations.filter(bmId => bmId.host ==
requesterHost)
-
val blockStatusWithBlockManagerId: Option[(BlockStatus, BlockManagerId)] =
(if (externalShuffleServiceRddFetchEnabled) {
- // if fetching RDD is enabled from the external shuffle service then
first try to find
- // the block in the external shuffle service of the same host
- val location = hostLocalLocations.find(_.port ==
externalShuffleServicePort)
- location
- .flatMap(blockStatusByShuffleService.get(_).flatMap(_.get(blockId)))
- .zip(location)
- } else {
- None
- })
- .orElse {
- // if the block is not found via the external shuffle service trying
to find it in the
- // executors running on the same host and persisted on the disk
- // using flatMap on iterators makes the transformation lazy
- hostLocalLocations.iterator
- .flatMap { bmId =>
- blockManagerInfo.get(bmId).flatMap { blockInfo =>
- blockInfo.getStatus(blockId).map((_, bmId))
- }
- }
- .find(_._1.storageLevel.useDisk)
- }
- .orElse {
- // if the block cannot be found in the same host search it in all
the executors
- val location = allLocations.headOption
-
location.flatMap(blockManagerInfo.get(_)).flatMap(_.getStatus(blockId)).zip(location)
+ // If fetching disk persisted RDD from the external shuffle service is
enabled then first
+ // try to find the block in the external shuffle service preferring
the one running on
+ // the same host. This search includes blocks stored on already killed
executors as well.
+ val hostLocalLocations = allLocations.find { bmId =>
+ bmId.host == requesterHost && bmId.port == externalShuffleServicePort
}
+ val location = hostLocalLocations
+ .orElse(allLocations.find(_.port == externalShuffleServicePort))
+ location
+ .flatMap(blockStatusByShuffleService.get(_).flatMap(_.get(blockId)))
+ .zip(location)
+ } else {
Review Comment:
When `externalShuffleServiceRddFetchEnabled` == `true` and we are not able
to find an ess location - but there could be an executor hosting the block, we
will now skip that else block.
I will need to see whether there is any race/etc where this is possible, but
why not preserve the earlier formulation ?
```
(if (externalShuffleServiceRddFetchEnabled) {
// current
} else {
None
}).orElse {
// else block
}.orElse ...
```
--
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: reviews-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
