Backward-compatible support

Author: zsxwing

core/src/main/scala/org/apache/spark/util/JsonProtocol.scala

@@ -628,8 +628,9 @@ private[spark] object JsonProtocol {
         val shuffleId = (json \ "Shuffle ID").extract[Int]
         val mapId = (json \ "Map ID").extract[Int]
         val reduceId = (json \ "Reduce ID").extract[Int]
-        val message = (json \ "Message").extract[String]
-        new FetchFailed(blockManagerAddress, shuffleId, mapId, reduceId, message)
+        val message = Utils.jsonOption(json \ "Message").map(_.extract[String])
+        new FetchFailed(blockManagerAddress, shuffleId, mapId, reduceId,
+          message.getOrElse("Unknown"))
       case `exceptionFailure` =>
         val className = (json \ "Class Name").extract[String]
         val description = (json \ "Description").extract[String]