{"action":"created","comment":{"url":"https://api.github.com/repos/apache/spark/pulls/comments/53665765","id":53665765,"diff_hunk":"@@ -17,69 +17,92 @@\n \n package org.apache.spark.deploy.mesos\n \n-import java.net.SocketAddress\n import java.nio.ByteBuffer\n+import java.util.concurrent.{ConcurrentHashMap, TimeUnit}\n \n-import scala.collection.mutable\n+import scala.collection.JavaConverters._\n \n import org.apache.spark.{Logging, SecurityManager, SparkConf}\n import org.apache.spark.deploy.ExternalShuffleService\n import org.apache.spark.network.client.{RpcResponseCallback, TransportClient}\n import org.apache.spark.network.shuffle.ExternalShuffleBlockHandler\n import org.apache.spark.network.shuffle.protocol.BlockTransferMessage\n-import org.apache.spark.network.shuffle.protocol.mesos.RegisterDriver\n+import org.apache.spark.network.shuffle.protocol.mesos.{RegisterDriver, ShuffleServiceHeartbeat}\n import org.apache.spark.network.util.TransportConf\n+import org.apache.spark.util.ThreadUtils\n \n /**\n  * An RPC endpoint that receives registration requests from Spark drivers running on Mesos.\n  * It detects driver termination and calls the cleanup callback to [[ExternalShuffleService]].\n  */\n-private[mesos] class MesosExternalShuffleBlockHandler(transportConf: TransportConf)\n+private[mesos] class MesosExternalShuffleBlockHandler(\n+  transportConf: TransportConf, cleanerIntervalS: Long)\n   extends ExternalShuffleBlockHandler(transportConf, null) with Logging {\n \n-  // Stores a map of driver socket addresses to app ids\n-  private val connectedApps = new mutable.HashMap[SocketAddress, String]\n+  ThreadUtils.newDaemonSingleThreadScheduledExecutor(\"shuffle-cleaner-watcher\")\n+    .scheduleAtFixedRate(new CleanerThread(), 0, cleanerIntervalS, TimeUnit.SECONDS)\n+\n+  // Stores a map of app id to app state (timeout value and last heartbeat)\n+  private val connectedApps = new ConcurrentHashMap[String, AppState]()\n \n   protected override def handleMessage(\n       message: BlockTransferMessage,\n       client: TransportClient,\n       callback: RpcResponseCallback): Unit = {\n     message match {\n-      case RegisterDriverParam(appId) =>\n+      case RegisterDriverParam(appId, appState) =>\n         val address = client.getSocketAddress\n-        logDebug(s\"Received registration request from app $appId (remote address $address).\")\n-        if (connectedApps.contains(address)) {\n-          val existingAppId = connectedApps(address)\n-          if (!existingAppId.equals(appId)) {\n-            logError(s\"A new app '$appId' has connected to existing address $address, \" +\n-              s\"removing previously registered app '$existingAppId'.\")\n-            applicationRemoved(existingAppId, true)\n-          }\n+        val timeout = appState.heartbeatTimeout\n+        logInfo(s\"Received registration request from app $appId (remote address $address, \" +\n+          s\"heartbeat timeout $timeout ms).\")\n+        if (connectedApps.containsKey(appId)) {\n+          logWarning(s\"Received a registration request from app $appId, but it was already \" +\n+            s\"registered\")\n         }\n-        connectedApps(address) = appId\n+        connectedApps.put(appId, appState)\n         callback.onSuccess(ByteBuffer.allocate(0))\n+      case Heartbeat(appId) =>\n+        val address = client.getSocketAddress\n+        Option(connectedApps.get(appId)) match {\n+          case Some(existingAppState) =>\n+            logTrace(s\"Received ShuffleServiceHeartbeat from app '$appId' (remote \" +\n+              s\"address $address).\")\n+            existingAppState.lastHeartbeat = System.nanoTime()\n+          case None =>\n+            logWarning(s\"Received ShuffleServiceHeartbeat from an unknown app (remote \" +\n+              s\"address $address, appId '$appId').\")\n+        }\n       case _ => super.handleMessage(message, client, callback)\n     }\n   }\n \n-  /**\n-   * On connection termination, clean up shuffle files written by the associated application.\n-   */\n-  override def channelInactive(client: TransportClient): Unit = {\n-    val address = client.getSocketAddress\n-    if (connectedApps.contains(address)) {\n-      val appId = connectedApps(address)\n-      logInfo(s\"Application $appId disconnected (address was $address).\")\n-      applicationRemoved(appId, true /* cleanupLocalDirs */)\n-      connectedApps.remove(address)\n-    } else {\n-      logWarning(s\"Unknown $address disconnected.\")\n-    }\n-  }\n-\n   /** An extractor object for matching [[RegisterDriver]] message. */\n   private object RegisterDriverParam {\n-    def unapply(r: RegisterDriver): Option[String] = Some(r.getAppId)\n+    def unapply(r: RegisterDriver): Option[(String, AppState)] =\n+      Some((r.getAppId, AppState(r.getHeartbeatTimeoutMs, System.nanoTime())))\n+  }\n+\n+  private object Heartbeat {\n+    def unapply(h: ShuffleServiceHeartbeat): Option[String] = Some(h.getAppId)\n+  }\n+\n+  private case class AppState(heartbeatTimeout: Long, var lastHeartbeat: Long)","path":"core/src/main/scala/org/apache/spark/deploy/mesos/MesosExternalShuffleService.scala","position":105,"original_position":105,"commit_id":"9a3d625ffe664f3f247c1a513142f22a65fb46bf","original_commit_id":"9a3d625ffe664f3f247c1a513142f22a65fb46bf","user":{"login":"bbossy","id":4428853,"avatar_url":"https://avatars.githubusercontent.com/u/4428853?v=3","gravatar_id":"","url":"https://api.github.com/users/bbossy","html_url":"https://github.com/bbossy","followers_url":"https://api.github.com/users/bbossy/followers","following_url":"https://api.github.com/users/bbossy/following{/other_user}","gists_url":"https://api.github.com/users/bbossy/gists{/gist_id}","starred_url":"https://api.github.com/users/bbossy/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/bbossy/subscriptions","organizations_url":"https://api.github.com/users/bbossy/orgs","repos_url":"https://api.github.com/users/bbossy/repos","events_url":"https://api.github.com/users/bbossy/events{/privacy}","received_events_url":"https://api.github.com/users/bbossy/received_events","type":"User","site_admin":false},"body":"good catch!","created_at":"2016-02-22T18:10:02Z","updated_at":"2016-02-22T18:10:02Z","html_url":"https://github.com/apache/spark/pull/11272#discussion_r53665765","pull_request_url":"https://api.github.com/repos/apache/spark/pulls/11272","_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/comments/53665765"},"html":{"href":"https://github.com/apache/spark/pull/11272#discussion_r53665765"},"pull_request":{"href":"https://api.github.com/repos/apache/spark/pulls/11272"}}},"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/11272","id":59954912,"html_url":"https://github.com/apache/spark/pull/11272","diff_url":"https://github.com/apache/spark/pull/11272.diff","patch_url":"https://github.com/apache/spark/pull/11272.patch","issue_url":"https://api.github.com/repos/apache/spark/issues/11272","number":11272,"state":"open","locked":false,"title":"[SPARK-12583][Mesos] Mesos shuffle service: Don't delete shuffle files before application has stopped","user":{"login":"bbossy","id":4428853,"avatar_url":"https://avatars.githubusercontent.com/u/4428853?v=3","gravatar_id":"","url":"https://api.github.com/users/bbossy","html_url":"https://github.com/bbossy","followers_url":"https://api.github.com/users/bbossy/followers","following_url":"https://api.github.com/users/bbossy/following{/other_user}","gists_url":"https://api.github.com/users/bbossy/gists{/gist_id}","starred_url":"https://api.github.com/users/bbossy/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/bbossy/subscriptions","organizations_url":"https://api.github.com/users/bbossy/orgs","repos_url":"https://api.github.com/users/bbossy/repos","events_url":"https://api.github.com/users/bbossy/events{/privacy}","received_events_url":"https://api.github.com/users/bbossy/received_events","type":"User","site_admin":false},"body":"## Problem description:\r\n\r\nMesos 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.\r\n\r\n### Context and analysis:\r\n\r\nspark shuffle fails with mesos after 2mins: https://issues.apache.org/jira/browse/SPARK-12583\r\nExternal shuffle service broken w/ Mesos: https://issues.apache.org/jira/browse/SPARK-13159\r\n\r\nThis is a follow up on #11207 .\r\n\r\n## What changes were proposed in this pull request?\r\n\r\nThis 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.\r\n\r\n## How was the this patch tested?\r\n\r\nThis patch has been tested on a small mesos test cluster using the spark-shell. Log output from mesos shuffle service:\r\n```\r\n16/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).\r\n16/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}\r\n16/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}\r\n16/02/19 15:16:02 INFO mesos.MesosExternalShuffleBlockHandler: Application 294def07-3249-4e0f-8d71-bf8c83c58a50-0018 timed out. Removing shuffle files.\r\n16/02/19 15:16:02 INFO shuffle.ExternalShuffleBlockResolver: Application 294def07-3249-4e0f-8d71-bf8c83c58a50-0018 removed, cleanupLocalDirs = true\r\n16/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\r\n16/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\r\n```\r\nNote: there are 2 executors running on this slave.\r\n\r\n","created_at":"2016-02-19T14:20:23Z","updated_at":"2016-02-22T18:10:02Z","closed_at":null,"merged_at":null,"merge_commit_sha":"87567b2b9b9dadd99092cec9db5a89a803914f8c","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/apache/spark/pulls/11272/commits","review_comments_url":"https://api.github.com/repos/apache/spark/pulls/11272/comments","review_comment_url":"https://api.github.com/repos/apache/spark/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/spark/issues/11272/comments","statuses_url":"https://api.github.com/repos/apache/spark/statuses/9a3d625ffe664f3f247c1a513142f22a65fb46bf","head":{"label":"bbossy:SPARK-12583-mesos-shuffle-service-heartbeat","ref":"SPARK-12583-mesos-shuffle-service-heartbeat","sha":"9a3d625ffe664f3f247c1a513142f22a65fb46bf","user":{"login":"bbossy","id":4428853,"avatar_url":"https://avatars.githubusercontent.com/u/4428853?v=3","gravatar_id":"","url":"https://api.github.com/users/bbossy","html_url":"https://github.com/bbossy","followers_url":"https://api.github.com/users/bbossy/followers","following_url":"https://api.github.com/users/bbossy/following{/other_user}","gists_url":"https://api.github.com/users/bbossy/gists{/gist_id}","starred_url":"https://api.github.com/users/bbossy/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/bbossy/subscriptions","organizations_url":"https://api.github.com/users/bbossy/orgs","repos_url":"https://api.github.com/users/bbossy/repos","events_url":"https://api.github.com/users/bbossy/events{/privacy}","received_events_url":"https://api.github.com/users/bbossy/received_events","type":"User","site_admin":false},"repo":{"id":22955093,"name":"spark","full_name":"bbossy/spark","owner":{"login":"bbossy","id":4428853,"avatar_url":"https://avatars.githubusercontent.com/u/4428853?v=3","gravatar_id":"","url":"https://api.github.com/users/bbossy","html_url":"https://github.com/bbossy","followers_url":"https://api.github.com/users/bbossy/followers","following_url":"https://api.github.com/users/bbossy/following{/other_user}","gists_url":"https://api.github.com/users/bbossy/gists{/gist_id}","starred_url":"https://api.github.com/users/bbossy/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/bbossy/subscriptions","organizations_url":"https://api.github.com/users/bbossy/orgs","repos_url":"https://api.github.com/users/bbossy/repos","events_url":"https://api.github.com/users/bbossy/events{/privacy}","received_events_url":"https://api.github.com/users/bbossy/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/bbossy/spark","description":"Mirror of Apache Spark","fork":true,"url":"https://api.github.com/repos/bbossy/spark","forks_url":"https://api.github.com/repos/bbossy/spark/forks","keys_url":"https://api.github.com/repos/bbossy/spark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/bbossy/spark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/bbossy/spark/teams","hooks_url":"https://api.github.com/repos/bbossy/spark/hooks","issue_events_url":"https://api.github.com/repos/bbossy/spark/issues/events{/number}","events_url":"https://api.github.com/repos/bbossy/spark/events","assignees_url":"https://api.github.com/repos/bbossy/spark/assignees{/user}","branches_url":"https://api.github.com/repos/bbossy/spark/branches{/branch}","tags_url":"https://api.github.com/repos/bbossy/spark/tags","blobs_url":"https://api.github.com/repos/bbossy/spark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/bbossy/spark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/bbossy/spark/git/refs{/sha}","trees_url":"https://api.github.com/repos/bbossy/spark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/bbossy/spark/statuses/{sha}","languages_url":"https://api.github.com/repos/bbossy/spark/languages","stargazers_url":"https://api.github.com/repos/bbossy/spark/stargazers","contributors_url":"https://api.github.com/repos/bbossy/spark/contributors","subscribers_url":"https://api.github.com/repos/bbossy/spark/subscribers","subscription_url":"https://api.github.com/repos/bbossy/spark/subscription","commits_url":"https://api.github.com/repos/bbossy/spark/commits{/sha}","git_commits_url":"https://api.github.com/repos/bbossy/spark/git/commits{/sha}","comments_url":"https://api.github.com/repos/bbossy/spark/comments{/number}","issue_comment_url":"https://api.github.com/repos/bbossy/spark/issues/comments{/number}","contents_url":"https://api.github.com/repos/bbossy/spark/contents/{+path}","compare_url":"https://api.github.com/repos/bbossy/spark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/bbossy/spark/merges","archive_url":"https://api.github.com/repos/bbossy/spark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/bbossy/spark/downloads","issues_url":"https://api.github.com/repos/bbossy/spark/issues{/number}","pulls_url":"https://api.github.com/repos/bbossy/spark/pulls{/number}","milestones_url":"https://api.github.com/repos/bbossy/spark/milestones{/number}","notifications_url":"https://api.github.com/repos/bbossy/spark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/bbossy/spark/labels{/name}","releases_url":"https://api.github.com/repos/bbossy/spark/releases{/id}","deployments_url":"https://api.github.com/repos/bbossy/spark/deployments","created_at":"2014-08-14T13:39:24Z","updated_at":"2016-02-08T17:21:22Z","pushed_at":"2016-02-19T18:17:26Z","git_url":"git://github.com/bbossy/spark.git","ssh_url":"git@github.com:bbossy/spark.git","clone_url":"https://github.com/bbossy/spark.git","svn_url":"https://github.com/bbossy/spark","homepage":null,"size":140149,"stargazers_count":0,"watchers_count":0,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"fb7e21797ed618d9754545a44f8f95f75b66757a","user":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":17165658,"name":"spark","full_name":"apache/spark","owner":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/apache/spark","description":"Mirror of Apache Spark","fork":false,"url":"https://api.github.com/repos/apache/spark","forks_url":"https://api.github.com/repos/apache/spark/forks","keys_url":"https://api.github.com/repos/apache/spark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/spark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/spark/teams","hooks_url":"https://api.github.com/repos/apache/spark/hooks","issue_events_url":"https://api.github.com/repos/apache/spark/issues/events{/number}","events_url":"https://api.github.com/repos/apache/spark/events","assignees_url":"https://api.github.com/repos/apache/spark/assignees{/user}","branches_url":"https://api.github.com/repos/apache/spark/branches{/branch}","tags_url":"https://api.github.com/repos/apache/spark/tags","blobs_url":"https://api.github.com/repos/apache/spark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/spark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/spark/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/spark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/spark/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/spark/languages","stargazers_url":"https://api.github.com/repos/apache/spark/stargazers","contributors_url":"https://api.github.com/repos/apache/spark/contributors","subscribers_url":"https://api.github.com/repos/apache/spark/subscribers","subscription_url":"https://api.github.com/repos/apache/spark/subscription","commits_url":"https://api.github.com/repos/apache/spark/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/spark/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/spark/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/spark/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/spark/contents/{+path}","compare_url":"https://api.github.com/repos/apache/spark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/spark/merges","archive_url":"https://api.github.com/repos/apache/spark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/spark/downloads","issues_url":"https://api.github.com/repos/apache/spark/issues{/number}","pulls_url":"https://api.github.com/repos/apache/spark/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/spark/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/spark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/spark/labels{/name}","releases_url":"https://api.github.com/repos/apache/spark/releases{/id}","deployments_url":"https://api.github.com/repos/apache/spark/deployments","created_at":"2014-02-25T08:00:08Z","updated_at":"2016-02-22T17:42:12Z","pushed_at":"2016-02-22T18:08:52Z","git_url":"git://github.com/apache/spark.git","ssh_url":"git@github.com:apache/spark.git","clone_url":"https://github.com/apache/spark.git","svn_url":"https://github.com/apache/spark","homepage":null,"size":432884,"stargazers_count":7524,"watchers_count":7524,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":6837,"mirror_url":"git://git.apache.org/spark.git","open_issues_count":387,"forks":6837,"open_issues":387,"watchers":7524,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/11272"},"html":{"href":"https://github.com/apache/spark/pull/11272"},"issue":{"href":"https://api.github.com/repos/apache/spark/issues/11272"},"comments":{"href":"https://api.github.com/repos/apache/spark/issues/11272/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/spark/pulls/11272/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/spark/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/spark/pulls/11272/commits"},"statuses":{"href":"https://api.github.com/repos/apache/spark/statuses/9a3d625ffe664f3f247c1a513142f22a65fb46bf"}}}}