[SPARK-15606][core] Use non-blocking removeExecutor call to avoid deadlocks

core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala

@@ -295,7 +295,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
           // manager to reregister itself. If that happens, the block manager master will know
           // about the executor, but the scheduler will not. Therefore, we should remove the
           // executor from the block manager when we hit this case.
-          scheduler.sc.env.blockManager.master.removeExecutor(executorId)
+          scheduler.sc.env.blockManager.master.removeExecutorAsync(executorId)
           logInfo(s"Asked to remove non-existent executor $executorId")
       }
     }

core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala

@@ -42,6 +42,14 @@ class BlockManagerMaster(
     logInfo("Removed " + execId + " successfully in removeExecutor")
   }
 
+  /** Request removal of a dead executor from the driver endpoint.
+   *  This is only called on the driver side. Non-blocking
+   */
+  def removeExecutorAsync(execId: String) {
+    driverEndpoint.ask[Boolean](RemoveExecutor(execId))
+    logInfo("Removal of executor " + execId + " requested")
+  }
+
   /** Register the BlockManager's id with the driver. */
   def registerBlockManager(
       blockManagerId: BlockManagerId, maxMemSize: Long, slaveEndpoint: RpcEndpointRef): Unit = {

core/src/test/scala/org/apache/spark/DistributedSuiteMinTHreads.scala

@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark
+
+
+class DistributedSuiteMinThreads extends DistributedSuite {
+
+  // This suite runs DistributeSuite with the number of dispatcher
+  // threads set to the minimum of 2 to help identify deadlocks
+
+  val numThreads = System.getProperty("spark.rpc.netty.dispatcher.numThreads")
+
+  override def beforeAll() {
+    super.beforeAll()
+    System.setProperty("spark.rpc.netty.dispatcher.numThreads", "2")
+  }
+
+  override def afterAll() {
+    if (numThreads == null) {
+      System.clearProperty("spark.rpc.netty.dispatcher.numThreads")
+    } else {
+      System.setProperty("spark.rpc.netty.dispatcher.numThreads", numThreads)
+    }
+    super.afterAll()
+  }
+
+}