[SPARK-20428][Core]REST interface about 'v1/submissions/ kill/', currently only supports delete a single ‘driver’, now let spark support delete some ‘drivers’

core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionServer.scala

@@ -214,15 +214,15 @@ private[rest] abstract class KillRequestServlet extends RestServlet {
   protected override def doPost(
       request: HttpServletRequest,
       response: HttpServletResponse): Unit = {
-    val submissionId = parseSubmissionId(request.getPathInfo)
-    val responseMessage = submissionId.map(handleKill).getOrElse {
+    val submissionIds = parseSubmissionId(request.getPathInfo)
+    val responseMessage = submissionIds.map(handleKill).getOrElse {
       response.setStatus(HttpServletResponse.SC_BAD_REQUEST)
       handleError("Submission ID is missing in kill request.")
     }
     sendResponse(responseMessage, response)
   }
 
-  protected def handleKill(submissionId: String): KillSubmissionResponse
+  protected def handleKill(submissionIds: String): KillSubmissionResponse
 }
 
 /**

core/src/main/scala/org/apache/spark/deploy/rest/StandaloneRestServer.scala

@@ -70,14 +70,24 @@ private[deploy] class StandaloneRestServer(
 private[rest] class StandaloneKillRequestServlet(masterEndpoint: RpcEndpointRef, conf: SparkConf)
   extends KillRequestServlet {
 
-  protected def handleKill(submissionId: String): KillSubmissionResponse = {
-    val response = masterEndpoint.askSync[DeployMessages.KillDriverResponse](
-      DeployMessages.RequestKillDriver(submissionId))
+  protected def handleKill(submissionIds: String): KillSubmissionResponse = {
     val k = new KillSubmissionResponse
-    k.serverSparkVersion = sparkVersion
-    k.message = response.message
-    k.submissionId = submissionId
+    Some(submissionIds).getOrElse("").split(",").map(submissionId => {
+    val response = masterEndpoint.askSync[DeployMessages.KillDriverResponse] (
+    DeployMessages.RequestKillDriver(submissionId))
+    if(k.submissionId.nonEmpty) {
+      k.submissionId = k.submissionId.concat(",").concat(submissionId)
+    } else {
+      k.submissionId = submissionId
+    }
+    if(k.message.nonEmpty) {
+      k.message = response.message
+    } else {
+      k.message = k.message.concat(",").concat(response.message)
+    }
     k.success = response.success
+    })
+    k.serverSparkVersion = sparkVersion
     k
   }
 }

core/src/test/scala/org/apache/spark/deploy/rest/StandaloneRestSubmitSuite.scala

@@ -607,8 +607,8 @@ private class FaultyStandaloneRestServer(
 
   /** A faulty servlet that produces invalid responses. */
   class InvalidKillServlet extends StandaloneKillRequestServlet(masterEndpoint, masterConf) {
-    protected override def handleKill(submissionId: String): KillSubmissionResponse = {
-      val k = super.handleKill(submissionId)
+    protected override def handleKill(submissionIds: String): KillSubmissionResponse = {
+      val k = super.handleKill(submissionIds)
       k.submissionId = null
       k
     }