Uh oh!
There was an error while loading. Please reload this page.
[SPARK-32417] Fix flakyness of BlockManagerDecommissionIntegrationSuite - #29226
Closed
agrawaldevesh wants to merge 8 commits into
Closed
[SPARK-32417] Fix flakyness of BlockManagerDecommissionIntegrationSuite#29226agrawaldevesh wants to merge 8 commits into
agrawaldevesh wants to merge 8 commits into
Conversation
agrawaldeveshforce-pushed
the
block-manager-decom-flaky
branch
2 times, most recently
from
July 25, 2020 00:28
7659d22 to
558702eCompareagrawaldevesh
commented
Jul 25, 2020
Author
cc: @holdenk@attilapiros for review please. |
agrawaldevesh
marked this pull request as draft
July 25, 2020 02:07
holdenk
commented
Jul 25, 2020
via email
Contributor
Oh awesome, thanks for fixing this. I totally misdiagnosed this issue
(thought it was around placement of the partitions). On Fri, Jul 24, 2020 at 5:36 PM Apache Spark QA ***@***.***> wrote:
*Test build #126518 has started
<https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/126518/testReport>*
for PR 29226 at commit 558702e
<558702e>
.
—
You are receiving this because you were mentioned.
Reply to this email directly, view it on GitHub
<#29226 (comment)>, or
unsubscribe
<https://github.com/notifications/unsubscribe-auth/AAAOT5OSXJNHXNXHEYMHJALR5ISHXANCNFSM4PHFIUWQ>
.
-- Cell : 425-233-8271 |
SparkQA
commented
Jul 25, 2020
Test build #126517 has finished for PR 29226 at commit
|
SparkQA
commented
Jul 25, 2020
Test build #126518 has finished for PR 29226 at commit
|
agrawaldevesh
commented
Jul 25, 2020
Author
Sorry it didn't work. I will make another go at it. |
SparkQA
commented
Jul 25, 2020
Test build #126526 has finished for PR 29226 at commit
|
agrawaldeveshforce-pushed
the
block-manager-decom-flaky
branch
3 times, most recently
from
July 25, 2020 07:06
4e34e96 to
199ea03Compareagrawaldeveshforce-pushed
the
block-manager-decom-flaky
branch
from
July 25, 2020 07:59
199ea03 to
b2c2fadCompareSparkQA
commented
Jul 25, 2020
Test build #126530 has finished for PR 29226 at commit
|
SparkQA
commented
Jul 25, 2020
Test build #126531 has finished for PR 29226 at commit
|
SparkQA
commented
Jul 25, 2020
Test build #126536 has finished for PR 29226 at commit
|
agrawaldevesh
commented
Jul 25, 2020
Author
The remaining failure is in SparkR and is probably environment related. |
agrawaldevesh
marked this pull request as ready for review
July 25, 2020 16:38
agrawaldeveshforce-pushed
the
block-manager-decom-flaky
branch
from
July 25, 2020 17:20
b2c2fad to
ba12869CompareSparkQA
commented
Jul 25, 2020
Test build #126550 has finished for PR 29226 at commit
|
agrawaldeveshforce-pushed
the
block-manager-decom-flaky
branch
from
July 25, 2020 20:05
ba12869 to
a939c67CompareSparkQA
commented
Jul 25, 2020
Test build #126554 has finished for PR 29226 at commit
|
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
attilapiros
requested changes
Jul 26, 2020
attilapiros
left a comment
Contributor
There was a problem hiding this comment.
This PR might resolve the issue at hand but during the problem analysis it stops at a level and fails to reveal the root cause. I would like to encourage you to look a bit deeper: there might be an easier fix or a more straightforward one.
Uh oh!
There was an error while loading. Please reload this page.
agrawaldeveshforce-pushed
the
block-manager-decom-flaky
branch
from
July 26, 2020 21:16
a939c67 to
2cea5f4CompareSparkQA
commented
Jul 28, 2020
Test build #126678 has finished for PR 29226 at commit
|
SparkQA
commented
Jul 28, 2020
Test build #126684 has finished for PR 29226 at commit
|
SparkQA
commented
Jul 28, 2020
Test build #126687 has finished for PR 29226 at commit
|
SparkQA
commented
Jul 28, 2020
Test build #126681 has finished for PR 29226 at commit
|
agrawaldevesh
commented
Jul 28, 2020
Author
I think this isn't needed anymore. |
holdenk
reviewed
Jul 28, 2020
| sc.addSparkListener(new SparkListener { | ||
| def getCandidateExecutorToDecom: Option[String] = if (whenToDecom == TaskStarted) { | ||
| accum.value.asScala.headOption |
Contributor
There was a problem hiding this comment.
I don't think this is going to work as intended, accumulators send updates back at the end of the task, unless something has changed.
Contributor
There was a problem hiding this comment.
If you don't believe that is the case, can we add an assertion in here that none of the tasks have finished yet?
holdenk
commented
Jul 28, 2020
Contributor
I think I understand your intention with accumulators now, I don't think that's going to work. |
agrawaldeveshforce-pushed
the
block-manager-decom-flaky
branch
from
July 29, 2020 01:23
bea7776 to
ed8224bCompareAn interesting failure happens when migrateDuring = true (and persist or shuffle is true): - We schedule the job with tasks on executors 0, 1, 2. - We wait 300 ms and decommission executor 0. - If the task is not yet done on executor 0, it will now fail because the block manager won't be able to save the block. This condition is easy to trigger on a loaded machine where the github checks run. - The task with retry on a different executor (1 or 2) and its shuffle blocks will land there. - No actual block migration happens here because the decommissioned executor technically failed before it could even produce a block. So this change makes two fixes to remove the above race condition. - When migrateDuring = true, wait for a task to complete and write the block, and then decommission that executor. - When migrateDuring = false, it is still possible (because of delay scheduling) for two tasks to be run on the same executor serially and one executor to go idle. In which case, we must make sure to decommission an executor that actually had a task run on it.
Now that we wait for an actual task to succeed, we don't need to wait for events prior to that: broadcast of job-info finished and task started. The waiting for the task end/success subsumes that. Simplifying the test even further.
agrawaldeveshforce-pushed
the
block-manager-decom-flaky
branch
from
July 29, 2020 01:25
ed8224b to
d709002Compareagrawaldeveshforce-pushed
the
block-manager-decom-flaky
branch
from
July 29, 2020 01:29
d709002 to
1b469fbCompareagrawaldevesh
commented
Jul 29, 2020
Author
Thanks for pointing this out @holdenk. As a spark n00b I have certainly learnt quite a bit by fixing this test. I tried another approach and also added the asserts you asked for. I learnt that while spark does not update the accumulator until the task completes, it does graciously trigger a listener event when the accumulators are received from an executor via the heartbeat mechanism. I am using that to know for sure that the task has started for "real". Please take a look. |
SparkQA
commented
Jul 29, 2020
Test build #126743 has finished for PR 29226 at commit
|
SparkQA
commented
Jul 29, 2020
Test build #126744 has finished for PR 29226 at commit
|
holdenk
commented
Jul 29, 2020
Contributor
Yeah if you wait long enough for a heartbeat and check the event that should do the trick, but to avoid a flaky test your going to need a good length of time. I'll take another look tomorrow afternoon (have meetings all morning). |
holdenk
approved these changes
Jul 29, 2020
holdenk
left a comment
Contributor
There was a problem hiding this comment.
LGTM. If no other comments I'll merge this tomorrow.
Member
For some reasons, the test seems still flaky:
|
dongjoon-hyun pushed a commit
that referenced
this pull request
Aug 10, 2020
…onIntegrationSuite to reduce flakyness ### What changes were proposed in this pull request? As reported by HyukjinKwon, BlockManagerDecommissionIntegrationSuite test is apparently still flaky (even after #29226): #29226 (comment). The new flakyness is because the executors are not launching in the 6 seconds time out I had given them when run under github checks. Bumped up the timeouts. ### Why are the changes needed? To make this test not flaky so that it can give us high signal if decommissioning regresses. ### Does this PR introduce _any_ user-facing change? No, unit test only check. ### How was this patch tested? No new tests. Just github and jenkins. Closes#29388 from agrawaldevesh/more_bm_harden. Authored-by: Devesh Agrawal <devesh.agrawal@gmail.com> Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>
holdenk added a commit
to holdenk/spark
that referenced
this pull request
Oct 27, 2020
[SPARK-21040][CORE] Speculate tasks which are running on decommission executors This PR adds functionality to consider the running tasks on decommission executors based on some config. In spark-on-cloud , we sometimes already know that an executor won't be alive for more than fix amount of time. Ex- In AWS Spot nodes, once we get the notification, we know that a node will be gone in 120 seconds. So if the running tasks on the decommissioning executors may run beyond currentTime+120 seconds, then they are candidate for speculation. Currently when an executor is decommission, we stop scheduling new tasks on those executors but the already running tasks keeps on running on them. Based on the cloud, we might know beforehand that an executor won't be alive for more than a preconfigured time. Different cloud providers gives different timeouts before they take away the nodes. For Ex- In case of AWS spot nodes, an executor won't be alive for more than 120 seconds. We can utilize this information in cloud environments and take better decisions about speculating the already running tasks on decommission executors. Yes. This PR adds a new config "spark.executor.decommission.killInterval" which they can explicitly set based on the cloud environment where they are running. Added UT. Closesapache#28619 from prakharjain09/SPARK-21040-speculate-decommission-exec-tasks. Authored-by: Prakhar Jain <prakharjain09@gmail.com> Signed-off-by: Holden Karau <hkarau@apple.com> [SPARK-20629][CORE][K8S] Copy shuffle data when nodes are being shutdown This pull request adds the ability to migrate shuffle files during Spark's decommissioning. The design document associated with this change is at https://docs.google.com/document/d/1xVO1b6KAwdUhjEJBolVPl9C6sLj7oOveErwDSYdT-pE . To allow this change the `MapOutputTracker` has been extended to allow the location of shuffle files to be updated with `updateMapOutput`. When a shuffle block is put, a block update message will be sent which triggers the `updateMapOutput`. Instead of rejecting remote puts of shuffle blocks `BlockManager` delegates the storage of shuffle blocks to it's shufflemanager's resolver (if supported). A new, experimental, trait is added for shuffle resolvers to indicate they handle remote putting of blocks. The existing block migration code is moved out into a separate file, and a producer/consumer model is introduced for migrating shuffle files from the host as quickly as possible while not overwhelming other executors. Recomputting shuffle blocks can be expensive, we should take advantage of our decommissioning time to migrate these blocks. This PR introduces two new configs parameters, `spark.storage.decommission.shuffleBlocks.enabled` & `spark.storage.decommission.rddBlocks.enabled` that control which blocks should be migrated during storage decommissioning. New unit test & expansion of the Spark on K8s decom test to assert that decommisioning with shuffle block migration means that the results are not recomputed even when the original executor is terminated. This PR is a cleaned-up version of the previous WIP PR I made apache#28331 (thanks to attilapiros for his very helpful reviewing on it :)). Closesapache#28708 from holdenk/SPARK-20629-copy-shuffle-data-when-nodes-are-being-shutdown-cleaned-up. Lead-authored-by: Holden Karau <hkarau@apple.com> Co-authored-by: Holden Karau <holden@pigscanfly.ca> Co-authored-by: “attilapiros” <piros.attila.zsolt@gmail.com> Co-authored-by: Attila Zsolt Piros <attilazsoltpiros@apiros-mbp16.lan> Signed-off-by: Holden Karau <hkarau@apple.com> [SPARK-24266][K8S] Restart the watcher when we receive a version changed from k8s Restart the watcher when it failed with a HTTP_GONE code from the kubernetes api. Which means a resource version has changed. For more relevant information see here: fabric8io/kubernetes-client#1075 No Running spark-submit to a k8s cluster. Not sure how to make an automated test for this. If someone can help me out that would be great. Closesapache#28423 from stijndehaes/bugfix/k8s-submit-resource-version-change. Authored-by: Stijn De Haes <stijndehaes@gmail.com> Signed-off-by: Holden Karau <hkarau@apple.com> [SPARK-32217] Plumb whether a worker would also be decommissioned along with executor This PR is a giant plumbing PR that plumbs an `ExecutorDecommissionInfo` along with the DecommissionExecutor message. The primary motivation is to know whether a decommissioned executor would also be loosing shuffle files -- and thus it is important to know whether the host would also be decommissioned. In the absence of this PR, the existing code assumes that decommissioning an executor does not loose the whole host with it, and thus does not clear the shuffle state if external shuffle service is enabled. While this may hold in some cases (like K8s decommissioning an executor pod, or YARN container preemption), it does not hold in others like when the cluster is managed by a Standalone Scheduler (Master). This is similar to the existing `workerLost` field in the `ExecutorProcessLost` message. In the future, this `ExecutorDecommissionInfo` can be embellished for knowing how long the executor has to live for scenarios like Cloud spot kills (or Yarn preemption) and the like. No Tweaked an existing unit test in `AppClientSuite` Closesapache#29032 from agrawaldevesh/plumb_decom_info. Authored-by: Devesh Agrawal <devesh.agrawal@gmail.com> Signed-off-by: Holden Karau <hkarau@apple.com> [SPARK-32199][SPARK-32198] Reduce job failures during decommissioning This PR reduces the prospect of a job loss during decommissioning. It fixes two holes in the current decommissioning framework: - (a) Loss of decommissioned executors is not treated as a job failure: We know that the decommissioned executor would be dying soon, so its death is clearly not caused by the application. - (b) Shuffle files on the decommissioned host are cleared when the first fetch failure is detected from a decommissioned host: This is a bit tricky in terms of when to clear the shuffle state ? Ideally you want to clear it the millisecond before the shuffle service on the node dies (or the executor dies when there is no external shuffle service) -- too soon and it could lead to some wastage and too late would lead to fetch failures. The approach here is to do this clearing when the very first fetch failure is observed on the decommissioned block manager, without waiting for other blocks to also signal a failure. Without them decommissioning a lot of executors at a time leads to job failures. The task scheduler tracks the executors that were decommissioned along with their `ExecutorDecommissionInfo`. This information is used by: (a) For handling a `ExecutorProcessLost` error, or (b) by the `DAGScheduler` when handling a fetch failure. No Added a new unit test `DecommissionWorkerSuite` to test the new behavior by exercising the Master-Worker decommissioning. I chose to add a new test since the setup logic was quite different from the existing `WorkerDecommissionSuite`. I am open to changing the name of the newly added test suite :-) - Should I add a feature flag to guard these two behaviors ? They seem safe to me that they should only get triggered by decommissioning, but you never know :-). Closesapache#29014 from agrawaldevesh/decom_harden. Authored-by: Devesh Agrawal <devesh.agrawal@gmail.com> Signed-off-by: Holden Karau <hkarau@apple.com> [SPARK-32417] Fix flakyness of BlockManagerDecommissionIntegrationSuite This test tries to fix the flakyness of BlockManagerDecommissionIntegrationSuite. Make the block manager decommissioning test be less flaky An interesting failure happens when migrateDuring = true (and persist or shuffle is true): - We schedule the job with tasks on executors 0, 1, 2. - We wait 300 ms and decommission executor 0. - If the task is not yet done on executor 0, it will now fail because the block manager won't be able to save the block. This condition is easy to trigger on a loaded machine where the github checks run. - The task with retry on a different executor (1 or 2) and its shuffle blocks will land there. - No actual block migration happens here because the decommissioned executor technically failed before it could even produce a block. To remove the above race, this change replaces the fixed wait for 300 ms to wait for an actual task to succeed. When a task has succeeded, we know its blocks would have been written for sure and thus its executor would certainly be forced to migrate those blocks when it is decommissioned. The change always decommissions an executor on which a real task finished successfully instead of picking the first executor. Because the system may choose to schedule nothing on the first executor and instead run the two tasks on one executor. I have had bad luck with BlockManagerDecommissionIntegrationSuite and it has failed several times on my PRs. So fixing it. No, unit test only change. Github checks. Ran this test 100 times, 10 at a time in parallel in a script. Closesapache#29226 from agrawaldevesh/block-manager-decom-flaky. Authored-by: Devesh Agrawal <devesh.agrawal@gmail.com> Signed-off-by: Holden Karau <hkarau@apple.com> [SPARK-31197][CORE] Shutdown executor once we are done decommissioning Exit the executor when it has been asked to decommission and there is nothing left for it to do. This is a rebase of apache#28817 If we want to use decommissioning in Spark's own scale down we should terminate the executor once finished. Furthermore, in graceful shutdown it makes sense to release resources we no longer need if we've been asked to shutdown by the cluster manager instead of always holding the resources as long as possible. The decommissioned executors will exit and the end of decommissioning. This is sort of a user facing change, however decommissioning hasn't been in any releases yet. I changed the unit test to not send the executor exit message and still wait on the executor exited message. Closesapache#29211 from holdenk/SPARK-31197-exit-execs-redone. Authored-by: Holden Karau <hkarau@apple.com> Signed-off-by: Holden Karau <hkarau@apple.com> Connect decommissioning to dynamic scaling Because the mock always says there is an RDD we may replicate more than once, and now that there are independent threads Make Spark's dynamic allocation use decommissioning Track the decommissioning executors in the core dynamic scheduler so we don't scale down too low, update the streaming ExecutorAllocationManager to also delegate to decommission Fix up executor add for resource profile Fix our exiting and cleanup thread for better debugging next time. Cleanup the locks we use in decommissioning and clarify some more bits. Verify executors decommissioned, then killed by external external cluster manager are re-launched Verify some additional calls are not occuring in the executor allocation manager suite. Dont' close the watcher until the end of the test Use decommissionExecutors and set adjustTargetNumExecutors to false so that we can match the pattern for killExecutor/killExecutors bump numparts up to 6 Revert "bump numparts up to 6" This reverts commit daf96dd. Small coment & visibility cleanup CR feedback/cleanup Cleanup the merge [SPARK-21040][CORE] Speculate tasks which are running on decommission executors This PR adds functionality to consider the running tasks on decommission executors based on some config. In spark-on-cloud , we sometimes already know that an executor won't be alive for more than fix amount of time. Ex- In AWS Spot nodes, once we get the notification, we know that a node will be gone in 120 seconds. So if the running tasks on the decommissioning executors may run beyond currentTime+120 seconds, then they are candidate for speculation. Currently when an executor is decommission, we stop scheduling new tasks on those executors but the already running tasks keeps on running on them. Based on the cloud, we might know beforehand that an executor won't be alive for more than a preconfigured time. Different cloud providers gives different timeouts before they take away the nodes. For Ex- In case of AWS spot nodes, an executor won't be alive for more than 120 seconds. We can utilize this information in cloud environments and take better decisions about speculating the already running tasks on decommission executors. Yes. This PR adds a new config "spark.executor.decommission.killInterval" which they can explicitly set based on the cloud environment where they are running. Added UT. Closesapache#28619 from prakharjain09/SPARK-21040-speculate-decommission-exec-tasks. Authored-by: Prakhar Jain <prakharjain09@gmail.com> Signed-off-by: Holden Karau <hkarau@apple.com> [SPARK-20629][CORE][K8S] Copy shuffle data when nodes are being shutdown This pull request adds the ability to migrate shuffle files during Spark's decommissioning. The design document associated with this change is at https://docs.google.com/document/d/1xVO1b6KAwdUhjEJBolVPl9C6sLj7oOveErwDSYdT-pE . To allow this change the `MapOutputTracker` has been extended to allow the location of shuffle files to be updated with `updateMapOutput`. When a shuffle block is put, a block update message will be sent which triggers the `updateMapOutput`. Instead of rejecting remote puts of shuffle blocks `BlockManager` delegates the storage of shuffle blocks to it's shufflemanager's resolver (if supported). A new, experimental, trait is added for shuffle resolvers to indicate they handle remote putting of blocks. The existing block migration code is moved out into a separate file, and a producer/consumer model is introduced for migrating shuffle files from the host as quickly as possible while not overwhelming other executors. Recomputting shuffle blocks can be expensive, we should take advantage of our decommissioning time to migrate these blocks. This PR introduces two new configs parameters, `spark.storage.decommission.shuffleBlocks.enabled` & `spark.storage.decommission.rddBlocks.enabled` that control which blocks should be migrated during storage decommissioning. New unit test & expansion of the Spark on K8s decom test to assert that decommisioning with shuffle block migration means that the results are not recomputed even when the original executor is terminated. This PR is a cleaned-up version of the previous WIP PR I made apache#28331 (thanks to attilapiros for his very helpful reviewing on it :)). Closesapache#28708 from holdenk/SPARK-20629-copy-shuffle-data-when-nodes-are-being-shutdown-cleaned-up. Lead-authored-by: Holden Karau <hkarau@apple.com> Co-authored-by: Holden Karau <holden@pigscanfly.ca> Co-authored-by: “attilapiros” <piros.attila.zsolt@gmail.com> Co-authored-by: Attila Zsolt Piros <attilazsoltpiros@apiros-mbp16.lan> Signed-off-by: Holden Karau <hkarau@apple.com> [SPARK-24266][K8S] Restart the watcher when we receive a version changed from k8s Restart the watcher when it failed with a HTTP_GONE code from the kubernetes api. Which means a resource version has changed. For more relevant information see here: fabric8io/kubernetes-client#1075 No Running spark-submit to a k8s cluster. Not sure how to make an automated test for this. If someone can help me out that would be great. Closesapache#28423 from stijndehaes/bugfix/k8s-submit-resource-version-change. Authored-by: Stijn De Haes <stijndehaes@gmail.com> Signed-off-by: Holden Karau <hkarau@apple.com> [SPARK-32217] Plumb whether a worker would also be decommissioned along with executor This PR is a giant plumbing PR that plumbs an `ExecutorDecommissionInfo` along with the DecommissionExecutor message. The primary motivation is to know whether a decommissioned executor would also be loosing shuffle files -- and thus it is important to know whether the host would also be decommissioned. In the absence of this PR, the existing code assumes that decommissioning an executor does not loose the whole host with it, and thus does not clear the shuffle state if external shuffle service is enabled. While this may hold in some cases (like K8s decommissioning an executor pod, or YARN container preemption), it does not hold in others like when the cluster is managed by a Standalone Scheduler (Master). This is similar to the existing `workerLost` field in the `ExecutorProcessLost` message. In the future, this `ExecutorDecommissionInfo` can be embellished for knowing how long the executor has to live for scenarios like Cloud spot kills (or Yarn preemption) and the like. No Tweaked an existing unit test in `AppClientSuite` Closesapache#29032 from agrawaldevesh/plumb_decom_info. Authored-by: Devesh Agrawal <devesh.agrawal@gmail.com> Signed-off-by: Holden Karau <hkarau@apple.com> [SPARK-32199][SPARK-32198] Reduce job failures during decommissioning This PR reduces the prospect of a job loss during decommissioning. It fixes two holes in the current decommissioning framework: - (a) Loss of decommissioned executors is not treated as a job failure: We know that the decommissioned executor would be dying soon, so its death is clearly not caused by the application. - (b) Shuffle files on the decommissioned host are cleared when the first fetch failure is detected from a decommissioned host: This is a bit tricky in terms of when to clear the shuffle state ? Ideally you want to clear it the millisecond before the shuffle service on the node dies (or the executor dies when there is no external shuffle service) -- too soon and it could lead to some wastage and too late would lead to fetch failures. The approach here is to do this clearing when the very first fetch failure is observed on the decommissioned block manager, without waiting for other blocks to also signal a failure. Without them decommissioning a lot of executors at a time leads to job failures. The task scheduler tracks the executors that were decommissioned along with their `ExecutorDecommissionInfo`. This information is used by: (a) For handling a `ExecutorProcessLost` error, or (b) by the `DAGScheduler` when handling a fetch failure. No Added a new unit test `DecommissionWorkerSuite` to test the new behavior by exercising the Master-Worker decommissioning. I chose to add a new test since the setup logic was quite different from the existing `WorkerDecommissionSuite`. I am open to changing the name of the newly added test suite :-) - Should I add a feature flag to guard these two behaviors ? They seem safe to me that they should only get triggered by decommissioning, but you never know :-). Closesapache#29014 from agrawaldevesh/decom_harden. Authored-by: Devesh Agrawal <devesh.agrawal@gmail.com> Signed-off-by: Holden Karau <hkarau@apple.com> [SPARK-32417] Fix flakyness of BlockManagerDecommissionIntegrationSuite This test tries to fix the flakyness of BlockManagerDecommissionIntegrationSuite. Make the block manager decommissioning test be less flaky An interesting failure happens when migrateDuring = true (and persist or shuffle is true): - We schedule the job with tasks on executors 0, 1, 2. - We wait 300 ms and decommission executor 0. - If the task is not yet done on executor 0, it will now fail because the block manager won't be able to save the block. This condition is easy to trigger on a loaded machine where the github checks run. - The task with retry on a different executor (1 or 2) and its shuffle blocks will land there. - No actual block migration happens here because the decommissioned executor technically failed before it could even produce a block. To remove the above race, this change replaces the fixed wait for 300 ms to wait for an actual task to succeed. When a task has succeeded, we know its blocks would have been written for sure and thus its executor would certainly be forced to migrate those blocks when it is decommissioned. The change always decommissions an executor on which a real task finished successfully instead of picking the first executor. Because the system may choose to schedule nothing on the first executor and instead run the two tasks on one executor. I have had bad luck with BlockManagerDecommissionIntegrationSuite and it has failed several times on my PRs. So fixing it. No, unit test only change. Github checks. Ran this test 100 times, 10 at a time in parallel in a script. Closesapache#29226 from agrawaldevesh/block-manager-decom-flaky. Authored-by: Devesh Agrawal <devesh.agrawal@gmail.com> Signed-off-by: Holden Karau <hkarau@apple.com> [SPARK-31197][CORE] Shutdown executor once we are done decommissioning Exit the executor when it has been asked to decommission and there is nothing left for it to do. This is a rebase of apache#28817 If we want to use decommissioning in Spark's own scale down we should terminate the executor once finished. Furthermore, in graceful shutdown it makes sense to release resources we no longer need if we've been asked to shutdown by the cluster manager instead of always holding the resources as long as possible. The decommissioned executors will exit and the end of decommissioning. This is sort of a user facing change, however decommissioning hasn't been in any releases yet. I changed the unit test to not send the executor exit message and still wait on the executor exited message. Closesapache#29211 from holdenk/SPARK-31197-exit-execs-redone. Authored-by: Holden Karau <hkarau@apple.com> Signed-off-by: Holden Karau <hkarau@apple.com> Connect decommissioning to dynamic scaling Because the mock always says there is an RDD we may replicate more than once, and now that there are independent threads Make Spark's dynamic allocation use decommissioning Track the decommissioning executors in the core dynamic scheduler so we don't scale down too low, update the streaming ExecutorAllocationManager to also delegate to decommission Fix up executor add for resource profile Fix our exiting and cleanup thread for better debugging next time. Cleanup the locks we use in decommissioning and clarify some more bits. Verify executors decommissioned, then killed by external external cluster manager are re-launched Verify some additional calls are not occuring in the executor allocation manager suite. Dont' close the watcher until the end of the test Use decommissionExecutors and set adjustTargetNumExecutors to false so that we can match the pattern for killExecutor/killExecutors bump numparts up to 6 Revert "bump numparts up to 6" This reverts commit daf96dd. Small coment & visibility cleanup CR feedback/cleanup Fix up the merge CR feedback, move adjustExecutors to a common utility function Exclude some non-public APIs Remove junk More CR feedback Fix adjustExecutors backport This test fails for me locally and from what I recall it's because we use a different method of resolving the bind address than upstream so disabling the test This test fails for me locally and from what I recall it's because we use a different method of resolving the bind address than upstream so disabling the test Cleanup and drop watcher changes from the backport
holdenk pushed a commit
to holdenk/spark
that referenced
this pull request
Oct 27, 2020
…onIntegrationSuite to reduce flakyness ### What changes were proposed in this pull request? As reported by HyukjinKwon, BlockManagerDecommissionIntegrationSuite test is apparently still flaky (even after apache#29226): apache#29226 (comment). The new flakyness is because the executors are not launching in the 6 seconds time out I had given them when run under github checks. Bumped up the timeouts. ### Why are the changes needed? To make this test not flaky so that it can give us high signal if decommissioning regresses. ### Does this PR introduce _any_ user-facing change? No, unit test only check. ### How was this patch tested? No new tests. Just github and jenkins. Closesapache#29388 from agrawaldevesh/more_bm_harden. Authored-by: Devesh Agrawal <devesh.agrawal@gmail.com> Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for freeto join this conversation on GitHub.
Already have an account?
Sign in to comment
6 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
This test tries to fix the flakyness of BlockManagerDecommissionIntegrationSuite.
Description of the problem
Make the block manager decommissioning test be less flaky
An interesting failure happens when migrateDuring = true (and persist or shuffle is true):
the block manager won't be able to save the block. This condition is
easy to trigger on a loaded machine where the github checks run.
blocks will land there.
executor technically failed before it could even produce a block.
To remove the above race, this change replaces the fixed wait for 300 ms to wait for an actual task to succeed. When a task has succeeded, we know its blocks would have been written for sure and thus its executor would certainly be forced to migrate those blocks when it is decommissioned.
The change always decommissions an executor on which a real task finished successfully instead of picking the first executor. Because the system may choose to schedule nothing on the first executor and instead run the two tasks on one executor.
Why are the changes needed?
I have had bad luck with BlockManagerDecommissionIntegrationSuite and it has failed several times on my PRs. So fixing it.
Does this PR introduce any user-facing change?
No, unit test only change.
How was this patch tested?
Github checks. Ran this test 100 times, 10 at a time in parallel in a script.