[SPARK-12583][Mesos] Mesos shuffle service: Don't delete shuffle files before application has stopped

[bbossy]

Problem description:

Mesos shuffle service is completely unusable since Spark 1.6.0 . The problem seems to occur since the move from akka to netty in the networking layer. Until now, a connection from the driver to each shuffle service was used as a signal for the shuffle service to determine, whether the driver is still running. Since 1.6.0, this connection is closed after spark.shuffle.io.connectionTimeout (or spark.network.timeout if the former is not set) due to it being idle. The shuffle service interprets this as a signal that the driver has stopped, despite the driver still being alive. Thus, shuffle files are deleted before the application has stopped.

Context and analysis:

spark shuffle fails with mesos after 2mins: https://issues.apache.org/jira/browse/SPARK-12583
External shuffle service broken w/ Mesos: https://issues.apache.org/jira/browse/SPARK-13159

This is a follow up on #11207 .

What changes were proposed in this pull request?

This PR adds a heartbeat signal from the Driver (in MesosExternalShuffleClient) to all registered external mesos shuffle service instances. In MesosExternalShuffleBlockHandler, a thread periodically checks whether a driver has timed out and cleans an application's shuffle files if this is the case.

How was the this patch tested?

This patch has been tested on a small mesos test cluster using the spark-shell. Log output from mesos shuffle service:

16/02/19 15:13:45 INFO mesos.MesosExternalShuffleBlockHandler: Received registration request from app 294def07-3249-4e0f-8d71-bf8c83c58a50-0018 (remote address /xxx.xxx.xxx.xxx:52391, heartbeat timeout 120000 ms).
16/02/19 15:13:47 INFO shuffle.ExternalShuffleBlockResolver: Registered executor AppExecId{appId=294def07-3249-4e0f-8d71-bf8c83c58a50-0018, execId=3} with ExecutorShuffleInfo{localDirs=[/foo/blockmgr-c84c0697-a3f9-4f61-9c64-4d3ee227c047], subDirsPerLocalDir=64, shuffleManager=sort}
16/02/19 15:13:47 INFO shuffle.ExternalShuffleBlockResolver: Registered executor AppExecId{appId=294def07-3249-4e0f-8d71-bf8c83c58a50-0018, execId=7} with ExecutorShuffleInfo{localDirs=[/foo/blockmgr-bf46497a-de80-47b9-88f9-563123b59e03], subDirsPerLocalDir=64, shuffleManager=sort}
16/02/19 15:16:02 INFO mesos.MesosExternalShuffleBlockHandler: Application 294def07-3249-4e0f-8d71-bf8c83c58a50-0018 timed out. Removing shuffle files.
16/02/19 15:16:02 INFO shuffle.ExternalShuffleBlockResolver: Application 294def07-3249-4e0f-8d71-bf8c83c58a50-0018 removed, cleanupLocalDirs = true
16/02/19 15:16:02 INFO shuffle.ExternalShuffleBlockResolver: Cleaning up executor AppExecId{appId=294def07-3249-4e0f-8d71-bf8c83c58a50-0018, execId=3}'s 1 local dirs
16/02/19 15:16:02 INFO shuffle.ExternalShuffleBlockResolver: Cleaning up executor AppExecId{appId=294def07-3249-4e0f-8d71-bf8c83c58a50-0018, execId=7}'s 1 local dirs

Note: there are 2 executors running on this slave.

[bbossy]

@dragos @tnachen : Let me know if you had something else in mind (or if I should wire it up differently)

[dragos]

I'll have to come back to this on Monday. Judging by the description, it looks good.

[andrewor14]

ok to test

[SparkQA]

Test build #51562 has finished for PR 11272 at commit bd30655.

• This patch fails RAT tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • public class ShuffleServiceHeartbeat extends BlockTransferMessage

[bbossy]

Oops. Sorry about that..

[SparkQA]

Test build #51567 has finished for PR 11272 at commit 9a3d625.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[dragos]

lastHeartbeat is used from different threads, so it should be @volatile

[bbossy]

good catch!

[dragos]

I'm done reviewing, I only have a couple of small observations.

[bbossy]

@dragos Thanks for the review! Let me know if you find anything else.

[andrewor14]

I would like to backport a variant of this patch into 1.6.1. This was clearly working in 1.5 but then became unusable in 1.6. This means people who upgrade to 1.6 will no longer be able to use dynamic allocation in Spark, a huge regression.

As with all backports there is some risk that we might introduce a regression between 1.6.0 and 1.6.1. It's especially bad if we introduce a regression between maintenance releases, however, because then users will lose confidence in upgrading.

Therefore, it would be best to keep the changed surface area minimal, i.e. to ensure that this patch only affects users of Spark on Mesos with external shuffle service. I haven't looked at this patch in great detail but @bbossy @dragos it would be great if you could keep that in mind when reviewing / addressing review comments.

Thanks for fixing this critical issue!

[andrewor14]

style:

private[mesos] class MesosExternalShuffleBlockHandler(
    transportConf: TransportConf,
    cleanerIntervalSeconds: Long)
  extends ...

[andrewor14]

(Update: this will have to go into 1.6.2 since the 1.6.1 RC is being cut today already)

[SparkQA]

Test build #51670 has finished for PR 11272 at commit d4d1ad7.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #51739 has finished for PR 11272 at commit 0a2d4cb.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[andrewor14]

@dragos any other comments?

[andrewor14]

Also cc @tnachen and @mgummelt for another pair of eyes

[tnachen]

Is it a Spark convention to add a "S" in the end to tag it as Seconds? I'm seeing more example in the code base to have it spelled out. I'd say we should be consistent and use cleanerIntervalSeconds.
Also this is only used in the Mesos shuffle service, I think should name space the configuration and also be consistent with other interval setting names, how about "spark.mesos.shuffle.cleaner.interval"

[tnachen]

This looks like a great candidate to add integration tests for in the mesos-spark-integration-tests suite. Ideally we have something long running that we can run. @dragos perhaps either you guys or we can add it. Ideally longer term we can let community contributors like @bbossy to also add tests there and verify everything runs e2e, but I think currently we don't really have docs to show how to add a test :(

[dragos]

Yes, we could have an integration test for this, shouldn't be too hard to add. The basic idea is to decrease the network timeout and have a job that exceeds that but still succeeds. But I don't think it should block this PR

[tnachen]

@dragos Yes it shouldn't block the PR, just mentioning it as I see the need for it. I have two comments on this, otherwise it LGTM too.

[andrewor14]

What's the status on this patch? Have we at least tested it manually? At the very least we should merge it into 2.0.

@bbossy would you mind rebasing this?

[tnachen]

@andrewor14 I'll try to create a test to verify this, when is the 2.0 closing date?

[andrewor14]

end of month I think?

[dragos]

AFAIK this could go in. I did test it manually and things worked well.

[bbossy]

Sorry about the long silence. Rebased the PR. I haven't had the time yet to test it after the rebase. I should be able to do this until the beginning of next week.

[SparkQA]

Test build #53004 has finished for PR 11272 at commit 3a9914b.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[bbossy]

I tested the rebased version manually. It is still working after the rebase.

[andrewor14]

OK, I'm going to merge this into master. This could potentially go into 1.6 but it's a little scary to merge it there. I'm inclined to leave it out of 1.6 to be conservative. If we can work out a less invasive change later for 1.6 then we could consider that.