You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "bozhang2820 (via GitHub)" <gi...@apache.org> on 2023/08/02 11:01:49 UTC
[GitHub] [spark] bozhang2820 opened a new pull request, #42296: [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions
bozhang2820 opened a new pull request, #42296:
URL: https://github.com/apache/spark/pull/42296
### What changes were proposed in this pull request?
This change tries to handle shuffle fetch failures due to decommissions.
When encountering a fetch failure for a block, `ShuffleBlockFetcherIterator` will check the updated map output location (`BlockManagerId`) for it, and see if the location changes. When it does, the fetch failure will be ignored and the updated location and the block id will be recorded. At the end of the fetch request (when all blocks in the request are processed), multiple requests will be assembled and enqueued, to retry the fetches from the updated locations for the corresponding blocks.
### Why are the changes needed?
This is to improve query stability when decommission is enabled.
### Does this PR introduce _any_ user-facing change?
This change comes with a feature flag `spark.storage.decommission.shuffleBlocks.refreshLocationsEnabled`.
### How was this patch tested?
Added a new unit test.
--
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-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions [spark]
Posted by "EnricoMi (via GitHub)" <gi...@apache.org>.
EnricoMi commented on PR #42296:
URL: https://github.com/apache/spark/pull/42296#issuecomment-4133368892
@Ngone51 @mridulm what are your thoughts on this? Would you be happy to get this over the finish line? Shall we create a new pull request base on this branch?
--
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-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions [spark]
Posted by "bozhang2820 (via GitHub)" <gi...@apache.org>.
bozhang2820 commented on PR #42296:
URL: https://github.com/apache/spark/pull/42296#issuecomment-1772350247
> @bozhang2820 not sure if you still have time to work on this PR. I opened [another PR](https://github.com/apache/spark/pull/43443) to address some issues mentioned above.
We have encountered some performance issues during our tests with this change, and will have to address those before moving forward.
--
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-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on pull request #42296: [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #42296:
URL: https://github.com/apache/spark/pull/42296#issuecomment-1663278588
+CC @otterc
--
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-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ukby1234 commented on a diff in pull request #42296: [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions
Posted by "ukby1234 (via GitHub)" <gi...@apache.org>.
ukby1234 commented on code in PR #42296:
URL: https://github.com/apache/spark/pull/42296#discussion_r1292045322
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -1288,6 +1288,30 @@ private[spark] class MapOutputTrackerWorker(conf: SparkConf) extends MapOutputTr
mapSizesByExecutorId.iter
}
+ def getMapOutputLocationWithRefresh(
+ shuffleId: Int,
+ mapId: Long,
+ prevLocation: BlockManagerId): BlockManagerId = {
+ // Try to get the cached location first in case other concurrent tasks
+ // fetched the fresh location already
+ var currentLocationOpt = getMapOutputLocation(shuffleId, mapId)
+ if (currentLocationOpt.isDefined && currentLocationOpt.get == prevLocation) {
+ // Address in the cache unchanged. Try to clean cache and get a fresh location
+ unregisterShuffle(shuffleId)
+ currentLocationOpt = getMapOutputLocation(shuffleId, mapId)
+ }
+ if (currentLocationOpt.isEmpty) {
+ throw new MetadataFetchFailedException(shuffleId, -1,
+ message = s"Failed to get map output location for shuffleId $shuffleId, mapId $mapId")
+ }
+ currentLocationOpt.get
Review Comment:
When shuffle fallback storage is enabled, this `currentLocationOpt`can be the `FALLBACK_BLOCK_MANAGER_ID`, and `DeferFetchRequestResult` below doesn't handle this special case.
so either 1) check the FetchRequest for fallback storage special ID 2)rewrite the RPC address to localhost so we get the blocks inside the fallback storage.
--
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-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions [spark]
Posted by "ukby1234 (via GitHub)" <gi...@apache.org>.
ukby1234 commented on code in PR #42296:
URL: https://github.com/apache/spark/pull/42296#discussion_r1366447643
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -355,8 +367,28 @@ final class ShuffleBlockFetcherIterator(
updateMergedReqsDuration(wasReqForMergedChunks = true)
results.put(FallbackOnPushMergedFailureResult(
block, address, infoMap(blockId)._1, remainingBlocks.isEmpty))
- } else {
+ } else if (!shouldPerformShuffleLocationRefresh) {
results.put(FailureFetchResult(block, infoMap(blockId)._2, address, e))
+ } else {
+ val (shuffleId, mapId) = BlockId.getShuffleIdAndMapId(block)
+ val mapOutputTrackerWorker = mapOutputTracker.asInstanceOf[MapOutputTrackerWorker]
+ Try(mapOutputTrackerWorker
+ .getMapOutputLocationWithRefresh(shuffleId, mapId, address)) match {
Review Comment:
Refreshing map output locations in a Netty callback thread will cause potential deadlock. Here is the reason:
1. Some map output locations are stored via broadcast variables
2. This code has a synchronization block
3. The netty response to fetch broadcast variables might be blocked by other handlers like the shuffle success handler
4. In the above case, because the shuffle success handler also requires the same lock from 2), this is a deadlock
The above situation happened during my test of this code running this patch.
--
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-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #42296: [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #42296:
URL: https://github.com/apache/spark/pull/42296#discussion_r1318086203
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -264,18 +272,22 @@ final class ShuffleBlockFetcherIterator(
case FetchBlockInfo(blockId, size, mapIndex) => (blockId.toString, (size, mapIndex))
}.toMap
val remainingBlocks = new HashSet[String]() ++= infoMap.keys
- val deferredBlocks = new ArrayBuffer[String]()
+ val deferredBlocks = new HashMap[BlockManagerId, Queue[String]]()
val blockIds = req.blocks.map(_.blockId.toString)
val address = req.address
val requestStartTime = clock.nanoTime()
@inline def enqueueDeferredFetchRequestIfNecessary(): Unit = {
if (remainingBlocks.isEmpty && deferredBlocks.nonEmpty) {
- val blocks = deferredBlocks.map { blockId =>
- val (size, mapIndex) = infoMap(blockId)
- FetchBlockInfo(BlockId(blockId), size, mapIndex)
+ val newAddressToBlocks = new HashMap[BlockManagerId, Queue[FetchBlockInfo]]()
+ deferredBlocks.foreach { case (blockManagerId, blockIds) =>
+ val blocks = blockIds.map { blockId =>
+ val (size, mapIndex) = infoMap(blockId)
+ FetchBlockInfo(BlockId(blockId), size, mapIndex)
+ }
+ newAddressToBlocks.put(blockManagerId, blocks)
}
Review Comment:
Do we need the before address if it has been migrated ?
--
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-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #42296: [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #42296:
URL: https://github.com/apache/spark/pull/42296#discussion_r1291923436
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -1288,6 +1288,30 @@ private[spark] class MapOutputTrackerWorker(conf: SparkConf) extends MapOutputTr
mapSizesByExecutorId.iter
}
+ def getMapOutputLocationWithRefresh(
+ shuffleId: Int,
+ mapId: Long,
+ prevLocation: BlockManagerId): BlockManagerId = {
+ // Try to get the cached location first in case other concurrent tasks
+ // fetched the fresh location already
+ var currentLocationOpt = getMapOutputLocation(shuffleId, mapId)
+ if (currentLocationOpt.isDefined && currentLocationOpt.get == prevLocation) {
Review Comment:
nit:
```suggestion
if (currentLocationOpt.exists(_ == prevLocation)) {
```
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -264,18 +272,22 @@ final class ShuffleBlockFetcherIterator(
case FetchBlockInfo(blockId, size, mapIndex) => (blockId.toString, (size, mapIndex))
}.toMap
val remainingBlocks = new HashSet[String]() ++= infoMap.keys
- val deferredBlocks = new ArrayBuffer[String]()
+ val deferredBlocks = new HashMap[BlockManagerId, Queue[String]]()
val blockIds = req.blocks.map(_.blockId.toString)
val address = req.address
val requestStartTime = clock.nanoTime()
@inline def enqueueDeferredFetchRequestIfNecessary(): Unit = {
if (remainingBlocks.isEmpty && deferredBlocks.nonEmpty) {
- val blocks = deferredBlocks.map { blockId =>
- val (size, mapIndex) = infoMap(blockId)
- FetchBlockInfo(BlockId(blockId), size, mapIndex)
+ val newAddressToBlocks = new HashMap[BlockManagerId, Queue[FetchBlockInfo]]()
+ deferredBlocks.foreach { case (blockManagerId, blockIds) =>
+ val blocks = blockIds.map { blockId =>
+ val (size, mapIndex) = infoMap(blockId)
+ FetchBlockInfo(BlockId(blockId), size, mapIndex)
+ }
+ newAddressToBlocks.put(blockManagerId, blocks)
}
Review Comment:
Why not something like:
```
deferredBlocks.foreach { case (blockManagerId, blockIds) =>
val blocks = blockIds.map { blockId =>
val (size, mapIndex) = infoMap(blockId)
FetchBlockInfo(BlockId(blockId), size, mapIndex)
}
results.put(DeferFetchRequestResult(FetchRequest(blockManagerId, blocks)))
}
```
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -528,6 +528,15 @@ package object config {
.bytesConf(ByteUnit.BYTE)
.createOptional
+ private[spark] val STORAGE_DECOMMISSION_SHUFFLE_REFRESH =
+ ConfigBuilder("spark.storage.decommission.shuffleBlocks.refreshLocationsEnabled")
+ .doc("If true, executors will try to refresh the cached locations for the shuffle blocks" +
+ "when fetch failures happens (and decommission shuffle block migration is enabled), " +
+ "and retry fetching when the location changes.")
+ .version("3.5.0")
Review Comment:
Change to `4.0.0`
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -1288,6 +1288,30 @@ private[spark] class MapOutputTrackerWorker(conf: SparkConf) extends MapOutputTr
mapSizesByExecutorId.iter
}
+ def getMapOutputLocationWithRefresh(
+ shuffleId: Int,
+ mapId: Long,
+ prevLocation: BlockManagerId): BlockManagerId = {
+ // Try to get the cached location first in case other concurrent tasks
+ // fetched the fresh location already
+ var currentLocationOpt = getMapOutputLocation(shuffleId, mapId)
+ if (currentLocationOpt.isDefined && currentLocationOpt.get == prevLocation) {
+ // Address in the cache unchanged. Try to clean cache and get a fresh location
+ unregisterShuffle(shuffleId)
+ currentLocationOpt = getMapOutputLocation(shuffleId, mapId)
Review Comment:
Note: we end up removing both map and merge status here - for this call second call, pass `canFetchMergeResult = true` in ` getMapOutputLocation`
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -264,18 +272,22 @@ final class ShuffleBlockFetcherIterator(
case FetchBlockInfo(blockId, size, mapIndex) => (blockId.toString, (size, mapIndex))
}.toMap
val remainingBlocks = new HashSet[String]() ++= infoMap.keys
- val deferredBlocks = new ArrayBuffer[String]()
+ val deferredBlocks = new HashMap[BlockManagerId, Queue[String]]()
Review Comment:
nit:
```suggestion
val deferredBlocks = new HashMap[BlockManagerId, ArrayBuffer[String]]()
```
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -1288,6 +1288,30 @@ private[spark] class MapOutputTrackerWorker(conf: SparkConf) extends MapOutputTr
mapSizesByExecutorId.iter
}
+ def getMapOutputLocationWithRefresh(
+ shuffleId: Int,
+ mapId: Long,
+ prevLocation: BlockManagerId): BlockManagerId = {
+ // Try to get the cached location first in case other concurrent tasks
+ // fetched the fresh location already
+ var currentLocationOpt = getMapOutputLocation(shuffleId, mapId)
+ if (currentLocationOpt.isDefined && currentLocationOpt.get == prevLocation) {
+ // Address in the cache unchanged. Try to clean cache and get a fresh location
+ unregisterShuffle(shuffleId)
+ currentLocationOpt = getMapOutputLocation(shuffleId, mapId)
+ }
+ if (currentLocationOpt.isEmpty) {
+ throw new MetadataFetchFailedException(shuffleId, -1,
+ message = s"Failed to get map output location for shuffleId $shuffleId, mapId $mapId")
+ }
+ currentLocationOpt.get
Review Comment:
nit: `currentLocationOpt.getOrElse( throw ... )`
--
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-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a diff in pull request #42296: [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #42296:
URL: https://github.com/apache/spark/pull/42296#discussion_r1321386699
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -1580,7 +1625,9 @@ object ShuffleBlockFetcherIterator {
* Result of a fetch request that should be deferred for some reasons, e.g., Netty OOM
Review Comment:
nit:
```suggestion
* Result of a fetch request that should be deferred for some reasons, e.g., Netty OOM, decommission
```
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -1167,8 +1200,10 @@ final class ShuffleBlockFetcherIterator(
// immediately, defer the request until the next time it can be processed.
// Process any outstanding deferred fetch requests if possible.
+ // Skip when the address is the local BM Id (this may happen when shuffle migration is enabled)
if (deferredFetchRequests.nonEmpty) {
- for ((remoteAddress, defReqQueue) <- deferredFetchRequests) {
+ for ((remoteAddress, defReqQueue) <- deferredFetchRequests
+ if remoteAddress != blockManager.blockManagerId) {
Review Comment:
nit 2 indents:
```suggestion
if remoteAddress != blockManager.blockManagerId) {
```
--
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-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on pull request #42296: [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #42296:
URL: https://github.com/apache/spark/pull/42296#issuecomment-1718716659
@bozhang2820, the test failure might be resolved by updating you branch to the latest master.
--
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-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions [spark]
Posted by "ukby1234 (via GitHub)" <gi...@apache.org>.
ukby1234 commented on code in PR #42296:
URL: https://github.com/apache/spark/pull/42296#discussion_r1366453322
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -111,6 +111,14 @@ final class ShuffleBlockFetcherIterator(
// nodes, rather than blocking on reading output from one node.
private val targetRemoteRequestSize = math.max(maxBytesInFlight / 5, 1L)
+ private val isShuffleMigrationEnabled =
+ SparkEnv.get.conf.get(config.DECOMMISSION_ENABLED) &&
+ SparkEnv.get.conf.get(config.STORAGE_DECOMMISSION_ENABLED) &&
+ SparkEnv.get.conf.get(config.STORAGE_DECOMMISSION_SHUFFLE_BLOCKS_ENABLED)
+
+ private val shouldPerformShuffleLocationRefresh =
Review Comment:
what about make this one of the constructor argument? One of the benefit is that you don't need to write tests with `TestUtils.withConf`
--
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-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions [spark]
Posted by "ukby1234 (via GitHub)" <gi...@apache.org>.
ukby1234 commented on PR #42296:
URL: https://github.com/apache/spark/pull/42296#issuecomment-1772062940
@bozhang2820 not sure if you still have time to work on this PR. I opened [another PR](https://github.com/apache/spark/pull/43443) to address some issues mentioned above.
--
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-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ukby1234 commented on a diff in pull request #42296: [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions
Posted by "ukby1234 (via GitHub)" <gi...@apache.org>.
ukby1234 commented on code in PR #42296:
URL: https://github.com/apache/spark/pull/42296#discussion_r1292923823
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -1288,6 +1288,30 @@ private[spark] class MapOutputTrackerWorker(conf: SparkConf) extends MapOutputTr
mapSizesByExecutorId.iter
}
+ def getMapOutputLocationWithRefresh(
Review Comment:
Maybe return `Option[BlockManagerId]`?:
```
def getMapOutputLocationWithRefresh(
shuffleId: Int,
mapId: Long,
prevLocation: BlockManagerId): Option[BlockManagerId] = {
// Try to get the cached location first in case other concurrent tasks
// fetched the fresh location already
getMapOutputLocation(shuffleId, mapId) match {
case Some(location) =>
if (location == prevLocation) {
unregisterShuffle(shuffleId)
getMapOutputLocation(shuffleId, mapId)
} else {
None
}
case _ =>
None
}
}
```
--
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-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions [spark]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #42296:
URL: https://github.com/apache/spark/pull/42296#issuecomment-1772716786
Do you have details that can be shared @bozhang2820 ? Thanks
--
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-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #42296: [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #42296:
URL: https://github.com/apache/spark/pull/42296#discussion_r1292611462
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -1288,6 +1288,30 @@ private[spark] class MapOutputTrackerWorker(conf: SparkConf) extends MapOutputTr
mapSizesByExecutorId.iter
}
+ def getMapOutputLocationWithRefresh(
+ shuffleId: Int,
+ mapId: Long,
+ prevLocation: BlockManagerId): BlockManagerId = {
+ // Try to get the cached location first in case other concurrent tasks
+ // fetched the fresh location already
+ var currentLocationOpt = getMapOutputLocation(shuffleId, mapId)
+ if (currentLocationOpt.isDefined && currentLocationOpt.get == prevLocation) {
+ // Address in the cache unchanged. Try to clean cache and get a fresh location
+ unregisterShuffle(shuffleId)
+ currentLocationOpt = getMapOutputLocation(shuffleId, mapId)
+ }
+ if (currentLocationOpt.isEmpty) {
+ throw new MetadataFetchFailedException(shuffleId, -1,
+ message = s"Failed to get map output location for shuffleId $shuffleId, mapId $mapId")
+ }
+ currentLocationOpt.get
Review Comment:
Let us filter it out here, and add support for fetching from fallback in a separate pr.
+CC @dongjoon-hyun as well.
--
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-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ukby1234 commented on a diff in pull request #42296: [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions
Posted by "ukby1234 (via GitHub)" <gi...@apache.org>.
ukby1234 commented on code in PR #42296:
URL: https://github.com/apache/spark/pull/42296#discussion_r1323912669
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -1288,6 +1288,30 @@ private[spark] class MapOutputTrackerWorker(conf: SparkConf) extends MapOutputTr
mapSizesByExecutorId.iter
}
+ def getMapOutputLocationWithRefresh(
Review Comment:
We can do the following with `Option`:
```
val currentAddressOpt = mapOutputTrackerWorker
.getMapOutputLocationWithRefresh(shuffleId, mapId, address)
currentAddressOpt match {
case Some(currentAddress) =>
if (currentAddress != address) {
logInfo(s"Map status location for block $blockId changed from $address " +
s"to $currentAddress")
remainingBlocks -= blockId
deferredBlocks.getOrElseUpdate(currentAddress, new ArrayBuffer[String]())
.append(blockId)
enqueueDeferredFetchRequestIfNecessary()
} else {
results.put(FailureFetchResult(block, infoMap(blockId)._2, address, e))
}
case None =>
results.put(FailureFetchResult(block, infoMap(blockId)._2, address, e))
}
```
It is also consistent with other function signatures with `getMapOutputLocation`.
--
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-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] melihsozdinler commented on a diff in pull request #42296: [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions
Posted by "melihsozdinler (via GitHub)" <gi...@apache.org>.
melihsozdinler commented on code in PR #42296:
URL: https://github.com/apache/spark/pull/42296#discussion_r1297700192
##########
core/src/main/scala/org/apache/spark/TestUtils.scala:
##########
@@ -491,6 +491,27 @@ private[spark] object TestUtils {
EnumSet.of(OWNER_READ, OWNER_EXECUTE, OWNER_WRITE))
file.getPath
}
+
+ def withConf[T](confPairs: (String, String)*)(f: => T): T = {
Review Comment:
It is nice to add some comments for the usage of this function, we see that some other public functions had before in this file. Better to follow the same.
--
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-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on pull request #42296: [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on PR #42296:
URL: https://github.com/apache/spark/pull/42296#issuecomment-1736614122
@bozhang2820 Could you rebase the PR?
--
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-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] bozhang2820 commented on pull request #42296: [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions
Posted by "bozhang2820 (via GitHub)" <gi...@apache.org>.
bozhang2820 commented on PR #42296:
URL: https://github.com/apache/spark/pull/42296#issuecomment-1709333763
Also CC @jiangxb1987
--
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-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions [spark]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #42296:
URL: https://github.com/apache/spark/pull/42296#issuecomment-2081261651
We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!
--
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-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions [spark]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #42296: [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions
URL: https://github.com/apache/spark/pull/42296
--
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-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a diff in pull request #42296: [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #42296:
URL: https://github.com/apache/spark/pull/42296#discussion_r1282588775
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -1288,6 +1288,30 @@ private[spark] class MapOutputTrackerWorker(conf: SparkConf) extends MapOutputTr
mapSizesByExecutorId.iter
}
+ def getMapOutputLocationWithRefresh(
+ shuffleId: Int,
+ mapId: Long,
+ prevLocation: BlockManagerId): BlockManagerId = {
+ // Try to get the cached location first in case other concurrent tasks
+ // fetched the fresh location already
+ var currentLocationOpt = getMapOutputLocation(shuffleId, mapId)
+ if (currentLocationOpt.isDefined && currentLocationOpt.get == prevLocation) {
+ // Address in the cache unchanged. Try to clean cache and get a fresh location
+ unregisterShuffle(shuffleId)
+ currentLocationOpt = getMapOutputLocation(shuffleId, mapId)
+ }
+ if (currentLocationOpt.isEmpty) {
+ throw new MetadataUpdateFailedException(shuffleId, mapId,
Review Comment:
Could you reuse `MetadataFetchFailedException`? We can use the `message` field to distinguish the error case.
--
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-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #42296: [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #42296:
URL: https://github.com/apache/spark/pull/42296#discussion_r1318086203
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -264,18 +272,22 @@ final class ShuffleBlockFetcherIterator(
case FetchBlockInfo(blockId, size, mapIndex) => (blockId.toString, (size, mapIndex))
}.toMap
val remainingBlocks = new HashSet[String]() ++= infoMap.keys
- val deferredBlocks = new ArrayBuffer[String]()
+ val deferredBlocks = new HashMap[BlockManagerId, Queue[String]]()
val blockIds = req.blocks.map(_.blockId.toString)
val address = req.address
val requestStartTime = clock.nanoTime()
@inline def enqueueDeferredFetchRequestIfNecessary(): Unit = {
if (remainingBlocks.isEmpty && deferredBlocks.nonEmpty) {
- val blocks = deferredBlocks.map { blockId =>
- val (size, mapIndex) = infoMap(blockId)
- FetchBlockInfo(BlockId(blockId), size, mapIndex)
+ val newAddressToBlocks = new HashMap[BlockManagerId, Queue[FetchBlockInfo]]()
+ deferredBlocks.foreach { case (blockManagerId, blockIds) =>
+ val blocks = blockIds.map { blockId =>
+ val (size, mapIndex) = infoMap(blockId)
+ FetchBlockInfo(BlockId(blockId), size, mapIndex)
+ }
+ newAddressToBlocks.put(blockManagerId, blocks)
}
Review Comment:
Why do we need the before address if it has been migrated ?
--
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-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] bozhang2820 commented on a diff in pull request #42296: [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions
Posted by "bozhang2820 (via GitHub)" <gi...@apache.org>.
bozhang2820 commented on code in PR #42296:
URL: https://github.com/apache/spark/pull/42296#discussion_r1317184031
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -264,18 +272,22 @@ final class ShuffleBlockFetcherIterator(
case FetchBlockInfo(blockId, size, mapIndex) => (blockId.toString, (size, mapIndex))
}.toMap
val remainingBlocks = new HashSet[String]() ++= infoMap.keys
- val deferredBlocks = new ArrayBuffer[String]()
+ val deferredBlocks = new HashMap[BlockManagerId, Queue[String]]()
val blockIds = req.blocks.map(_.blockId.toString)
val address = req.address
val requestStartTime = clock.nanoTime()
@inline def enqueueDeferredFetchRequestIfNecessary(): Unit = {
if (remainingBlocks.isEmpty && deferredBlocks.nonEmpty) {
- val blocks = deferredBlocks.map { blockId =>
- val (size, mapIndex) = infoMap(blockId)
- FetchBlockInfo(BlockId(blockId), size, mapIndex)
+ val newAddressToBlocks = new HashMap[BlockManagerId, Queue[FetchBlockInfo]]()
+ deferredBlocks.foreach { case (blockManagerId, blockIds) =>
+ val blocks = blockIds.map { blockId =>
+ val (size, mapIndex) = infoMap(blockId)
+ FetchBlockInfo(BlockId(blockId), size, mapIndex)
+ }
+ newAddressToBlocks.put(blockManagerId, blocks)
}
Review Comment:
The original definition for `DeferFetchRequestResult` has only one address, and here we changed it to contain both the addresses before and after the refresh (due to shuffle block migrations).
--
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-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #42296: [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #42296:
URL: https://github.com/apache/spark/pull/42296#discussion_r1325310110
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -355,8 +367,28 @@ final class ShuffleBlockFetcherIterator(
updateMergedReqsDuration(wasReqForMergedChunks = true)
results.put(FallbackOnPushMergedFailureResult(
block, address, infoMap(blockId)._1, remainingBlocks.isEmpty))
- } else {
+ } else if (!shouldPerformShuffleLocationRefresh) {
results.put(FailureFetchResult(block, infoMap(blockId)._2, address, e))
+ } else {
+ val (shuffleId, mapId) = BlockId.getShuffleIdAndMapId(block)
Review Comment:
Should we move the `getShuffleIdAndMapId` into the `Try` ?
We will effectively block shuffle indefinitely in case `getShuffleIdAndMapId` throws an exception (it should not currently - but code could evolve).
Something like:
```
Try {
val (shuffleId, mapId) = BlockId.getShuffleIdAndMapId(block)
mapOutputTrackerWorker
.getMapOutputLocationWithRefresh(shuffleId, mapId, address)
} match {
```
##########
core/src/main/scala/org/apache/spark/TestUtils.scala:
##########
@@ -491,6 +491,28 @@ private[spark] object TestUtils {
EnumSet.of(OWNER_READ, OWNER_EXECUTE, OWNER_WRITE))
file.getPath
}
+
+ /** Sets all configs specified in `confPairs`, calls `f`, and then restores them. */
+ def withConf[T](confPairs: (String, String)*)(f: => T): T = {
+ val conf = SparkEnv.get.conf
+ val (keys, values) = confPairs.unzip
+ val currentValues = keys.map { key =>
+ if (conf.contains(key)) {
+ Some(conf.get(key))
+ } else {
+ None
+ }
+ }
+ (keys, values).zipped.foreach { (key, value) =>
+ conf.set(key, value)
+ }
+ try f finally {
+ keys.zip(currentValues).foreach {
+ case (key, Some(value)) => conf.set(key, value)
+ case (key, None) => conf.remove(key)
+ }
+ }
+ }
Review Comment:
nit:
```suggestion
def withConf[T](confPairs: (String, String)*)(f: => T): T = {
val conf = SparkEnv.get.conf
val inputConfMap = confPairs.toMap
val modifiedValues = conf.getAll.filter(kv => inputConfMap.contains(kv._1)).toMap
inputConfMap.foreach { kv =>
conf.set(kv._1, kv._2)
}
try f finally {
inputConfMap.keys.foreach { key =>
if (modifiedValues.contains(key)) {
conf.set(key, modifiedValues(key))
} else {
conf.remove(key)
}
}
}
}
```
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -1288,6 +1288,32 @@ private[spark] class MapOutputTrackerWorker(conf: SparkConf) extends MapOutputTr
mapSizesByExecutorId.iter
}
+ def getMapOutputLocationWithRefresh(
+ shuffleId: Int,
+ mapId: Long,
+ prevLocation: BlockManagerId): BlockManagerId = {
+ // Try to get the cached location first in case other concurrent tasks
+ // fetched the fresh location already
+ var currentLocationOpt = getMapOutputLocation(shuffleId, mapId)
+ if (currentLocationOpt.contains(prevLocation)) {
+ // Address in the cache unchanged. Try to clean cache and get a fresh location
+ unregisterShuffle(shuffleId)
+ currentLocationOpt = getMapOutputLocation(shuffleId, mapId, canFetchMergeResult = true)
Review Comment:
```suggestion
currentLocationOpt = getMapOutputLocation(shuffleId, mapId, fetchMergeResult)
```
##########
core/src/test/scala/org/apache/spark/storage/ShuffleBlockFetcherIteratorSuite.scala:
##########
@@ -664,6 +667,107 @@ class ShuffleBlockFetcherIteratorSuite extends SparkFunSuite with PrivateMethodT
verify(blocks(ShuffleBlockId(0, 2, 0)), times(0)).release()
}
+ test("handle map output location change") {
+ TestUtils.withConf(
+ "spark.decommission.enabled" -> "true",
+ "spark.storage.decommission.enabled" -> "true",
+ "spark.storage.decommission.shuffleBlocks.enabled" -> "true"
+ ) {
+ val blockManager = createMockBlockManager()
+ val localBmId = blockManager.blockManagerId
+ val remoteBmId = BlockManagerId("test-remote-1", "test-remote-1", 2)
+ val blocks = Map[BlockId, ManagedBuffer](
+ ShuffleBlockId(0, 0, 0) -> createMockManagedBuffer(),
+ ShuffleBlockId(0, 1, 0) -> createMockManagedBuffer(),
+ ShuffleBlockId(0, 2, 0) -> createMockManagedBuffer()
+ )
+
+ doReturn(blocks(ShuffleBlockId(0, 2, 0))).when(blockManager)
+ .getLocalBlockData(meq(ShuffleBlockId(0, 2, 0)))
+
+ answerFetchBlocks { invocation =>
+ val host = invocation.getArgument[String](0)
+ val listener = invocation.getArgument[BlockFetchingListener](4)
+ host match {
+ case "test-remote-1" =>
+ listener.onBlockFetchSuccess(
+ ShuffleBlockId(0, 0, 0).toString, blocks(ShuffleBlockId(0, 0, 0)))
+ // TODO: update exception type here
+ listener.onBlockFetchFailure(
+ ShuffleBlockId(0, 1, 0).toString, new RuntimeException())
+ listener.onBlockFetchFailure(
+ ShuffleBlockId(0, 2, 0).toString, new RuntimeException())
+ case "test-remote-2" =>
+ listener.onBlockFetchSuccess(
+ ShuffleBlockId(0, 1, 0).toString, blocks(ShuffleBlockId(0, 1, 0)))
+ case "test-local-host" =>
+ listener.onBlockFetchSuccess(
+ ShuffleBlockId(0, 2, 0).toString, blocks(ShuffleBlockId(0, 2, 0)))
+ }
+ }
+
+ when(mapOutputTracker.getMapOutputLocationWithRefresh(any(), any(), any()))
+ .thenAnswer { invocation =>
+ val mapId = invocation.getArgument[Long](1)
+ mapId match {
+ case 0 => BlockManagerId("test-remote-1", "test-remote-1", 2)
+ case 1 => BlockManagerId("test-remote-2", "test-remote-2", 2)
+ case 2 => localBmId
+ }
+ }
+
+ Seq(true, false).foreach { isEnabled =>
+ SparkEnv.get.conf.set(
+ "spark.storage.decommission.shuffleBlocks.refreshLocationsEnabled", isEnabled.toString)
+ val iterator = createShuffleBlockIteratorWithDefaults(
+ Map(remoteBmId -> toBlockList(blocks.keys, 1L, 0)),
+ blockManager = Some(blockManager)
+ )
+ if (isEnabled) {
+ assert(iterator.map(_._1).toSet == blocks.keys)
+ val bytesInFlightAccess = PrivateMethod[Long](Symbol("bytesInFlight"))
+ assert(iterator.invokePrivate(bytesInFlightAccess()) == 0)
+ } else {
+ intercept[FetchFailedException] {
+ iterator.toList
+ }
+ }
+ }
+ }
+ }
+
+
+ test("metadata fetch failure in handle map output location change") {
+ TestUtils.withConf(
+ "spark.decommission.enabled" -> "true",
+ "spark.storage.decommission.enabled" -> "true",
+ "spark.storage.decommission.shuffleBlocks.enabled" -> "true",
+ "spark.storage.decommission.shuffleBlocks.refreshLocationsEnabled" -> "true"
+ ) {
+ val remoteBmId = BlockManagerId("test-remote-1", "test-remote-1", 2)
+ val blocks = Map[BlockId, ManagedBuffer](
+ ShuffleBlockId(0, 0, 0) -> createMockManagedBuffer()
+ )
+ answerFetchBlocks { invocation =>
+ val host = invocation.getArgument[String](0)
+ val listener = invocation.getArgument[BlockFetchingListener](4)
+ host match {
+ case "test-remote-1" =>
+ listener.onBlockFetchFailure(
+ ShuffleBlockId(0, 0, 0).toString, new RuntimeException())
+ }
+ }
+ when(mapOutputTracker.getMapOutputLocationWithRefresh(any(), any(), any()))
+ .thenAnswer(_ => throw new MetadataFetchFailedException(0, 0, ""))
Review Comment:
super nit: set `mapId` to `-1`
##########
core/src/test/scala/org/apache/spark/storage/ShuffleBlockFetcherIteratorSuite.scala:
##########
@@ -664,6 +667,107 @@ class ShuffleBlockFetcherIteratorSuite extends SparkFunSuite with PrivateMethodT
verify(blocks(ShuffleBlockId(0, 2, 0)), times(0)).release()
}
+ test("handle map output location change") {
+ TestUtils.withConf(
+ "spark.decommission.enabled" -> "true",
+ "spark.storage.decommission.enabled" -> "true",
+ "spark.storage.decommission.shuffleBlocks.enabled" -> "true"
+ ) {
+ val blockManager = createMockBlockManager()
+ val localBmId = blockManager.blockManagerId
+ val remoteBmId = BlockManagerId("test-remote-1", "test-remote-1", 2)
+ val blocks = Map[BlockId, ManagedBuffer](
+ ShuffleBlockId(0, 0, 0) -> createMockManagedBuffer(),
+ ShuffleBlockId(0, 1, 0) -> createMockManagedBuffer(),
+ ShuffleBlockId(0, 2, 0) -> createMockManagedBuffer()
+ )
+
+ doReturn(blocks(ShuffleBlockId(0, 2, 0))).when(blockManager)
+ .getLocalBlockData(meq(ShuffleBlockId(0, 2, 0)))
+
+ answerFetchBlocks { invocation =>
+ val host = invocation.getArgument[String](0)
+ val listener = invocation.getArgument[BlockFetchingListener](4)
+ host match {
+ case "test-remote-1" =>
+ listener.onBlockFetchSuccess(
+ ShuffleBlockId(0, 0, 0).toString, blocks(ShuffleBlockId(0, 0, 0)))
+ // TODO: update exception type here
+ listener.onBlockFetchFailure(
+ ShuffleBlockId(0, 1, 0).toString, new RuntimeException())
+ listener.onBlockFetchFailure(
+ ShuffleBlockId(0, 2, 0).toString, new RuntimeException())
+ case "test-remote-2" =>
+ listener.onBlockFetchSuccess(
+ ShuffleBlockId(0, 1, 0).toString, blocks(ShuffleBlockId(0, 1, 0)))
+ case "test-local-host" =>
+ listener.onBlockFetchSuccess(
+ ShuffleBlockId(0, 2, 0).toString, blocks(ShuffleBlockId(0, 2, 0)))
+ }
+ }
+
+ when(mapOutputTracker.getMapOutputLocationWithRefresh(any(), any(), any()))
+ .thenAnswer { invocation =>
+ val mapId = invocation.getArgument[Long](1)
+ mapId match {
+ case 0 => BlockManagerId("test-remote-1", "test-remote-1", 2)
+ case 1 => BlockManagerId("test-remote-2", "test-remote-2", 2)
+ case 2 => localBmId
+ }
+ }
+
+ Seq(true, false).foreach { isEnabled =>
+ SparkEnv.get.conf.set(
+ "spark.storage.decommission.shuffleBlocks.refreshLocationsEnabled", isEnabled.toString)
+ val iterator = createShuffleBlockIteratorWithDefaults(
+ Map(remoteBmId -> toBlockList(blocks.keys, 1L, 0)),
+ blockManager = Some(blockManager)
+ )
+ if (isEnabled) {
+ assert(iterator.map(_._1).toSet == blocks.keys)
+ val bytesInFlightAccess = PrivateMethod[Long](Symbol("bytesInFlight"))
+ assert(iterator.invokePrivate(bytesInFlightAccess()) == 0)
+ } else {
+ intercept[FetchFailedException] {
+ iterator.toList
+ }
+ }
+ }
+ }
+ }
+
+
+ test("metadata fetch failure in handle map output location change") {
Review Comment:
Can you also add a simple test to ensure existing behavior is preserved when no migration has happened ?
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -264,18 +272,22 @@ final class ShuffleBlockFetcherIterator(
case FetchBlockInfo(blockId, size, mapIndex) => (blockId.toString, (size, mapIndex))
}.toMap
val remainingBlocks = new HashSet[String]() ++= infoMap.keys
- val deferredBlocks = new ArrayBuffer[String]()
+ val deferredBlocks = new HashMap[BlockManagerId, Queue[String]]()
val blockIds = req.blocks.map(_.blockId.toString)
val address = req.address
val requestStartTime = clock.nanoTime()
@inline def enqueueDeferredFetchRequestIfNecessary(): Unit = {
if (remainingBlocks.isEmpty && deferredBlocks.nonEmpty) {
- val blocks = deferredBlocks.map { blockId =>
- val (size, mapIndex) = infoMap(blockId)
- FetchBlockInfo(BlockId(blockId), size, mapIndex)
+ val newAddressToBlocks = new HashMap[BlockManagerId, Queue[FetchBlockInfo]]()
+ deferredBlocks.foreach { case (blockManagerId, blockIds) =>
+ val blocks = blockIds.map { blockId =>
+ val (size, mapIndex) = infoMap(blockId)
+ FetchBlockInfo(BlockId(blockId), size, mapIndex)
+ }
+ newAddressToBlocks.put(blockManagerId, blocks)
}
Review Comment:
Scratch this - required for decrementing accounting details.
--
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-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions [spark]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #42296:
URL: https://github.com/apache/spark/pull/42296#issuecomment-1898002147
Any updates on this @bozhang2820 ? Thanks
--
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-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ukby1234 commented on a diff in pull request #42296: [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions
Posted by "ukby1234 (via GitHub)" <gi...@apache.org>.
ukby1234 commented on code in PR #42296:
URL: https://github.com/apache/spark/pull/42296#discussion_r1292923823
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -1288,6 +1288,30 @@ private[spark] class MapOutputTrackerWorker(conf: SparkConf) extends MapOutputTr
mapSizesByExecutorId.iter
}
+ def getMapOutputLocationWithRefresh(
Review Comment:
Maybe return `Option[BlockManagerId]`?:
```
def getMapOutputLocationWithRefresh(
shuffleId: Int,
mapId: Long,
prevLocation: BlockManagerId): Option[BlockManagerId] = {
// Try to get the cached location first in case other concurrent tasks
// fetched the fresh location already
getMapOutputLocation(shuffleId, mapId) match {
case Some(location) =>
if (location == prevLocation) {
unregisterShuffle(shuffleId)
getMapOutputLocation(shuffleId, mapId)
} else {
Some(location)
}
case _ =>
None
}
}
```
--
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-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #42296: [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #42296:
URL: https://github.com/apache/spark/pull/42296#discussion_r1292611462
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -1288,6 +1288,30 @@ private[spark] class MapOutputTrackerWorker(conf: SparkConf) extends MapOutputTr
mapSizesByExecutorId.iter
}
+ def getMapOutputLocationWithRefresh(
+ shuffleId: Int,
+ mapId: Long,
+ prevLocation: BlockManagerId): BlockManagerId = {
+ // Try to get the cached location first in case other concurrent tasks
+ // fetched the fresh location already
+ var currentLocationOpt = getMapOutputLocation(shuffleId, mapId)
+ if (currentLocationOpt.isDefined && currentLocationOpt.get == prevLocation) {
+ // Address in the cache unchanged. Try to clean cache and get a fresh location
+ unregisterShuffle(shuffleId)
+ currentLocationOpt = getMapOutputLocation(shuffleId, mapId)
+ }
+ if (currentLocationOpt.isEmpty) {
+ throw new MetadataFetchFailedException(shuffleId, -1,
+ message = s"Failed to get map output location for shuffleId $shuffleId, mapId $mapId")
+ }
+ currentLocationOpt.get
Review Comment:
Let us filter it out here, and add support for fetching from fallback in a separate pr.
--
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-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] bozhang2820 commented on a diff in pull request #42296: [SPARK-44635][CORE] Handle shuffle fetch failures in decommissions
Posted by "bozhang2820 (via GitHub)" <gi...@apache.org>.
bozhang2820 commented on code in PR #42296:
URL: https://github.com/apache/spark/pull/42296#discussion_r1311172320
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -1288,6 +1288,30 @@ private[spark] class MapOutputTrackerWorker(conf: SparkConf) extends MapOutputTr
mapSizesByExecutorId.iter
}
+ def getMapOutputLocationWithRefresh(
+ shuffleId: Int,
+ mapId: Long,
+ prevLocation: BlockManagerId): BlockManagerId = {
+ // Try to get the cached location first in case other concurrent tasks
+ // fetched the fresh location already
+ var currentLocationOpt = getMapOutputLocation(shuffleId, mapId)
+ if (currentLocationOpt.isDefined && currentLocationOpt.get == prevLocation) {
+ // Address in the cache unchanged. Try to clean cache and get a fresh location
+ unregisterShuffle(shuffleId)
+ currentLocationOpt = getMapOutputLocation(shuffleId, mapId)
Review Comment:
Good catch. Will do.
##########
core/src/main/scala/org/apache/spark/TestUtils.scala:
##########
@@ -491,6 +491,27 @@ private[spark] object TestUtils {
EnumSet.of(OWNER_READ, OWNER_EXECUTE, OWNER_WRITE))
file.getPath
}
+
+ def withConf[T](confPairs: (String, String)*)(f: => T): T = {
Review Comment:
Sure. Will do.
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -1288,6 +1288,30 @@ private[spark] class MapOutputTrackerWorker(conf: SparkConf) extends MapOutputTr
mapSizesByExecutorId.iter
}
+ def getMapOutputLocationWithRefresh(
+ shuffleId: Int,
+ mapId: Long,
+ prevLocation: BlockManagerId): BlockManagerId = {
+ // Try to get the cached location first in case other concurrent tasks
+ // fetched the fresh location already
+ var currentLocationOpt = getMapOutputLocation(shuffleId, mapId)
+ if (currentLocationOpt.isDefined && currentLocationOpt.get == prevLocation) {
Review Comment:
Will change to `currentLocationOpt.contains(prevLocation)`.
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -1288,6 +1288,30 @@ private[spark] class MapOutputTrackerWorker(conf: SparkConf) extends MapOutputTr
mapSizesByExecutorId.iter
}
+ def getMapOutputLocationWithRefresh(
Review Comment:
We still want to throw a `MetadataFetchFailedException` when failing to get a refreshed location here. So I would prefer returning a `BlockManagerId` and make it specific.
--
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-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org