peter-toth commented on code in PR #50439:
URL: https://github.com/apache/spark/pull/50439#discussion_r2039636849
##########
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 {
+ // trying to find it in the executors running on the same host and
persisted on the disk
+ // Implementation detail: using flatMap on iterators makes the
transformation lazy.
+ allLocations.find(_.host == requesterHost).iterator
Review Comment:
Shouldn't this be `.filter`?
--
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
