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

[bbossy]

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

Proposed fix:

Instead of relying on a connection being disconnected, or a heartbeat signal lost, the mesos shuffle service periodically checks with the mesos master whether the framework (spark application) is still running:

Every spark.storage.blockManagerSlaveTimeoutMs / 4 the mesos shuffle service retrieves the leading master's /master/state.json. It checks whether it got a reply from the actual leading master and updates a "last seen" timestamp in its internal state (spark applications on mesos register with the external shuffle service using their framework id). Then, it deletes the temporary files of all the frameworks (that have previously registered), that have not been reported as running in the past spark.storage.blockManagerSlaveTimeoutMs.

It requires mesos-resolve to be on the PATH where the shuffle service is running. This is used to find the leading master in a mesos-HA setup (through zookeeper).

Further, spark.master needs to be set when the service is started.

[AmplabJenkins]

Can one of the admins verify this patch?

[bbossy]

Should I use a different config key here?

[JoshRosen]

No comment on the contents of this PR (since I haven't looked at it yet), but would you mind changing the PR to something more descriptive? As it stands now, "Fix Mesos shuffle service" is a lot less descriptive than, say, "Delete shuffle files after Mesos shuffle service exits" or something similar.

Could you also edit the description to include a concise one-sentence description of the user-facing bug / symptom that this fixes? Right now this describes a lot of mechanism, but I feel like the description is a bit thin on context for newcomers who are trying to understand what this patch is doing.

[bbossy]

@JoshRosen changed to a more descriptive title and added a more detailed problem description.

[bbossy]

@dragos Take a look, please

[dragos]

@bbossy thanks for picking this up!

I have a problem with the bandwidth this design implies. For instance, my state.json is 200KB (a cluster of 1 master and 2 nodes, with virtually no frameworks running). This is most likely a very low minimum. Multiply this by the number of nodes (let's say we want to scale to 10.000 nodes).

200KB * 10,000 nodes = 2GB

2GB of traffic for each check seems pretty bad.

I think it'd be better (and probably simpler) to increase the timeout for this particular connection, or send a heartbeat message (whatever akka remote did before).

@tnachen @mgummelt what do you think?

[bbossy]

@dragos You're right. I'll have a look at your proposals.

[tnachen]

I also agree with @dragos, and I think we should keep the same semantics by having a heartbeat instead.

[IgorBerman]

Hi @bbossy
just to make sure this PR wasn't merged in favor to #11272?

[bbossy]

@IgorBerman correct, #11272 was a follow up on this and got merged. I still experience issues with the shuffle service on Mesos, but I haven't been able to pin point the cause.

[IgorBerman]

@bbossy thanks. I'm experiencing them too despite the merged PR. I think I have some direction why it fails, here is my setup:

0. v2.2.0
1. Dynamic allocation is on
2. min executors is 2(e.g.)
3. I've enabled 2 loggers MesosCoarseGrainedSchedulerBackend and MesosExternalShuffleClient to be debug
4. My workload is empty-high-empty

So it behaves like following:

1. In the beginning driver registers to external shuffle service(for 2 initial executors which are tasks in mesos). I can see in log "Mesos task 1 is now TASK_RUNNING" and then "Connecting to shuffle service on slave..." two times for 2 inital executors
2. After the load increases I see that dynamic allocation works and there are new executors created(in mesos console I see new tasks), however driver doesn't get notification on status update, so it doesn't register to new external shuffle service.(i.e. there is no additional Mesos task x is now TASK_RUNNING"

WDYT?

[IgorBerman]

@bbossy so my work around: ive disabled cleanup of external shuffle service and removing shuffle files by cron that finds files that were not accessed in last X hours.