You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "ivoson (via GitHub)" <gi...@apache.org> on 2025/06/09 10:55:42 UTC
[PR] [WIP][SPARK-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
ivoson opened a new pull request, #51131:
URL: https://github.com/apache/spark/pull/51131
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as new features, bug fixes, or other behavior changes. Documentation-only updates are not considered user-facing changes.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for 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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on code in PR #51131:
URL: https://github.com/apache/spark/pull/51131#discussion_r2144683867
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -1800,6 +1800,16 @@ package object config {
.booleanConf
.createWithDefault(true)
+ private[spark] val SHUFFLE_FAST_FAIL_ON_FETCH_FAILURE_ENABLED =
+ ConfigBuilder("spark.shuffle.fastFailOnFetchFailure")
+ .doc("If enabled FailureFetchResult will be added to the head of the result queue in " +
+ "ShuffleBlockFetcherIterator and fail the task as soon as the failure result is " +
+ "consumed. Otherwise, will have to process all the successfully fetched blocks in " +
Review Comment:
```suggestion
"consumed. Otherwise, the iterator must first process all the successfully fetched blocks that are ahead in the queue before failing the task " +
```
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on code in PR #51131:
URL: https://github.com/apache/spark/pull/51131#discussion_r2144679825
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -1800,6 +1800,16 @@ package object config {
.booleanConf
.createWithDefault(true)
+ private[spark] val SHUFFLE_FAST_FAIL_ON_FETCH_FAILURE_ENABLED =
+ ConfigBuilder("spark.shuffle.fastFailOnFetchFailure")
+ .doc("If enabled FailureFetchResult will be added to the head of the result queue in " +
Review Comment:
```suggestion
.doc("If enabled, FailureFetchResult will be added to the head of the result queue in " +
```
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on code in PR #51131:
URL: https://github.com/apache/spark/pull/51131#discussion_r2158148888
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -102,7 +102,7 @@ final class ShuffleBlockFetcherIterator(
checksumAlgorithm: String,
shuffleMetrics: ShuffleReadMetricsReporter,
doBatchFetch: Boolean,
- clock: Clock = new SystemClock())
+ clock: Clock = new SystemClock())
Review Comment:
Although there's a bit of an issue with this indentation, perhaps there's no need to make changes.
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #51131:
URL: https://github.com/apache/spark/pull/51131#discussion_r2158167928
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -1261,6 +1262,10 @@ final class ShuffleBlockFetcherIterator(
}
}
+ private[this] def addFailureFetchResult(result: FailureFetchResult): Unit = {
Review Comment:
removed the method
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -102,7 +102,7 @@ final class ShuffleBlockFetcherIterator(
checksumAlgorithm: String,
shuffleMetrics: ShuffleReadMetricsReporter,
doBatchFetch: Boolean,
- clock: Clock = new SystemClock())
+ clock: Clock = new SystemClock())
Review Comment:
reverted
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on PR #51131:
URL: https://github.com/apache/spark/pull/51131#issuecomment-2998793854
Thanks for your review @LuciferYang @jiangxb1987 @Ngone51 @attilapiros @mridulm
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #51131:
URL: https://github.com/apache/spark/pull/51131#discussion_r2151706348
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -1261,6 +1271,15 @@ final class ShuffleBlockFetcherIterator(
}
}
+ private[this] def addFailureFetchResult(result: FailureFetchResult): Unit = {
+ (failFastOnFetchFailureEnabled, results) match {
+ case (true, deque: LinkedBlockingDeque[FetchResult]) =>
+ deque.putFirst(result)
+ case _ =>
+ results.put(result)
Review Comment:
nit:
```suggestion
results match {
case deque: LinkedBlockingDeque[FetchResult] =>
deque.putFirst(result)
case _ =>
results.put(result)
```
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on PR #51131:
URL: https://github.com/apache/spark/pull/51131#issuecomment-2988201068
We already has too many config parameters and I cannot see any risk here.
@LuciferYang, @jiangxb1987 WDYT?
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson closed pull request #51131: [SPARK-52395][CORE] Fast fail when shuffle fetch failure happens
URL: https://github.com/apache/spark/pull/51131
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on code in PR #51131:
URL: https://github.com/apache/spark/pull/51131#discussion_r2158148011
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -1261,6 +1262,10 @@ final class ShuffleBlockFetcherIterator(
}
}
+ private[this] def addFailureFetchResult(result: FailureFetchResult): Unit = {
Review Comment:
Is it still necessary to extract a method?
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on code in PR #51131:
URL: https://github.com/apache/spark/pull/51131#discussion_r2159233978
##########
core/src/test/scala/org/apache/spark/storage/ShuffleBlockFetcherIteratorSuite.scala:
##########
@@ -1968,4 +1973,148 @@ class ShuffleBlockFetcherIteratorSuite extends SparkFunSuite {
"shuffle checksum support for ShuffleBlockBatchId")))
}
}
+
+ test("SPARK-52395: Fast fail when fetch failure happens for local blocks") {
+ val blockManager = createMockBlockManager()
+ val localBmId = blockManager.blockManagerId
+
+ // Make sure blockManager.getBlockData would return the blocks
+ val localBlocks = Map[BlockId, ManagedBuffer](
+ ShuffleBlockId(0, 0, 0) -> createMockManagedBuffer(),
+ ShuffleBlockId(0, 1, 0) -> createMockManagedBuffer(),
+ ShuffleBlockId(0, 2, 0) -> createMockManagedBuffer())
+ localBlocks.take(2).foreach { case (blockId, buf) =>
+ doReturn(buf).when(blockManager).getLocalBlockData(meq(blockId))
+ }
+ localBlocks.takeRight(1).foreach { case (blockId, buf) =>
+ doThrow(new RuntimeException("force fail")).when(blockManager)
+ .getLocalBlockData(meq(blockId))
+ }
+
+ val iterator = createShuffleBlockIteratorWithDefaults(
+ Map(localBmId -> toBlockList(localBlocks.keys, 1L, 0)),
+ blockManager = Some(blockManager)
+ )
+
+ // Fetch failure should be placed in the head of results, exception should be thrown for the
+ // 1st instance.
+ intercept[FetchFailedException] { iterator.next() }
+ assert(iterator.next()._1 === ShuffleBlockId(0, 0, 0))
+ assert(iterator.next()._1 === ShuffleBlockId(0, 1, 0))
+ assert(!iterator.hasNext)
+ }
+
+ test("SPARK-52395: Fast fail when fetch failure happens for host local blocks") {
+ val blockManager = createMockBlockManager()
+ // Create a block manager running on the same host (host-local)
+ val hostLocalBmId = BlockManagerId("test-host-local-client-1", "test-local-host", 3)
+ val hostLocalBlocks = 0.to(2).map(ShuffleBlockId(0, _, 0) -> createMockManagedBuffer()).toMap
+
+ hostLocalBlocks.take(2).foreach { case (blockId, buf) =>
+ doReturn(buf).when(blockManager)
+ .getHostLocalShuffleData(meq(blockId), any())
+ }
+ hostLocalBlocks.takeRight(1).foreach { case (blockId, buf) =>
+ doThrow(new RuntimeException("force fail")).when(blockManager)
+ .getHostLocalShuffleData(meq(blockId), any())
+ }
+ val hostLocalDirs = Map("test-host-local-client-1" -> Array("local-dir"))
+ // returning local dir for hostLocalBmId
+ initHostLocalDirManager(blockManager, hostLocalDirs)
+
+ val iterator = createShuffleBlockIteratorWithDefaults(
+ Map(hostLocalBmId -> toBlockList(hostLocalBlocks.keys, 1L, 0)),
+ blockManager = Some(blockManager)
+ )
+
+ // Fetch failure should be placed in the head of results, exception should be thrown for the
+ // 1st instance.
+ intercept[FetchFailedException] { iterator.next() }
+ assert(iterator.next()._1 === ShuffleBlockId(0, 0, 0))
+ assert(iterator.next()._1 === ShuffleBlockId(0, 1, 0))
+ assert(!iterator.hasNext)
+ }
+
+ test("SPARK-52395: Fast fail when failed to get host local dirs") {
+ val blockManager = createMockBlockManager()
+ val localBmId = blockManager.blockManagerId
+
+ // Make sure blockManager.getBlockData would return the blocks
+ val localBlocks = Map[BlockId, ManagedBuffer](
+ ShuffleBlockId(0, 0, 0) -> createMockManagedBuffer(),
+ ShuffleBlockId(0, 1, 0) -> createMockManagedBuffer())
+ localBlocks.foreach { case (blockId, buf) =>
+ doReturn(buf).when(blockManager).getLocalBlockData(meq(blockId))
+ }
+
+ // Create a block manager running on the same host (host-local)
+ val hostLocalBmId = BlockManagerId("test-host-local-client-1", "test-local-host", 3)
+ val hostLocalBlocks = Map[BlockId, ManagedBuffer](
+ ShuffleBlockId(0, 2, 0) -> createMockManagedBuffer())
+ hostLocalBlocks.foreach { case (blockId, buf) =>
+ doReturn(buf).when(blockManager)
+ .getHostLocalShuffleData(meq(blockId), any())
+ }
+
+ // Setting hostLocalDirs as null to force fail `getHostLocalDirs`.
+ initHostLocalDirManager(blockManager, null)
+
+ val iterator = createShuffleBlockIteratorWithDefaults(
+ Map(
+ localBmId -> toBlockList(localBlocks.keys, 1L, 0),
+ hostLocalBmId -> toBlockList(hostLocalBlocks.keys, 1L, 1)
+ ),
+ blockManager = Some(blockManager)
+ )
+
+ verify(blockManager, times(0)).getHostLocalShuffleData(any(), any())
+
+ // Fetch failure should be placed in the head of results, exception should be thrown for the
+ // 1st instance.
+ intercept[FetchFailedException] { iterator.next() }
+ assert(iterator.next()._1 === ShuffleBlockId(0, 0, 0))
+ assert(iterator.next()._1 === ShuffleBlockId(0, 1, 0))
+ assert(!iterator.hasNext)
+ }
+
+ test("SPARK-52395: Fail fast when fetch failure happens for remote blocks") {
Review Comment:
Do you think this test is still needed?
It is very close to the test "fail all blocks if any of the remote request fails".
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on PR #51131:
URL: https://github.com/apache/spark/pull/51131#issuecomment-2969050023
cc @mridulm @attilapiros
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #51131:
URL: https://github.com/apache/spark/pull/51131#discussion_r2151134709
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -1800,6 +1800,17 @@ package object config {
.booleanConf
.createWithDefault(true)
+ private[spark] val SHUFFLE_FAST_FAIL_ON_FETCH_FAILURE_ENABLED =
+ ConfigBuilder("spark.shuffle.fastFailOnFetchFailure")
+ .doc("If enabled, FailureFetchResult will be added to the head of the result queue in " +
+ "ShuffleBlockFetcherIterator and fail the task as soon as the failure result is " +
+ "consumed. Otherwise, the iterator must first process all the successfully fetched " +
+ "blocks that are ahead in the queue before failing the task front of the result " +
+ "queue and then fail the task.")
+ .version("4.1.0")
+ .booleanConf
+ .createWithDefault(false)
Review Comment:
done
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on PR #51131:
URL: https://github.com/apache/spark/pull/51131#issuecomment-2982695034
> > Use LinkedBlockingDeque instead of LinkedBlockingQueue for the results.
>
> Another possible solution to consider would be collecting the `FailureFetchResult` in a separate collection (which can be `LinkedBlockingQueue`) and use this new collection first at `next()` method.
>
> WDYT?
Thanks @attilapiros
I've thought about this, one concern about having 2 collections is that it may lead to some race condition issues such as:
1. Checking failure results -> Empty
2. Failure added
3. Take results from the success queue which might never have new elements added.
Dealing with the race conditions might introduce some complexity here, so this PR proposed to use `LinkedBlockingDeque` instead.
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on code in PR #51131:
URL: https://github.com/apache/spark/pull/51131#discussion_r2155205418
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -615,7 +624,8 @@ final class ShuffleBlockFetcherIterator(
case e: Exception =>
// If we see an exception, stop immediately.
logError(s"Error occurred while fetching local blocks", e)
- results.put(FailureFetchResult(blockId, mapIndex, blockManagerId, e))
+ addFailureFetchResult(FailureFetchResult(blockId, mapIndex, blockManagerId, e))
+
Review Comment:
Nit: this extra line is not needed
##########
core/src/test/scala/org/apache/spark/storage/ShuffleBlockFetcherIteratorSuite.scala:
##########
@@ -136,8 +136,13 @@ class ShuffleBlockFetcherIteratorSuite extends SparkFunSuite {
when(mockExternalBlockStoreClient.getHostLocalDirs(any(), any(), any(), any()))
.thenAnswer { invocation =>
import scala.jdk.CollectionConverters._
- invocation.getArgument[CompletableFuture[java.util.Map[String, Array[String]]]](3)
- .complete(hostLocalDirs.asJava)
+ if (hostLocalDirs != null) {
Review Comment:
Nit: I would prefer avoiding unnecessary negate operators so:
```
if (hostLocalDirs == null) {
invocation.getArgument[CompletableFuture[java.util.Map[String, Array[String]]]](3)
.completeExceptionally(new RuntimeException("force fail"))
} else {
invocation.getArgument[CompletableFuture[java.util.Map[String, Array[String]]]](3)
.complete(hostLocalDirs.asJava)
}
```
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on code in PR #51131:
URL: https://github.com/apache/spark/pull/51131#discussion_r2144691631
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -1261,6 +1265,14 @@ final class ShuffleBlockFetcherIterator(
}
}
+ private[this] def addFailureFetchResult(result: FailureFetchResult): Unit = {
+ if (fastFailOnFetchFailureEnabled) {
+ results.putFirst(result)
+ } else {
+ results.put(result)
Review Comment:
Would using `putLast` in the demonstration better highlight the differences?
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #51131:
URL: https://github.com/apache/spark/pull/51131#discussion_r2151430945
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -102,7 +103,8 @@ final class ShuffleBlockFetcherIterator(
checksumAlgorithm: String,
shuffleMetrics: ShuffleReadMetricsReporter,
doBatchFetch: Boolean,
- clock: Clock = new SystemClock())
+ fastFailOnFetchFailureEnabled: Boolean = false,
Review Comment:
done
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -132,7 +134,7 @@ final class ShuffleBlockFetcherIterator(
* A queue to hold our results. This turns the asynchronous model provided by
* [[org.apache.spark.network.BlockTransferService]] into a synchronous model (iterator).
*/
- private[this] val results = new LinkedBlockingQueue[FetchResult]
+ private[this] val results = new LinkedBlockingDeque[FetchResult]
Review Comment:
Sure, it make sense to fallback both the behavior and data structure with feature flag is disabled. updated.
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on code in PR #51131:
URL: https://github.com/apache/spark/pull/51131#discussion_r2157076614
##########
core/src/test/scala/org/apache/spark/storage/ShuffleBlockFetcherIteratorSuite.scala:
##########
@@ -136,8 +136,13 @@ class ShuffleBlockFetcherIteratorSuite extends SparkFunSuite {
when(mockExternalBlockStoreClient.getHostLocalDirs(any(), any(), any(), any()))
.thenAnswer { invocation =>
import scala.jdk.CollectionConverters._
- invocation.getArgument[CompletableFuture[java.util.Map[String, Array[String]]]](3)
- .complete(hostLocalDirs.asJava)
+ if (hostLocalDirs != null) {
Review Comment:
Readability and cognitive load.
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang closed pull request #51131: [SPARK-52395][CORE] Fast fail when shuffle fetch failure happens
URL: https://github.com/apache/spark/pull/51131
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #51131:
URL: https://github.com/apache/spark/pull/51131#discussion_r2156163003
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -615,7 +624,8 @@ final class ShuffleBlockFetcherIterator(
case e: Exception =>
// If we see an exception, stop immediately.
logError(s"Error occurred while fetching local blocks", e)
- results.put(FailureFetchResult(blockId, mapIndex, blockManagerId, e))
+ addFailureFetchResult(FailureFetchResult(blockId, mapIndex, blockManagerId, e))
+
Review Comment:
thanks, extra line removed.
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on code in PR #51131:
URL: https://github.com/apache/spark/pull/51131#discussion_r2159233978
##########
core/src/test/scala/org/apache/spark/storage/ShuffleBlockFetcherIteratorSuite.scala:
##########
@@ -1968,4 +1973,148 @@ class ShuffleBlockFetcherIteratorSuite extends SparkFunSuite {
"shuffle checksum support for ShuffleBlockBatchId")))
}
}
+
+ test("SPARK-52395: Fast fail when fetch failure happens for local blocks") {
+ val blockManager = createMockBlockManager()
+ val localBmId = blockManager.blockManagerId
+
+ // Make sure blockManager.getBlockData would return the blocks
+ val localBlocks = Map[BlockId, ManagedBuffer](
+ ShuffleBlockId(0, 0, 0) -> createMockManagedBuffer(),
+ ShuffleBlockId(0, 1, 0) -> createMockManagedBuffer(),
+ ShuffleBlockId(0, 2, 0) -> createMockManagedBuffer())
+ localBlocks.take(2).foreach { case (blockId, buf) =>
+ doReturn(buf).when(blockManager).getLocalBlockData(meq(blockId))
+ }
+ localBlocks.takeRight(1).foreach { case (blockId, buf) =>
+ doThrow(new RuntimeException("force fail")).when(blockManager)
+ .getLocalBlockData(meq(blockId))
+ }
+
+ val iterator = createShuffleBlockIteratorWithDefaults(
+ Map(localBmId -> toBlockList(localBlocks.keys, 1L, 0)),
+ blockManager = Some(blockManager)
+ )
+
+ // Fetch failure should be placed in the head of results, exception should be thrown for the
+ // 1st instance.
+ intercept[FetchFailedException] { iterator.next() }
+ assert(iterator.next()._1 === ShuffleBlockId(0, 0, 0))
+ assert(iterator.next()._1 === ShuffleBlockId(0, 1, 0))
+ assert(!iterator.hasNext)
+ }
+
+ test("SPARK-52395: Fast fail when fetch failure happens for host local blocks") {
+ val blockManager = createMockBlockManager()
+ // Create a block manager running on the same host (host-local)
+ val hostLocalBmId = BlockManagerId("test-host-local-client-1", "test-local-host", 3)
+ val hostLocalBlocks = 0.to(2).map(ShuffleBlockId(0, _, 0) -> createMockManagedBuffer()).toMap
+
+ hostLocalBlocks.take(2).foreach { case (blockId, buf) =>
+ doReturn(buf).when(blockManager)
+ .getHostLocalShuffleData(meq(blockId), any())
+ }
+ hostLocalBlocks.takeRight(1).foreach { case (blockId, buf) =>
+ doThrow(new RuntimeException("force fail")).when(blockManager)
+ .getHostLocalShuffleData(meq(blockId), any())
+ }
+ val hostLocalDirs = Map("test-host-local-client-1" -> Array("local-dir"))
+ // returning local dir for hostLocalBmId
+ initHostLocalDirManager(blockManager, hostLocalDirs)
+
+ val iterator = createShuffleBlockIteratorWithDefaults(
+ Map(hostLocalBmId -> toBlockList(hostLocalBlocks.keys, 1L, 0)),
+ blockManager = Some(blockManager)
+ )
+
+ // Fetch failure should be placed in the head of results, exception should be thrown for the
+ // 1st instance.
+ intercept[FetchFailedException] { iterator.next() }
+ assert(iterator.next()._1 === ShuffleBlockId(0, 0, 0))
+ assert(iterator.next()._1 === ShuffleBlockId(0, 1, 0))
+ assert(!iterator.hasNext)
+ }
+
+ test("SPARK-52395: Fast fail when failed to get host local dirs") {
+ val blockManager = createMockBlockManager()
+ val localBmId = blockManager.blockManagerId
+
+ // Make sure blockManager.getBlockData would return the blocks
+ val localBlocks = Map[BlockId, ManagedBuffer](
+ ShuffleBlockId(0, 0, 0) -> createMockManagedBuffer(),
+ ShuffleBlockId(0, 1, 0) -> createMockManagedBuffer())
+ localBlocks.foreach { case (blockId, buf) =>
+ doReturn(buf).when(blockManager).getLocalBlockData(meq(blockId))
+ }
+
+ // Create a block manager running on the same host (host-local)
+ val hostLocalBmId = BlockManagerId("test-host-local-client-1", "test-local-host", 3)
+ val hostLocalBlocks = Map[BlockId, ManagedBuffer](
+ ShuffleBlockId(0, 2, 0) -> createMockManagedBuffer())
+ hostLocalBlocks.foreach { case (blockId, buf) =>
+ doReturn(buf).when(blockManager)
+ .getHostLocalShuffleData(meq(blockId), any())
+ }
+
+ // Setting hostLocalDirs as null to force fail `getHostLocalDirs`.
+ initHostLocalDirManager(blockManager, null)
+
+ val iterator = createShuffleBlockIteratorWithDefaults(
+ Map(
+ localBmId -> toBlockList(localBlocks.keys, 1L, 0),
+ hostLocalBmId -> toBlockList(hostLocalBlocks.keys, 1L, 1)
+ ),
+ blockManager = Some(blockManager)
+ )
+
+ verify(blockManager, times(0)).getHostLocalShuffleData(any(), any())
+
+ // Fetch failure should be placed in the head of results, exception should be thrown for the
+ // 1st instance.
+ intercept[FetchFailedException] { iterator.next() }
+ assert(iterator.next()._1 === ShuffleBlockId(0, 0, 0))
+ assert(iterator.next()._1 === ShuffleBlockId(0, 1, 0))
+ assert(!iterator.hasNext)
+ }
+
+ test("SPARK-52395: Fail fast when fetch failure happens for remote blocks") {
Review Comment:
Do you think this test is still needed?
It is very close to the test "fail all blocks if any of the remote request fails".
And please check the other newly added tests too whether they are still relevant!
--
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] [WIP][SPARK-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson closed pull request #51131: [WIP][SPARK-52395][CORE] Fast fail when shuffle fetch failure happens
URL: https://github.com/apache/spark/pull/51131
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #51131:
URL: https://github.com/apache/spark/pull/51131#discussion_r2162734701
##########
core/src/test/scala/org/apache/spark/storage/ShuffleBlockFetcherIteratorSuite.scala:
##########
@@ -1968,4 +1973,148 @@ class ShuffleBlockFetcherIteratorSuite extends SparkFunSuite {
"shuffle checksum support for ShuffleBlockBatchId")))
}
}
+
+ test("SPARK-52395: Fast fail when fetch failure happens for local blocks") {
+ val blockManager = createMockBlockManager()
+ val localBmId = blockManager.blockManagerId
+
+ // Make sure blockManager.getBlockData would return the blocks
+ val localBlocks = Map[BlockId, ManagedBuffer](
+ ShuffleBlockId(0, 0, 0) -> createMockManagedBuffer(),
+ ShuffleBlockId(0, 1, 0) -> createMockManagedBuffer(),
+ ShuffleBlockId(0, 2, 0) -> createMockManagedBuffer())
+ localBlocks.take(2).foreach { case (blockId, buf) =>
+ doReturn(buf).when(blockManager).getLocalBlockData(meq(blockId))
+ }
+ localBlocks.takeRight(1).foreach { case (blockId, buf) =>
+ doThrow(new RuntimeException("force fail")).when(blockManager)
+ .getLocalBlockData(meq(blockId))
+ }
+
+ val iterator = createShuffleBlockIteratorWithDefaults(
+ Map(localBmId -> toBlockList(localBlocks.keys, 1L, 0)),
+ blockManager = Some(blockManager)
+ )
+
+ // Fetch failure should be placed in the head of results, exception should be thrown for the
+ // 1st instance.
+ intercept[FetchFailedException] { iterator.next() }
+ assert(iterator.next()._1 === ShuffleBlockId(0, 0, 0))
+ assert(iterator.next()._1 === ShuffleBlockId(0, 1, 0))
+ assert(!iterator.hasNext)
+ }
+
+ test("SPARK-52395: Fast fail when fetch failure happens for host local blocks") {
+ val blockManager = createMockBlockManager()
+ // Create a block manager running on the same host (host-local)
+ val hostLocalBmId = BlockManagerId("test-host-local-client-1", "test-local-host", 3)
+ val hostLocalBlocks = 0.to(2).map(ShuffleBlockId(0, _, 0) -> createMockManagedBuffer()).toMap
+
+ hostLocalBlocks.take(2).foreach { case (blockId, buf) =>
+ doReturn(buf).when(blockManager)
+ .getHostLocalShuffleData(meq(blockId), any())
+ }
+ hostLocalBlocks.takeRight(1).foreach { case (blockId, buf) =>
+ doThrow(new RuntimeException("force fail")).when(blockManager)
+ .getHostLocalShuffleData(meq(blockId), any())
+ }
+ val hostLocalDirs = Map("test-host-local-client-1" -> Array("local-dir"))
+ // returning local dir for hostLocalBmId
+ initHostLocalDirManager(blockManager, hostLocalDirs)
+
+ val iterator = createShuffleBlockIteratorWithDefaults(
+ Map(hostLocalBmId -> toBlockList(hostLocalBlocks.keys, 1L, 0)),
+ blockManager = Some(blockManager)
+ )
+
+ // Fetch failure should be placed in the head of results, exception should be thrown for the
+ // 1st instance.
+ intercept[FetchFailedException] { iterator.next() }
+ assert(iterator.next()._1 === ShuffleBlockId(0, 0, 0))
+ assert(iterator.next()._1 === ShuffleBlockId(0, 1, 0))
+ assert(!iterator.hasNext)
+ }
+
+ test("SPARK-52395: Fast fail when failed to get host local dirs") {
+ val blockManager = createMockBlockManager()
+ val localBmId = blockManager.blockManagerId
+
+ // Make sure blockManager.getBlockData would return the blocks
+ val localBlocks = Map[BlockId, ManagedBuffer](
+ ShuffleBlockId(0, 0, 0) -> createMockManagedBuffer(),
+ ShuffleBlockId(0, 1, 0) -> createMockManagedBuffer())
+ localBlocks.foreach { case (blockId, buf) =>
+ doReturn(buf).when(blockManager).getLocalBlockData(meq(blockId))
+ }
+
+ // Create a block manager running on the same host (host-local)
+ val hostLocalBmId = BlockManagerId("test-host-local-client-1", "test-local-host", 3)
+ val hostLocalBlocks = Map[BlockId, ManagedBuffer](
+ ShuffleBlockId(0, 2, 0) -> createMockManagedBuffer())
+ hostLocalBlocks.foreach { case (blockId, buf) =>
+ doReturn(buf).when(blockManager)
+ .getHostLocalShuffleData(meq(blockId), any())
+ }
+
+ // Setting hostLocalDirs as null to force fail `getHostLocalDirs`.
+ initHostLocalDirManager(blockManager, null)
+
+ val iterator = createShuffleBlockIteratorWithDefaults(
+ Map(
+ localBmId -> toBlockList(localBlocks.keys, 1L, 0),
+ hostLocalBmId -> toBlockList(hostLocalBlocks.keys, 1L, 1)
+ ),
+ blockManager = Some(blockManager)
+ )
+
+ verify(blockManager, times(0)).getHostLocalShuffleData(any(), any())
+
+ // Fetch failure should be placed in the head of results, exception should be thrown for the
+ // 1st instance.
+ intercept[FetchFailedException] { iterator.next() }
+ assert(iterator.next()._1 === ShuffleBlockId(0, 0, 0))
+ assert(iterator.next()._1 === ShuffleBlockId(0, 1, 0))
+ assert(!iterator.hasNext)
+ }
+
+ test("SPARK-52395: Fail fast when fetch failure happens for remote blocks") {
Review Comment:
Sure, done.
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #51131:
URL: https://github.com/apache/spark/pull/51131#discussion_r2151131495
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -1800,6 +1800,17 @@ package object config {
.booleanConf
.createWithDefault(true)
+ private[spark] val SHUFFLE_FAST_FAIL_ON_FETCH_FAILURE_ENABLED =
+ ConfigBuilder("spark.shuffle.fastFailOnFetchFailure")
+ .doc("If enabled, FailureFetchResult will be added to the head of the result queue in " +
+ "ShuffleBlockFetcherIterator and fail the task as soon as the failure result is " +
+ "consumed. Otherwise, the iterator must first process all the successfully fetched " +
+ "blocks that are ahead in the queue before failing the task front of the result " +
+ "queue and then fail the task.")
+ .version("4.1.0")
+ .booleanConf
+ .createWithDefault(false)
Review Comment:
Sure, let's do it as it's an improvement with low risk.
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #51131:
URL: https://github.com/apache/spark/pull/51131#discussion_r2151207968
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -132,7 +134,7 @@ final class ShuffleBlockFetcherIterator(
* A queue to hold our results. This turns the asynchronous model provided by
* [[org.apache.spark.network.BlockTransferService]] into a synchronous model (iterator).
*/
- private[this] val results = new LinkedBlockingQueue[FetchResult]
+ private[this] val results = new LinkedBlockingDeque[FetchResult]
Review Comment:
Shall we only use `LinkedBlockingDeque` when `fastFailOnFetchFailureEnabled=true`?
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on PR #51131:
URL: https://github.com/apache/spark/pull/51131#issuecomment-2985149588
> Dealing with the race conditions might introduce some complexity here
Right, with two collections this would be lot more complex.
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #51131:
URL: https://github.com/apache/spark/pull/51131#discussion_r2156164697
##########
core/src/test/scala/org/apache/spark/storage/ShuffleBlockFetcherIteratorSuite.scala:
##########
@@ -136,8 +136,13 @@ class ShuffleBlockFetcherIteratorSuite extends SparkFunSuite {
when(mockExternalBlockStoreClient.getHostLocalDirs(any(), any(), any(), any()))
.thenAnswer { invocation =>
import scala.jdk.CollectionConverters._
- invocation.getArgument[CompletableFuture[java.util.Map[String, Array[String]]]](3)
- .complete(hostLocalDirs.asJava)
+ if (hostLocalDirs != null) {
Review Comment:
Sure, updated. Would like to know why is the preference? Can you please share? 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
Re: [PR] [SPARK-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #51131:
URL: https://github.com/apache/spark/pull/51131#discussion_r2160629584
##########
core/src/test/scala/org/apache/spark/storage/ShuffleBlockFetcherIteratorSuite.scala:
##########
@@ -1968,4 +1973,148 @@ class ShuffleBlockFetcherIteratorSuite extends SparkFunSuite {
"shuffle checksum support for ShuffleBlockBatchId")))
}
}
+
+ test("SPARK-52395: Fast fail when fetch failure happens for local blocks") {
+ val blockManager = createMockBlockManager()
+ val localBmId = blockManager.blockManagerId
+
+ // Make sure blockManager.getBlockData would return the blocks
+ val localBlocks = Map[BlockId, ManagedBuffer](
+ ShuffleBlockId(0, 0, 0) -> createMockManagedBuffer(),
+ ShuffleBlockId(0, 1, 0) -> createMockManagedBuffer(),
+ ShuffleBlockId(0, 2, 0) -> createMockManagedBuffer())
+ localBlocks.take(2).foreach { case (blockId, buf) =>
+ doReturn(buf).when(blockManager).getLocalBlockData(meq(blockId))
+ }
+ localBlocks.takeRight(1).foreach { case (blockId, buf) =>
+ doThrow(new RuntimeException("force fail")).when(blockManager)
+ .getLocalBlockData(meq(blockId))
+ }
+
+ val iterator = createShuffleBlockIteratorWithDefaults(
+ Map(localBmId -> toBlockList(localBlocks.keys, 1L, 0)),
+ blockManager = Some(blockManager)
+ )
+
+ // Fetch failure should be placed in the head of results, exception should be thrown for the
+ // 1st instance.
+ intercept[FetchFailedException] { iterator.next() }
+ assert(iterator.next()._1 === ShuffleBlockId(0, 0, 0))
+ assert(iterator.next()._1 === ShuffleBlockId(0, 1, 0))
+ assert(!iterator.hasNext)
+ }
+
+ test("SPARK-52395: Fast fail when fetch failure happens for host local blocks") {
+ val blockManager = createMockBlockManager()
+ // Create a block manager running on the same host (host-local)
+ val hostLocalBmId = BlockManagerId("test-host-local-client-1", "test-local-host", 3)
+ val hostLocalBlocks = 0.to(2).map(ShuffleBlockId(0, _, 0) -> createMockManagedBuffer()).toMap
+
+ hostLocalBlocks.take(2).foreach { case (blockId, buf) =>
+ doReturn(buf).when(blockManager)
+ .getHostLocalShuffleData(meq(blockId), any())
+ }
+ hostLocalBlocks.takeRight(1).foreach { case (blockId, buf) =>
+ doThrow(new RuntimeException("force fail")).when(blockManager)
+ .getHostLocalShuffleData(meq(blockId), any())
+ }
+ val hostLocalDirs = Map("test-host-local-client-1" -> Array("local-dir"))
+ // returning local dir for hostLocalBmId
+ initHostLocalDirManager(blockManager, hostLocalDirs)
+
+ val iterator = createShuffleBlockIteratorWithDefaults(
+ Map(hostLocalBmId -> toBlockList(hostLocalBlocks.keys, 1L, 0)),
+ blockManager = Some(blockManager)
+ )
+
+ // Fetch failure should be placed in the head of results, exception should be thrown for the
+ // 1st instance.
+ intercept[FetchFailedException] { iterator.next() }
+ assert(iterator.next()._1 === ShuffleBlockId(0, 0, 0))
+ assert(iterator.next()._1 === ShuffleBlockId(0, 1, 0))
+ assert(!iterator.hasNext)
+ }
+
+ test("SPARK-52395: Fast fail when failed to get host local dirs") {
+ val blockManager = createMockBlockManager()
+ val localBmId = blockManager.blockManagerId
+
+ // Make sure blockManager.getBlockData would return the blocks
+ val localBlocks = Map[BlockId, ManagedBuffer](
+ ShuffleBlockId(0, 0, 0) -> createMockManagedBuffer(),
+ ShuffleBlockId(0, 1, 0) -> createMockManagedBuffer())
+ localBlocks.foreach { case (blockId, buf) =>
+ doReturn(buf).when(blockManager).getLocalBlockData(meq(blockId))
+ }
+
+ // Create a block manager running on the same host (host-local)
+ val hostLocalBmId = BlockManagerId("test-host-local-client-1", "test-local-host", 3)
+ val hostLocalBlocks = Map[BlockId, ManagedBuffer](
+ ShuffleBlockId(0, 2, 0) -> createMockManagedBuffer())
+ hostLocalBlocks.foreach { case (blockId, buf) =>
+ doReturn(buf).when(blockManager)
+ .getHostLocalShuffleData(meq(blockId), any())
+ }
+
+ // Setting hostLocalDirs as null to force fail `getHostLocalDirs`.
+ initHostLocalDirManager(blockManager, null)
+
+ val iterator = createShuffleBlockIteratorWithDefaults(
+ Map(
+ localBmId -> toBlockList(localBlocks.keys, 1L, 0),
+ hostLocalBmId -> toBlockList(hostLocalBlocks.keys, 1L, 1)
+ ),
+ blockManager = Some(blockManager)
+ )
+
+ verify(blockManager, times(0)).getHostLocalShuffleData(any(), any())
+
+ // Fetch failure should be placed in the head of results, exception should be thrown for the
+ // 1st instance.
+ intercept[FetchFailedException] { iterator.next() }
+ assert(iterator.next()._1 === ShuffleBlockId(0, 0, 0))
+ assert(iterator.next()._1 === ShuffleBlockId(0, 1, 0))
+ assert(!iterator.hasNext)
+ }
+
+ test("SPARK-52395: Fail fast when fetch failure happens for remote blocks") {
Review Comment:
This is the only test case which is close to an existing one.
I kept it as they were added for different purpose. Further changes may modify either of them and we may lose the test coverage. Pls let me know whether this make sense to you. Will remove it if I am overthinking about it.
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on PR #51131:
URL: https://github.com/apache/spark/pull/51131#issuecomment-2998791791
Merged into master for Apache Spark 4.1.0. Thanks @ivoson @attilapiros @Ngone51 @jiangxb1987 and @mridulm
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on PR #51131:
URL: https://github.com/apache/spark/pull/51131#issuecomment-2998656591
Hi @attilapiros pls take another look. 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
Re: [PR] [SPARK-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #51131:
URL: https://github.com/apache/spark/pull/51131#discussion_r2148936934
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -1261,6 +1265,14 @@ final class ShuffleBlockFetcherIterator(
}
}
+ private[this] def addFailureFetchResult(result: FailureFetchResult): Unit = {
+ if (fastFailOnFetchFailureEnabled) {
+ results.putFirst(result)
+ } else {
+ results.put(result)
Review Comment:
Sounds good. Changed to `putLast`.
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "jiangxb1987 (via GitHub)" <gi...@apache.org>.
jiangxb1987 commented on code in PR #51131:
URL: https://github.com/apache/spark/pull/51131#discussion_r2150897757
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -1800,6 +1800,17 @@ package object config {
.booleanConf
.createWithDefault(true)
+ private[spark] val SHUFFLE_FAST_FAIL_ON_FETCH_FAILURE_ENABLED =
+ ConfigBuilder("spark.shuffle.fastFailOnFetchFailure")
+ .doc("If enabled, FailureFetchResult will be added to the head of the result queue in " +
+ "ShuffleBlockFetcherIterator and fail the task as soon as the failure result is " +
+ "consumed. Otherwise, the iterator must first process all the successfully fetched " +
+ "blocks that are ahead in the queue before failing the task front of the result " +
+ "queue and then fail the task.")
+ .version("4.1.0")
+ .booleanConf
+ .createWithDefault(false)
Review Comment:
How about enable it by default?
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #51131:
URL: https://github.com/apache/spark/pull/51131#discussion_r2151208194
##########
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala:
##########
@@ -102,7 +103,8 @@ final class ShuffleBlockFetcherIterator(
checksumAlgorithm: String,
shuffleMetrics: ShuffleReadMetricsReporter,
doBatchFetch: Boolean,
- clock: Clock = new SystemClock())
+ fastFailOnFetchFailureEnabled: Boolean = false,
Review Comment:
nit:
```suggestion
failFastOnFetchFailureEnabled: Boolean = false,
```
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on PR #51131:
URL: https://github.com/apache/spark/pull/51131#issuecomment-2964795534
cc @Ngone51 @jiangxb1987 @cloud-fan Please take a look. 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
Re: [PR] [SPARK-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on PR #51131:
URL: https://github.com/apache/spark/pull/51131#issuecomment-2982599957
> Use LinkedBlockingDeque instead of LinkedBlockingQueue for the results.
Another possible solution to consider would be collecting the `FailureFetchResult` in a separate collection (which can be `LinkedBlockingQueue`) and use this new collection first at `next()` method.
WDYT?
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on PR #51131:
URL: https://github.com/apache/spark/pull/51131#issuecomment-2989839543
Thanks for sharing your options. Removing the feature flag. cc @attilapiros @LuciferYang @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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on PR #51131:
URL: https://github.com/apache/spark/pull/51131#issuecomment-2986710759
> Why we need the feature flag? I fail to see what we lose when this is always on but I can see the flag adds some extra complexity.
Adding feature flag is trying to keep a mitigation for any potential unforeseen issues since shuffle is critical component and we've made slightly behavior changes here.
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on PR #51131:
URL: https://github.com/apache/spark/pull/51131#issuecomment-2989636219
It indeed seems that there are no apparent risks. I agree to remove the newly added config.
--
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-52395][CORE] Fast fail when shuffle fetch failure happens [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on code in PR #51131:
URL: https://github.com/apache/spark/pull/51131#discussion_r2161840690
##########
core/src/test/scala/org/apache/spark/storage/ShuffleBlockFetcherIteratorSuite.scala:
##########
@@ -1968,4 +1973,148 @@ class ShuffleBlockFetcherIteratorSuite extends SparkFunSuite {
"shuffle checksum support for ShuffleBlockBatchId")))
}
}
+
+ test("SPARK-52395: Fast fail when fetch failure happens for local blocks") {
+ val blockManager = createMockBlockManager()
+ val localBmId = blockManager.blockManagerId
+
+ // Make sure blockManager.getBlockData would return the blocks
+ val localBlocks = Map[BlockId, ManagedBuffer](
+ ShuffleBlockId(0, 0, 0) -> createMockManagedBuffer(),
+ ShuffleBlockId(0, 1, 0) -> createMockManagedBuffer(),
+ ShuffleBlockId(0, 2, 0) -> createMockManagedBuffer())
+ localBlocks.take(2).foreach { case (blockId, buf) =>
+ doReturn(buf).when(blockManager).getLocalBlockData(meq(blockId))
+ }
+ localBlocks.takeRight(1).foreach { case (blockId, buf) =>
+ doThrow(new RuntimeException("force fail")).when(blockManager)
+ .getLocalBlockData(meq(blockId))
+ }
+
+ val iterator = createShuffleBlockIteratorWithDefaults(
+ Map(localBmId -> toBlockList(localBlocks.keys, 1L, 0)),
+ blockManager = Some(blockManager)
+ )
+
+ // Fetch failure should be placed in the head of results, exception should be thrown for the
+ // 1st instance.
+ intercept[FetchFailedException] { iterator.next() }
+ assert(iterator.next()._1 === ShuffleBlockId(0, 0, 0))
+ assert(iterator.next()._1 === ShuffleBlockId(0, 1, 0))
+ assert(!iterator.hasNext)
+ }
+
+ test("SPARK-52395: Fast fail when fetch failure happens for host local blocks") {
+ val blockManager = createMockBlockManager()
+ // Create a block manager running on the same host (host-local)
+ val hostLocalBmId = BlockManagerId("test-host-local-client-1", "test-local-host", 3)
+ val hostLocalBlocks = 0.to(2).map(ShuffleBlockId(0, _, 0) -> createMockManagedBuffer()).toMap
+
+ hostLocalBlocks.take(2).foreach { case (blockId, buf) =>
+ doReturn(buf).when(blockManager)
+ .getHostLocalShuffleData(meq(blockId), any())
+ }
+ hostLocalBlocks.takeRight(1).foreach { case (blockId, buf) =>
+ doThrow(new RuntimeException("force fail")).when(blockManager)
+ .getHostLocalShuffleData(meq(blockId), any())
+ }
+ val hostLocalDirs = Map("test-host-local-client-1" -> Array("local-dir"))
+ // returning local dir for hostLocalBmId
+ initHostLocalDirManager(blockManager, hostLocalDirs)
+
+ val iterator = createShuffleBlockIteratorWithDefaults(
+ Map(hostLocalBmId -> toBlockList(hostLocalBlocks.keys, 1L, 0)),
+ blockManager = Some(blockManager)
+ )
+
+ // Fetch failure should be placed in the head of results, exception should be thrown for the
+ // 1st instance.
+ intercept[FetchFailedException] { iterator.next() }
+ assert(iterator.next()._1 === ShuffleBlockId(0, 0, 0))
+ assert(iterator.next()._1 === ShuffleBlockId(0, 1, 0))
+ assert(!iterator.hasNext)
+ }
+
+ test("SPARK-52395: Fast fail when failed to get host local dirs") {
+ val blockManager = createMockBlockManager()
+ val localBmId = blockManager.blockManagerId
+
+ // Make sure blockManager.getBlockData would return the blocks
+ val localBlocks = Map[BlockId, ManagedBuffer](
+ ShuffleBlockId(0, 0, 0) -> createMockManagedBuffer(),
+ ShuffleBlockId(0, 1, 0) -> createMockManagedBuffer())
+ localBlocks.foreach { case (blockId, buf) =>
+ doReturn(buf).when(blockManager).getLocalBlockData(meq(blockId))
+ }
+
+ // Create a block manager running on the same host (host-local)
+ val hostLocalBmId = BlockManagerId("test-host-local-client-1", "test-local-host", 3)
+ val hostLocalBlocks = Map[BlockId, ManagedBuffer](
+ ShuffleBlockId(0, 2, 0) -> createMockManagedBuffer())
+ hostLocalBlocks.foreach { case (blockId, buf) =>
+ doReturn(buf).when(blockManager)
+ .getHostLocalShuffleData(meq(blockId), any())
+ }
+
+ // Setting hostLocalDirs as null to force fail `getHostLocalDirs`.
+ initHostLocalDirManager(blockManager, null)
+
+ val iterator = createShuffleBlockIteratorWithDefaults(
+ Map(
+ localBmId -> toBlockList(localBlocks.keys, 1L, 0),
+ hostLocalBmId -> toBlockList(hostLocalBlocks.keys, 1L, 1)
+ ),
+ blockManager = Some(blockManager)
+ )
+
+ verify(blockManager, times(0)).getHostLocalShuffleData(any(), any())
+
+ // Fetch failure should be placed in the head of results, exception should be thrown for the
+ // 1st instance.
+ intercept[FetchFailedException] { iterator.next() }
+ assert(iterator.next()._1 === ShuffleBlockId(0, 0, 0))
+ assert(iterator.next()._1 === ShuffleBlockId(0, 1, 0))
+ assert(!iterator.hasNext)
+ }
+
+ test("SPARK-52395: Fail fast when fetch failure happens for remote blocks") {
Review Comment:
It is technically the same:
```diff
1c1
< test("fail all blocks if any of the remote request fails") {
---
> test("SPARK-52395: Fail fast when fetch failure happens for remote blocks") {
21,22c21,22
< listener.onBlockFetchFailure(
< ShuffleBlockId(0, 2, 0).toString, new BlockNotFoundException("blah"))
---
> listener.onBlockFetchFailure(
> ShuffleBlockId(0, 2, 0).toString, new BlockNotFoundException("blah"))
34,35c34,36
< // The first two blocks should throw FetchFailedExceptions (due to failure), and the last
< // one should be returned without an exception
---
> // There are one success fetch and 2 failures. Since failures are consumed firstly so
> // the first 2 should throw FetchFailedExceptions (due to failure) and the last block
> // should be returned without an exception.
```
The only difference is an indentation and a comment.
So please remove the new test but fix the indentation and change the comment in the old one.
--
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