Add JsonProtocol backwards-compatibility

apache · Feb 4, 2015 · 04e410e · 04e410e
1 parent 259ffbf
commit 04e410e
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 4 deletions.
diff --git a/core/src/main/scala/org/apache/spark/util/JsonProtocol.scala b/core/src/main/scala/org/apache/spark/util/JsonProtocol.scala
@@ -22,7 +22,7 @@ import java.util.{Properties, UUID}
 import org.apache.spark.scheduler.cluster.ExecutorInfo
 
 import scala.collection.JavaConverters._
-import scala.collection.Map
+import scala.collection.{immutable, Map}
 
 import org.json4s.DefaultFormats
 import org.json4s.JsonDSL._
@@ -793,15 +793,15 @@ private[spark] object JsonProtocol {
   def executorInfoFromJson(json: JValue): ExecutorInfo = {
     val executorHost = (json \ "Host").extract[String]
     val totalCores = (json \ "Total Cores").extract[Int]
-    val logUrls = mapFromJson(json \ "Log Urls").toMap
+    val logUrls = Utils.jsonOption(json \ "Log Urls").map(mapFromJson).getOrElse(Map.empty)
     new ExecutorInfo(executorHost, totalCores, logUrls)
   }
 
   /** -------------------------------- *
    * Util JSON deserialization methods |
    * --------------------------------- */
 
-  def mapFromJson(json: JValue): Map[String, String] = {
+  def mapFromJson(json: JValue): immutable.Map[String, String] = {
     val jsonFields = json.asInstanceOf[JObject].obj
     jsonFields.map { case JField(k, JString(v)) => (k, v) }.toMap
   }

diff --git a/core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala b/core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala
@@ -22,7 +22,7 @@ import java.util.Properties
 import org.apache.spark.scheduler.cluster.ExecutorInfo
 import org.apache.spark.shuffle.MetadataFetchFailedException
 
-import scala.collection.Map
+import scala.collection.{immutable, Map}
 
 import org.json4s.jackson.JsonMethods._
 import org.scalatest.FunSuite
@@ -280,6 +280,16 @@ class JsonProtocolSuite extends FunSuite {
     assertEquals(expectedJobEnd, JsonProtocol.jobEndFromJson(oldEndEvent))
   }
 
+  test("ExecutorInfo backward compatibility") {
+    // Prior to Spark 1.3.0, ExecutorInfo did not have a "Log Urls" property
+    val executorInfo =
+      new ExecutorInfo("hostname", totalCores = 10, immutable.Map("stderr" -> "somePath"))
+    val oldExecutorInfoJson = JsonProtocol.executorInfoToJson(executorInfo)
+      .removeField{ _._1 == "Log Urls" }
+    val expectedExecutorInfo = new ExecutorInfo("hostname", totalCores = 10, immutable.Map.empty)
+    assertEquals(expectedExecutorInfo, JsonProtocol.executorInfoFromJson(oldExecutorInfoJson))
+  }
+
   /** -------------------------- *
    | Helper test running methods |
    * --------------------------- */