Swapped last 2 arguments passed to CoarseGrainedExecutorBackend

Refactored MetricsSystem#buildRegistryName

Author: sarutak

core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala

@@ -106,8 +106,8 @@ private[spark] object CoarseGrainedExecutorBackend extends Logging {
       executorId: String,
       hostname: String,
       cores: Int,
-      workerUrl: Option[String],
-      appId: String) {
+      appId: String,
+      workerUrl: Option[String]) {
 
     SignalLogger.register(log)
 
@@ -150,12 +150,12 @@ private[spark] object CoarseGrainedExecutorBackend extends Logging {
         System.err.println(
           // Worker url is used in spark standalone mode to enforce fate-sharing with worker
           "Usage: CoarseGrainedExecutorBackend <driverUrl> <executorId> <hostname> " +
-          "<cores> [<workerUrl>]")
+          "<cores> <appid> [<workerUrl>] ")
         System.exit(1)
       case 5 =>
-        run(args(0), args(1), args(2), args(3).toInt, None, args(4))
+        run(args(0), args(1), args(2), args(3).toInt, args(4), None)
       case x if x > 5 =>
-        run(args(0), args(1), args(2), args(3).toInt, Some(args(4)), args(5))
+        run(args(0), args(1), args(2), args(3).toInt, args(4), Some(args(5)))
     }
   }
 }

core/src/main/scala/org/apache/spark/metrics/MetricsSystem.scala

@@ -94,19 +94,15 @@ private[spark] class MetricsSystem private (val instance: String,
   }
 
   def buildRegistryName(source: Source) = {
-    val appNameOpt = conf.getOption("spark.app.name")
-    val appIdOpt = conf.getOption("spark.app.id")
-    val executorIdOpt = conf.getOption("spark.executor.id")
-    val registryName = {
-      if (appNameOpt.isDefined && appIdOpt.isDefined &&
-        executorIdOpt.isDefined) {
-        MetricRegistry.name(appIdOpt.get, appNameOpt.get,
-          executorIdOpt.get, source.sourceName)
-      } else {
-        MetricRegistry.name(source.sourceName)
-      }
+    for {
+      appName <- conf.getOption("spark.app.name")
+      appId <- conf.getOption("spark.app.id")
+      executorId <- conf.getOption("spark.executor.id")
+    } yield {
+      MetricRegistry.name(appId, appName, executorId, source.sourceName)
     }
-    registryName
+  }.getOrElse {
+    MetricRegistry.name(source.sourceName)
   }
 
   def registerSource(source: Source) {