[DON'T MERGE] Try to replace all json4s with Jackson

connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/JsonUtils.scala

@@ -21,27 +21,27 @@ import scala.collection.mutable.HashMap
 import scala.util.control.NonFatal
 
 import org.apache.kafka.common.TopicPartition
-import org.json4s.NoTypeHints
-import org.json4s.jackson.Serialization
+
+import org.apache.spark.util.JacksonUtils
 
 /**
  * Utilities for converting Kafka related objects to and from json.
  */
 private object JsonUtils {
-  private implicit val formats = Serialization.formats(NoTypeHints)
 
   /**
    * Read TopicPartitions from json string
    */
   def partitions(str: String): Array[TopicPartition] = {
     try {
-      Serialization.read[Map[String, Seq[Int]]](str).flatMap {  case (topic, parts) =>
+      JacksonUtils.readValue[Map[String, Seq[Int]]](str)
+        .flatMap { case (topic, parts) =>
           parts.map { part =>
             new TopicPartition(topic, part)
           }
-      }.toArray
+        }.toArray
     } catch {
-      case NonFatal(x) =>
+      case NonFatal(_) =>
         throw new IllegalArgumentException(
           s"""Expected e.g. {"topicA":[0,1],"topicB":[0,1]}, got $str""")
     }
@@ -56,35 +56,37 @@ private object JsonUtils {
       val parts: List[Int] = result.getOrElse(tp.topic, Nil)
       result += tp.topic -> (tp.partition::parts)
     }
-    Serialization.write(result)
+    JacksonUtils.writeValueAsString(result)
   }
 
   /**
    * Read per-TopicPartition offsets from json string
    */
   def partitionOffsets(str: String): Map[TopicPartition, Long] = {
     try {
-      Serialization.read[Map[String, Map[Int, Long]]](str).flatMap { case (topic, partOffsets) =>
+      JacksonUtils.readValue[Map[String, Map[Int, Long]]](str)
+        .flatMap { case (topic, partOffsets) =>
           partOffsets.map { case (part, offset) =>
-              new TopicPartition(topic, part) -> offset
+            new TopicPartition(topic, part) -> offset
           }
-      }
+        }
     } catch {
-      case NonFatal(x) =>
+      case NonFatal(_) =>
         throw new IllegalArgumentException(
           s"""Expected e.g. {"topicA":{"0":23,"1":-1},"topicB":{"0":-2}}, got $str""")
     }
   }
 
   def partitionTimestamps(str: String): Map[TopicPartition, Long] = {
     try {
-      Serialization.read[Map[String, Map[Int, Long]]](str).flatMap { case (topic, partTimestamps) =>
-        partTimestamps.map { case (part, timestamp) =>
-          new TopicPartition(topic, part) -> timestamp
+      JacksonUtils.readValue[Map[String, Map[Int, Long]]](str)
+        .flatMap { case (topic, partTimestamps) =>
+          partTimestamps.map { case (part, timestamp) =>
+            new TopicPartition(topic, part) -> timestamp
+          }
         }
-      }
     } catch {
-      case NonFatal(x) =>
+      case NonFatal(_) =>
         throw new IllegalArgumentException(
           s"""Expected e.g. {"topicA": {"0": 123456789, "1": 123456789},
              |"topicB": {"0": 123456789, "1": 123456789}}, got $str""".stripMargin)
@@ -103,12 +105,12 @@ private object JsonUtils {
     }
     val partitions = partitionOffsets.keySet.toSeq.sorted  // sort for more determinism
     partitions.foreach { tp =>
-        val off = partitionOffsets(tp)
-        val parts = result.getOrElse(tp.topic, new HashMap[Int, Long])
-        parts += tp.partition -> off
-        result += tp.topic -> parts
+      val off = partitionOffsets(tp)
+      val parts = result.getOrElse(tp.topic, new HashMap[Int, Long])
+      parts += tp.partition -> off
+      result += tp.topic -> parts
     }
-    Serialization.write(result)
+    JacksonUtils.writeValueAsString(result)
   }
 
   def partitionTimestamps(topicTimestamps: Map[TopicPartition, Long]): String = {

core/pom.xml

@@ -274,6 +274,7 @@
     <dependency>
       <groupId>org.json4s</groupId>
       <artifactId>json4s-jackson_${scala.binary.version}</artifactId>
+      <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>org.glassfish.jersey.core</groupId>

core/src/main/scala/org/apache/spark/TestUtils.scala

@@ -50,8 +50,6 @@ import org.eclipse.jetty.server.Server
 import org.eclipse.jetty.server.handler.DefaultHandler
 import org.eclipse.jetty.server.handler.HandlerList
 import org.eclipse.jetty.server.handler.ResourceHandler
-import org.json4s.JsonAST.JValue
-import org.json4s.jackson.JsonMethods.{compact, render}
 
 import org.apache.spark.executor.TaskMetrics
 import org.apache.spark.scheduler._
@@ -462,13 +460,6 @@ private[spark] object TestUtils {
     result.toArray
   }
 
-  /** Creates a temp JSON file that contains the input JSON record. */
-  def createTempJsonFile(dir: File, prefix: String, jsonValue: JValue): String = {
-    val file = File.createTempFile(prefix, ".json", dir)
-    JavaFiles.write(file.toPath, compact(render(jsonValue)).getBytes())
-    file.getPath
-  }
-
   /** Creates a temp bash script that prints the given output. */
   def createTempScriptWithExpectedOutput(dir: File, prefix: String, output: String): String = {
     val file = File.createTempFile(prefix, ".sh", dir)

core/src/main/scala/org/apache/spark/deploy/FaultToleranceTest.scala

@@ -30,8 +30,6 @@ import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.duration._
 import scala.sys.process._
 
-import org.json4s.jackson.JsonMethods
-
 import org.apache.spark.{SparkConf, SparkContext}
 import org.apache.spark.deploy.master.RecoveryState
 import org.apache.spark.internal.{config, Logging}
@@ -340,7 +338,6 @@ private object FaultToleranceTest extends App with Logging {
 private class TestMasterInfo(val ip: String, val dockerId: DockerId, val logFile: File)
   extends Logging  {
 
-  implicit val formats = org.json4s.DefaultFormats
   var state: RecoveryState.Value = _
   var liveWorkerIPs: List[String] = _
   var numLiveApps = 0
@@ -349,22 +346,25 @@ private class TestMasterInfo(val ip: String, val dockerId: DockerId, val logFile
 
   def readState(): Unit = {
     try {
+      import org.apache.spark.util.JacksonUtils
+      import scala.collection.JavaConverters._
       val masterStream = new InputStreamReader(
         new URL("http://%s:8080/json".format(ip)).openStream, StandardCharsets.UTF_8)
-      val json = JsonMethods.parse(masterStream)
+      val jsonNode = JacksonUtils.readTree(masterStream)
 
-      val workers = json \ "workers"
-      val liveWorkers = workers.children.filter(w => (w \ "state").extract[String] == "ALIVE")
+      val workersNode = jsonNode.get("workers")
+      val liveWorkerNodes = workersNode.elements().asScala
+        .filter(n => n.get("state").asText() == "ALIVE")
       // Extract the worker IP from "webuiaddress" (rather than "host") because the host name
       // on containers is a weird hash instead of the actual IP address.
-      liveWorkerIPs = liveWorkers.map {
-        w => (w \ "webuiaddress").extract[String].stripPrefix("http://").stripSuffix(":8081")
-      }
+      liveWorkerIPs = liveWorkerNodes.map {
+        w => w.get("webuiaddress").asText().stripPrefix("http://").stripSuffix(":8081")
+      }.toList
 
-      numLiveApps = (json \ "activeapps").children.size
+      numLiveApps = jsonNode.get("activeapps").size()
 
-      val status = json \\ "status"
-      val stateString = status.extract[String]
+      val status = jsonNode.get("status")
+      val stateString = status.asText()
       state = RecoveryState.values.filter(state => state.toString == stateString).head
     } catch {
       case e: Exception =>
@@ -383,8 +383,6 @@ private class TestMasterInfo(val ip: String, val dockerId: DockerId, val logFile
 private class TestWorkerInfo(val ip: String, val dockerId: DockerId, val logFile: File)
   extends Logging {
 
-  implicit val formats = org.json4s.DefaultFormats
-
   logDebug("Created worker: " + this)
 
   def kill(): Unit = { Docker.kill(dockerId) }