[SPARK-32199][SPARK-32198] Reduce job failures during decommissioning by agrawaldevesh · Pull Request #29014 · apache/spark

agrawaldevesh · 2020-07-06T17:58:20Z

What changes were proposed in this pull request?

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.

Why are the changes needed?

Without them decommissioning a lot of executors at a time leads to job failures.

Code overview

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.

Does this PR introduce any user-facing change?

No

How was this patch tested?

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 :-)

Questions for reviewers

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 :-).

agrawaldevesh · 2020-07-06T18:07:09Z

@holdenk ... can you take a quick look at the approach here please. Happy to chat further offline. Also happy to fold this PR (entirely or parts of it) into your PR.

holdenk · 2020-07-06T19:23:20Z

I'd rather avoid the hack configuration flag and just go straight to extending the current message. What do you think?

holdenk · 2020-07-06T19:29:08Z

If you need the changes I've made in the other (not yet merged PRs), feel free to base your PR on top of that and just make a note that it's the case for merge ordering and so folks don't duplicate review effort :)

agrawaldevesh · 2020-07-06T19:30:26Z

I'd rather avoid the hack configuration flag and just go straight to extending the current message. What do you think?

Sure !. I think obviously that is cleaner. If you aren't working on that extension, I would be happy to take a stab at that plumbing in a separate PR. In the meanwhile, I can also just back out that config since it is only needed when there are more than one executors in a node with external shuffle service. I will amend this PR to back out that config.

agrawaldevesh · 2020-07-10T04:25:20Z

@holdenk, @jiangxb1987 @cloud-fan @Ngone51 -- This PR is ready for your review please. Thanks !

jiangxb1987

Changes look good only some nits.

agrawaldevesh · 2020-07-14T23:07:17Z

I cannot reproduce the failures in the ExecutorAllocationManagerSuite: I ran 300 back to back invocations on my local SBT/Laptop and they all passed. I suspect that the test is flaky. I cannot see any interaction with the decommissioning codepath.

cloud-fan · 2020-07-16T04:50:34Z

ok to test

SparkQA · 2020-07-16T10:06:01Z

Test build #125948 has finished for PR 29014 at commit 30af8f4.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class ExecutorDecommissionInfo(message: String, isHostDecommissioned: Boolean)
case class ExecutorProcessLost(

SparkQA · 2020-07-28T13:52:29Z

Test build #126713 has finished for PR 29014 at commit 806630a.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class ExecutorProcessLost(

attilapiros · 2020-07-28T15:02:35Z

@agrawaldevesh we are really close, sorry for being this picky (today) but I am sure this extra care will serve us well.

agrawaldevesh · 2020-07-28T17:19:51Z

@agrawaldevesh we are really close, sorry for being this picky (today) but I am sure this extra care will serve us well.

@attilapiros Always always appreciated !. Better to be nitpicky once than confuse a thousand future readers :-P

I have reworked the whole config confusion in the test. Now there are only the following configs:

Master and worker config (that is used to create the master and workers before each test)
SparkConf proper for the spark job and all of the driver/scheduler related config.
Default config used for the external shuffle service

This has really cleaned up the config mess so thanks for spotting this.

SparkQA · 2020-07-28T20:10:44Z

Test build #126729 has finished for PR 29014 at commit 153e8e4.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class ExecutorProcessLost(

attilapiros · 2020-07-28T20:16:14Z

LGTM
cc @holdenk
thanks @agrawaldevesh

attilapiros · 2020-07-28T20:21:45Z

Locally lint runs successfully:

╰─$ ./dev/lint-scala
Scalastyle checks passed.

SparkQA · 2020-07-28T20:23:56Z

Test build #126730 has finished for PR 29014 at commit 6b020bf.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class ExecutorProcessLost(

SparkQA · 2020-07-28T20:50:52Z

Test build #126731 has finished for PR 29014 at commit a906041.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class ExecutorProcessLost(

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 decom'd 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 wasteage and too late would lead to fetch failures. (These two things are a bit intertwined so it is easier to do them in a single commit) The approach here is to do this clearing when the very first fetch failure is observed on the decom'd block manager, without waiting for other blocks to also signal a failure. Added a new unit test `DecommissionWorkerSuite` to test the new behavior.

agrawaldevesh · 2020-07-29T01:26:29Z

Thanks a lot for a very very solid review and co-authoring of this PR @attilapiros. I learnt a lot.

@holdenk all yours now :-)

SparkQA · 2020-07-29T04:06:47Z

Test build #126745 has finished for PR 29014 at commit dbc478b.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class ExecutorProcessLost(

holdenk

LGTM, I'll merge this today.

HyukjinKwon · 2020-08-12T06:43:26Z

FYI, I saw the flakiness of the test org.apache.spark.deploy.DecommissionWorkerSuite.decommission workers ensure that fetch failures lead to rerun here, https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/127357/testReport/org.apache.spark.deploy/DecommissionWorkerSuite/decommission_workers_ensure_that_fetch_failures_lead_to_rerun/

org.scalatest.exceptions.TestFailedException: 1173 was not less than or equal to 1000
sbt.ForkMain$ForkError: org.scalatest.exceptions.TestFailedException: 1173 was not less than or equal to 1000
at org.scalatest.Assertions.newAssertionFailedException(Assertions.scala:472)
at org.scalatest.Assertions.newAssertionFailedException$(Assertions.scala:471)
at org.scalatest.Assertions$.newAssertionFailedException(Assertions.scala:1231)
at org.scalatest.Assertions$AssertionsHelper.macroAssert(Assertions.scala:1295)
at org.apache.spark.scheduler.BarrierTaskContextSuite.$anonfun$new$13(BarrierTaskContextSuite.scala:143)
at org.scalatest.OutcomeOf.outcomeOf(OutcomeOf.scala:85)
at org.scalatest.OutcomeOf.outcomeOf$(OutcomeOf.scala:83)
at org.scalatest.OutcomeOf$.outcomeOf(OutcomeOf.scala:104)
at org.scalatest.Transformer.apply(Transformer.scala:22)
at org.scalatest.Transformer.apply(Transformer.scala:20)
at org.scalatest.funsuite.AnyFunSuiteLike$$anon$1.apply(AnyFunSuiteLike.scala:189)
at org.apache.spark.SparkFunSuite.withFixture(SparkFunSuite.scala:164)
at org.scalatest.funsuite.AnyFunSuiteLike.invokeWithFixture$1(AnyFunSuiteLike.scala:187)
at org.scalatest.funsuite.AnyFunSuiteLike.$anonfun$runTest$1(AnyFunSuiteLike.scala:199)
at org.scalatest.SuperEngine.runTestImpl(Engine.scala:306)
at org.scalatest.funsuite.AnyFunSuiteLike.runTest(AnyFunSuiteLike.scala:199)
at org.scalatest.funsuite.AnyFunSuiteLike.runTest$(AnyFunSuiteLike.scala:181)
at org.apache.spark.SparkFunSuite.org$scalatest$BeforeAndAfterEach$$super$runTest(SparkFunSuite.scala:60)
at org.scalatest.BeforeAndAfterEach.runTest(BeforeAndAfterEach.scala:234)
at org.scalatest.BeforeAndAfterEach.runTest$(BeforeAndAfterEach.scala:227)
at org.apache.spark.SparkFunSuite.runTest(SparkFunSuite.scala:60)
at org.scalatest.funsuite.AnyFunSuiteLike.$anonfun$runTests$1(AnyFunSuiteLike.scala:232)
at org.scalatest.SuperEngine.$anonfun$runTestsInBranch$1(Engine.scala:413)
at scala.collection.immutable.List.foreach(List.scala:392)
at org.scalatest.SuperEngine.traverseSubNodes$1(Engine.scala:401)

I will keep monitoring and see if it's frequent.

…kSchedulerImpl realm ### What changes were proposed in this pull request? The decommissioning state is a bit fragment across two places in the TaskSchedulerImpl: #29014 stored the incoming decommission info messages in TaskSchedulerImpl.executorsPendingDecommission. While #28619 was storing just the executor end time in the map TaskSetManager.tidToExecutorKillTimeMapping (which in turn is contained in TaskSchedulerImpl). While the two states are not really overlapping, it's a bit of a code hygiene concern to save this state in two places. With #29422, TaskSchedulerImpl is emerging as the place where all decommissioning book keeping is kept within the driver. So consolidate the information in _tidToExecutorKillTimeMapping_ into _executorsPendingDecommission_. However, in order to do so, we need to walk away from keeping the raw ExecutorDecommissionInfo messages and instead keep another class ExecutorDecommissionState. This decoupling will allow the RPC message class ExecutorDecommissionInfo to evolve independently from the book keeping ExecutorDecommissionState. ### Why are the changes needed? This is just a code cleanup. These two features were added independently and its time to consolidate their state for good hygiene. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Existing tests. Closes#29452 from agrawaldevesh/consolidate_decom_state. Authored-by: Devesh Agrawal <devesh.agrawal@gmail.com> Signed-off-by: Holden Karau <hkarau@apple.com>

[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

probot-autolabelerBot added the CORE label Jul 6, 2020

agrawaldevesh force-pushed the decom_harden branch from b6690d8 to 40cfea9Compare July 6, 2020 18:05

agrawaldevesh force-pushed the decom_harden branch 2 times, most recently from a4d7185 to 2254234Compare July 6, 2020 18:46

agrawaldevesh changed the title ~~[WIP]Reduce job failures during decommissioning~~[WIP] Reduce job failures during decommissioningJul 6, 2020

agrawaldevesh force-pushed the decom_harden branch from 2254234 to e7113a6Compare July 6, 2020 19:02

agrawaldevesh changed the title ~~[WIP] Reduce job failures during decommissioning~~[WIP][SPARK-32199][SPARK-32198] Reduce job failures during decommissioningJul 6, 2020

agrawaldevesh force-pushed the decom_harden branch 2 times, most recently from bbee57d to 01ebb47Compare July 8, 2020 00:57

agrawaldevesh changed the title ~~[WIP][SPARK-32199][SPARK-32198] Reduce job failures during decommissioning~~[SPARK-32199][SPARK-32198] Reduce job failures during decommissioningJul 8, 2020

jiangxb1987 reviewed Jul 10, 2020
View reviewed changes

Ngone51 mentioned this pull request Jul 13, 2020
[SPARK-32217] Plumb whether a worker would also be decommissioned along with executor #29032
Closed

Ngone51 reviewed Jul 13, 2020
View reviewed changes

Comment threadcore/src/main/scala/org/apache/spark/scheduler/ExecutorLossReason.scala Outdated

Ngone51 reviewed Jul 13, 2020
View reviewed changes

agrawaldevesh force-pushed the decom_harden branch from d36097c to 5ce99b1Compare July 13, 2020 17:55

agrawaldevesh force-pushed the decom_harden branch 2 times, most recently from 8d123c3 to 92d5e71Compare July 13, 2020 23:37

agrawaldevesh force-pushed the decom_harden branch 2 times, most recently from c476ce1 to 30af8f4Compare July 15, 2020 23:50

Ngone51 reviewed Jul 16, 2020
View reviewed changes

Comment threadcore/src/main/scala/org/apache/spark/scheduler/ExecutorDecommissionInfo.scala Outdated

attilapiros reviewed Jul 28, 2020
View reviewed changes

Comment threadcore/src/test/scala/org/apache/spark/deploy/DecommissionWorkerSuite.scala Outdated

attilapiros reviewed Jul 28, 2020
View reviewed changes

Comment threadcore/src/test/scala/org/apache/spark/deploy/DecommissionWorkerSuite.scala Outdated

attilapiros reviewed Jul 28, 2020
View reviewed changes

Comment threadcore/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala Outdated

attilapiros reviewed Jul 28, 2020
View reviewed changes

Comment threadcore/src/test/scala/org/apache/spark/deploy/DecommissionWorkerSuite.scala Outdated

agrawaldevesh force-pushed the decom_harden branch from 806630a to 153e8e4Compare July 28, 2020 17:16

agrawaldevesh force-pushed the decom_harden branch 3 times, most recently from 1fd2cca to a906041Compare July 28, 2020 17:54

attilapiros approved these changes Jul 28, 2020
View reviewed changes

agrawaldevesh force-pushed the decom_harden branch from a906041 to dbc478bCompare July 29, 2020 01:25

holdenk approved these changes Jul 30, 2020
View reviewed changes

asfgit closed this in 366a178Jul 30, 2020

agrawaldevesh mentioned this pull request Aug 12, 2020
[SPARK-31198][CORE] Use graceful decommissioning as part of dynamic scaling #29367
Closed

c21 mentioned this pull request Aug 15, 2020
[SPARK-32399][SQL] Full outer shuffled hash join #29342
Closed

agrawaldevesh mentioned this pull request Aug 18, 2020
[SPARK-32643][CORE][K8s] Consolidate state decommissioning in the TaskSchedulerImpl realm #29452
Closed

Uh oh!

Conversation

agrawaldevesh commented Jul 6, 2020• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Code overview

Does this PR introduce any user-facing change?

How was this patch tested?

Questions for reviewers

Uh oh!

agrawaldevesh commented Jul 6, 2020

Uh oh!

holdenk commented Jul 6, 2020

Uh oh!

holdenk commented Jul 6, 2020

Uh oh!

agrawaldevesh commented Jul 6, 2020• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

agrawaldevesh commented Jul 10, 2020

Uh oh!

jiangxb1987 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

agrawaldevesh commented Jul 14, 2020

Uh oh!

cloud-fan commented Jul 16, 2020

Uh oh!

SparkQA commented Jul 16, 2020

Uh oh!

Uh oh!

SparkQA commented Jul 28, 2020

Uh oh!

Uh oh!

Uh oh!

Uh oh!

attilapiros commented Jul 28, 2020

Uh oh!

Uh oh!

agrawaldevesh commented Jul 28, 2020• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

SparkQA commented Jul 28, 2020

Uh oh!

attilapiros commented Jul 28, 2020

Uh oh!

attilapiros commented Jul 28, 2020• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

SparkQA commented Jul 28, 2020

Uh oh!

SparkQA commented Jul 28, 2020

Uh oh!

agrawaldevesh commented Jul 29, 2020

Uh oh!

SparkQA commented Jul 29, 2020

Uh oh!

holdenk left a comment

Choose a reason for hiding this comment

Uh oh!

HyukjinKwon commented Aug 12, 2020• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

agrawaldevesh commented Jul 6, 2020•
edited
Loading

agrawaldevesh commented Jul 6, 2020•
edited
Loading

agrawaldevesh commented Jul 28, 2020•
edited
Loading

attilapiros commented Jul 28, 2020•
edited
Loading

HyukjinKwon commented Aug 12, 2020•
edited
Loading