From 97c0a4d1fc1d64392dc0df2709823a24084df19d Mon Sep 17 00:00:00 2001
From: Nicholas Travers <nickt@squareup.com>
Date: Sat, 5 Aug 2017 12:24:07 -0700
Subject: [PATCH 01/12] KAFKA-4914: Partition re-assignment tool should check
 types before persisting state in ZooKeeper

Prior to this, there have been instances where invalid data was allowed to be persisted in ZooKeeper, which causes ClassCastExceptions when a broker is restarted and reads this type-unsafe data.

Adds basic structural and type validation for the reassignment JSON via
introduction of Scala case classes that map to the expected JSON
structure.
---
 core/src/main/scala/kafka/utils/Json.scala    |  9 ++++
 core/src/main/scala/kafka/utils/ZkUtils.scala | 41 +++++++++++++++----
 .../scala/unit/kafka/utils/JsonTest.scala     | 29 +++++++++++--
 .../scala/unit/kafka/utils/ZkUtilsTest.scala  | 40 ++++++++++++++++++
 4 files changed, 107 insertions(+), 12 deletions(-)

diff --git a/core/src/main/scala/kafka/utils/Json.scala b/core/src/main/scala/kafka/utils/Json.scala
index cbb8dac520498..5850eeed1153c 100644
--- a/core/src/main/scala/kafka/utils/Json.scala
+++ b/core/src/main/scala/kafka/utils/Json.scala
@@ -95,4 +95,13 @@ object Json {
    * a jackson-scala dependency).
    */
   def encodeAsBytes(obj: Any): Array[Byte] = mapper.writeValueAsBytes(obj)
+
+  /**
+    * Parse a JSON string into either a generic type T, or a Throwable in the case of exception.
+    */
+  def parseTo[T](input: String, klass: Class[T]): Either[Throwable, T] = {
+    try Right(mapper.readValue(input, klass))
+    catch { case e: Throwable => Left(e)}
+  }
+
 }
diff --git a/core/src/main/scala/kafka/utils/ZkUtils.scala b/core/src/main/scala/kafka/utils/ZkUtils.scala
index d5fde4db559bc..364ea77aa5ac7 100644
--- a/core/src/main/scala/kafka/utils/ZkUtils.scala
+++ b/core/src/main/scala/kafka/utils/ZkUtils.scala
@@ -20,6 +20,7 @@ package kafka.utils
 import java.nio.charset.StandardCharsets
 import java.util.concurrent.CountDownLatch
 
+import com.fasterxml.jackson.annotation.JsonProperty
 import kafka.admin._
 import kafka.api.{ApiVersion, KAFKA_0_10_0_IV1, LeaderAndIsr}
 import kafka.cluster._
@@ -142,19 +143,19 @@ object ZkUtils {
   def getDeleteTopicPath(topic: String): String =
     DeleteTopicsPath + "/" + topic
 
-  // Parses without deduplicating keys so the data can be checked before allowing reassignment to proceed
   def parsePartitionReassignmentData(jsonData: String): Map[TopicAndPartition, Seq[Int]] = {
+    val parseResult = Json.parseTo(jsonData, classOf[PartitionAssignment])
+
+    if (parseResult.isLeft)
+      throw new ConfigException(s"Invalid reassignment config: ${parseResult.left}")
+
+    val assignments = parseResult.right
     val seq = for {
-      js <- Json.parseFull(jsonData).toSeq
-      partitionsSeq <- js.asJsonObject.get("partitions").toSeq
-      p <- partitionsSeq.asJsonArray.iterator
+      assignment <- assignments.get.partitions.asScala
     } yield {
-      val partitionFields = p.asJsonObject
-      val topic = partitionFields("topic").to[String]
-      val partition = partitionFields("partition").to[Int]
-      val newReplicas = partitionFields("replicas").to[Seq[Int]]
-      TopicAndPartition(topic, partition) -> newReplicas
+      (TopicAndPartition(assignment.topic, assignment.partitions), assignment.replicas.asScala)
     }
+
     seq.toMap
   }
 
@@ -1204,3 +1205,25 @@ class ZKCheckedEphemeral(path: String,
     }
   }
 }
+
+// Case classes for JSON deserialization
+
+/**
+  * Deserialized representation of a partition assignment.
+  *
+  * An assignment consists of a `version` and a list of `partitions`, which represent the assignment
+  * of topic-partitions to brokers.
+  */
+case class PartitionAssignment(@JsonProperty("version") version: Int,
+                               @JsonProperty("partitions") partitions: java.util.List[ReplicaAssignment])
+
+/**
+  * Deserialized representation of a replica assignment for a `TopicPartition`, i.e. the assignment
+  * of brokers for a given `TopicPartition`.
+  *
+  * A replica assignment consists of a `topic`, `partition` and a list of `replicas`, which
+  * represent the broker ids that the `TopicPartition` is assigned to.
+  */
+case class ReplicaAssignment(@JsonProperty("topic") topic: String,
+                             @JsonProperty("partition") partitions: Int,
+                             @JsonProperty("replicas") replicas: java.util.List[Int])
diff --git a/core/src/test/scala/unit/kafka/utils/JsonTest.scala b/core/src/test/scala/unit/kafka/utils/JsonTest.scala
index fa2a030f0b5b3..6c04eb3d5db85 100644
--- a/core/src/test/scala/unit/kafka/utils/JsonTest.scala
+++ b/core/src/test/scala/unit/kafka/utils/JsonTest.scala
@@ -18,11 +18,13 @@ package kafka.utils
 
 import java.nio.charset.StandardCharsets
 
-import org.junit.Assert._
-import org.junit.Test
+import com.fasterxml.jackson.annotation.JsonProperty
+import com.fasterxml.jackson.core.JsonParseException
 import com.fasterxml.jackson.databind.JsonNode
 import com.fasterxml.jackson.databind.node._
 import kafka.utils.json.JsonValue
+import org.junit.Assert._
+import org.junit.Test
 
 import scala.collection.JavaConverters._
 import scala.collection.Map
@@ -125,5 +127,26 @@ class JsonTest {
     assertEquals(""""str1\\,str2"""", new String(Json.encodeAsBytes("""str1\,str2"""), StandardCharsets.UTF_8))
     assertEquals(""""\"quoted\""""", new String(Json.encodeAsBytes(""""quoted""""), StandardCharsets.UTF_8))
   }
-  
+
+  @Test
+  def testParseTo() = {
+    val foo = "baz"
+    val bar = 1
+
+    val result = Json.parseTo(s"""{"foo": "$foo", "bar": $bar}""", classOf[TestObject])
+
+    assertTrue(result.isRight)
+    assertEquals(TestObject(foo, bar), result.right.get)
+  }
+
+  @Test
+  def testParseTo_invalidJson() = {
+    val result = Json.parseTo("{invalid json}", classOf[TestObject])
+
+    assertTrue(result.isLeft)
+    assertEquals(classOf[JsonParseException], result.left.get.getClass)
+  }
+
 }
+
+case class TestObject(@JsonProperty("foo") foo: String, @JsonProperty("bar") bar: Int)
diff --git a/core/src/test/scala/unit/kafka/utils/ZkUtilsTest.scala b/core/src/test/scala/unit/kafka/utils/ZkUtilsTest.scala
index 292db8b88958e..7f18186120559 100755
--- a/core/src/test/scala/unit/kafka/utils/ZkUtilsTest.scala
+++ b/core/src/test/scala/unit/kafka/utils/ZkUtilsTest.scala
@@ -21,6 +21,7 @@ import kafka.api.LeaderAndIsr
 import kafka.common.TopicAndPartition
 import kafka.controller.LeaderIsrAndControllerEpoch
 import kafka.zk.ZooKeeperTestHarness
+import org.apache.kafka.common.config.ConfigException
 import org.apache.kafka.common.security.JaasUtils
 import org.junit.Assert._
 import org.junit.{After, Before, Test}
@@ -43,6 +44,21 @@ class ZkUtilsTest extends ZooKeeperTestHarness {
     super.tearDown
   }
 
+  val topic = "foo"
+  val partition1 = 0
+  val replica1 = 1
+  val replica2 = 2
+  val reassignmentJson =
+    s"""
+      |{
+      |  "version":1,
+      |  "partitions": [
+      |    { "topic": "$topic", "partition":$partition1, "replicas":[$replica1, $replica2]},
+      |    { "topic": "$topic", "partition":$partition1, "replicas":[$replica1, $replica2]}
+      |  ]
+      |}
+    """.stripMargin
+
   @Test
   def testSuccessfulConditionalDeletePath() {
     // Given an existing path
@@ -129,4 +145,28 @@ class ZkUtilsTest extends ZooKeeperTestHarness {
       assertEquals(seqid, zkUtils.getSequenceId(path))
     }
   }
+
+  @Test
+  def testParsePartitionReassignmentDataWithoutDedup_invalidJson() = {
+    val jsonStr = "{invalid json}"
+
+    try {
+      ZkUtils.parsePartitionReassignmentData(jsonStr)
+      fail("Should have thrown ConfigException");
+    } catch {
+      case e: ConfigException =>
+        assertTrue(e.getMessage.contains("Invalid reassignment config"))
+    }
+  }
+
+  @Test
+  def testParsePartitionReassignmentData() = {
+    val result = ZkUtils.parsePartitionReassignmentData(reassignmentJson)
+
+    // Duplicates are removed
+    assertEquals(1, result.size)
+    assertTrue(result.contains(TopicAndPartition(topic, partition1)))
+    assertEquals(Seq(replica1, replica2), result(TopicAndPartition(topic, partition1)))
+  }
+
 }

From f51e498892c3c01310186824f9523c0d9bc2ab52 Mon Sep 17 00:00:00 2001
From: Nicholas Travers <nickt@squareup.com>
Date: Mon, 20 Nov 2017 12:09:36 +0800
Subject: [PATCH 02/12] [feedback] Pattern matching

---
 core/src/main/scala/kafka/utils/ZkUtils.scala | 9 +++++----
 1 file changed, 5 insertions(+), 4 deletions(-)

diff --git a/core/src/main/scala/kafka/utils/ZkUtils.scala b/core/src/main/scala/kafka/utils/ZkUtils.scala
index 364ea77aa5ac7..ee0f6f28c685e 100644
--- a/core/src/main/scala/kafka/utils/ZkUtils.scala
+++ b/core/src/main/scala/kafka/utils/ZkUtils.scala
@@ -146,12 +146,13 @@ object ZkUtils {
   def parsePartitionReassignmentData(jsonData: String): Map[TopicAndPartition, Seq[Int]] = {
     val parseResult = Json.parseTo(jsonData, classOf[PartitionAssignment])
 
-    if (parseResult.isLeft)
-      throw new ConfigException(s"Invalid reassignment config: ${parseResult.left}")
+    val assignments = parseResult match {
+      case Left(throwable) => throw new ConfigException(s"Invalid reassignment config: $throwable")
+      case Right(result) => result
+    }
 
-    val assignments = parseResult.right
     val seq = for {
-      assignment <- assignments.get.partitions.asScala
+      assignment <- assignments.partitions.asScala
     } yield {
       (TopicAndPartition(assignment.topic, assignment.partitions), assignment.replicas.asScala)
     }

From 7b05f21c9b0b472108bb6b5c2bcb5d6be7d06cf6 Mon Sep 17 00:00:00 2001
From: Nicholas Travers <nickt@squareup.com>
Date: Mon, 20 Nov 2017 12:15:57 +0800
Subject: [PATCH 03/12] [feedback] Relocate POJOs and update documentation

---
 core/src/main/scala/kafka/utils/ZkUtils.scala | 24 +------------------
 core/src/main/scala/kafka/zk/ZkData.scala     | 18 ++++++++++++++
 2 files changed, 19 insertions(+), 23 deletions(-)

diff --git a/core/src/main/scala/kafka/utils/ZkUtils.scala b/core/src/main/scala/kafka/utils/ZkUtils.scala
index ee0f6f28c685e..c0e7d8e48e918 100644
--- a/core/src/main/scala/kafka/utils/ZkUtils.scala
+++ b/core/src/main/scala/kafka/utils/ZkUtils.scala
@@ -20,7 +20,6 @@ package kafka.utils
 import java.nio.charset.StandardCharsets
 import java.util.concurrent.CountDownLatch
 
-import com.fasterxml.jackson.annotation.JsonProperty
 import kafka.admin._
 import kafka.api.{ApiVersion, KAFKA_0_10_0_IV1, LeaderAndIsr}
 import kafka.cluster._
@@ -28,6 +27,7 @@ import kafka.common.{KafkaException, NoEpochForPartitionException, TopicAndParti
 import kafka.consumer.{ConsumerThreadId, TopicCount}
 import kafka.controller.{LeaderIsrAndControllerEpoch, ReassignedPartitionsContext}
 import kafka.zk.{BrokerIdZNode, ZkData}
+import kafka.zk.PartitionAssignment
 import org.I0Itec.zkclient.exception.{ZkBadVersionException, ZkException, ZkMarshallingError, ZkNoNodeException, ZkNodeExistsException}
 import org.I0Itec.zkclient.serialize.ZkSerializer
 import org.I0Itec.zkclient.{IZkChildListener, IZkDataListener, IZkStateListener, ZkClient, ZkConnection}
@@ -1206,25 +1206,3 @@ class ZKCheckedEphemeral(path: String,
     }
   }
 }
-
-// Case classes for JSON deserialization
-
-/**
-  * Deserialized representation of a partition assignment.
-  *
-  * An assignment consists of a `version` and a list of `partitions`, which represent the assignment
-  * of topic-partitions to brokers.
-  */
-case class PartitionAssignment(@JsonProperty("version") version: Int,
-                               @JsonProperty("partitions") partitions: java.util.List[ReplicaAssignment])
-
-/**
-  * Deserialized representation of a replica assignment for a `TopicPartition`, i.e. the assignment
-  * of brokers for a given `TopicPartition`.
-  *
-  * A replica assignment consists of a `topic`, `partition` and a list of `replicas`, which
-  * represent the broker ids that the `TopicPartition` is assigned to.
-  */
-case class ReplicaAssignment(@JsonProperty("topic") topic: String,
-                             @JsonProperty("partition") partitions: Int,
-                             @JsonProperty("replicas") replicas: java.util.List[Int])
diff --git a/core/src/main/scala/kafka/zk/ZkData.scala b/core/src/main/scala/kafka/zk/ZkData.scala
index b6352fa99996f..eac08ad44fcac 100644
--- a/core/src/main/scala/kafka/zk/ZkData.scala
+++ b/core/src/main/scala/kafka/zk/ZkData.scala
@@ -19,6 +19,7 @@ package kafka.zk
 import java.nio.charset.StandardCharsets.UTF_8
 import java.util.Properties
 
+import com.fasterxml.jackson.annotation.JsonProperty
 import kafka.api.{ApiVersion, KAFKA_0_10_0_IV1, LeaderAndIsr}
 import kafka.cluster.{Broker, EndPoint}
 import kafka.common.KafkaException
@@ -566,3 +567,20 @@ object ZkData {
     } else ZooDefs.Ids.OPEN_ACL_UNSAFE.asScala
   }
 }
+
+/**
+  * The assignment of brokers for a `TopicPartition`.
+  *
+  * A replica assignment consists of a `topic`, `partition` and a list of `replicas`, which
+  * represent the broker ids that the `TopicPartition` is assigned to.
+  */
+case class ReplicaAssignment(@JsonProperty("topic") topic: String,
+                             @JsonProperty("partition") partitions: Int,
+                             @JsonProperty("replicas") replicas: java.util.List[Int])
+
+/**
+  * An assignment consists of a `version` and a list of `partitions`, which represent the assignment
+  * of topic-partitions to brokers.
+  */
+case class PartitionAssignment(@JsonProperty("version") version: Int,
+                               @JsonProperty("partitions") partitions: java.util.List[ReplicaAssignment])

From 13583dcfe1776c7e61d13b25e243d2cb6d3db6e5 Mon Sep 17 00:00:00 2001
From: Nicholas Travers <nickt@squareup.com>
Date: Mon, 20 Nov 2017 12:20:49 +0800
Subject: [PATCH 04/12] [feedback] JsonTest companion object

---
 core/src/test/scala/unit/kafka/utils/JsonTest.scala | 7 +++++--
 1 file changed, 5 insertions(+), 2 deletions(-)

diff --git a/core/src/test/scala/unit/kafka/utils/JsonTest.scala b/core/src/test/scala/unit/kafka/utils/JsonTest.scala
index 6c04eb3d5db85..852c5b89d1f63 100644
--- a/core/src/test/scala/unit/kafka/utils/JsonTest.scala
+++ b/core/src/test/scala/unit/kafka/utils/JsonTest.scala
@@ -22,6 +22,7 @@ import com.fasterxml.jackson.annotation.JsonProperty
 import com.fasterxml.jackson.core.JsonParseException
 import com.fasterxml.jackson.databind.JsonNode
 import com.fasterxml.jackson.databind.node._
+import kafka.utils.JsonTest.TestObject
 import kafka.utils.json.JsonValue
 import org.junit.Assert._
 import org.junit.Test
@@ -29,6 +30,10 @@ import org.junit.Test
 import scala.collection.JavaConverters._
 import scala.collection.Map
 
+object JsonTest {
+  case class TestObject(@JsonProperty("foo") foo: String, @JsonProperty("bar") bar: Int)
+}
+
 class JsonTest {
 
   @Test
@@ -148,5 +153,3 @@ class JsonTest {
   }
 
 }
-
-case class TestObject(@JsonProperty("foo") foo: String, @JsonProperty("bar") bar: Int)

From 5aa7f4b3afaa2caf9987ca019209580509d10190 Mon Sep 17 00:00:00 2001
From: Nicholas Travers <nickt@squareup.com>
Date: Mon, 20 Nov 2017 12:00:35 +0800
Subject: [PATCH 05/12] [feedback] parseAs

---
 core/src/main/scala/kafka/utils/Json.scala          | 5 +++--
 core/src/main/scala/kafka/utils/ZkUtils.scala       | 2 +-
 core/src/test/scala/unit/kafka/utils/JsonTest.scala | 4 ++--
 3 files changed, 6 insertions(+), 5 deletions(-)

diff --git a/core/src/main/scala/kafka/utils/Json.scala b/core/src/main/scala/kafka/utils/Json.scala
index 5850eeed1153c..cf35e8812a658 100644
--- a/core/src/main/scala/kafka/utils/Json.scala
+++ b/core/src/main/scala/kafka/utils/Json.scala
@@ -21,6 +21,7 @@ import com.fasterxml.jackson.databind.ObjectMapper
 import kafka.utils.json.JsonValue
 
 import scala.collection._
+import scala.reflect.ClassTag
 
 /**
  * Provides methods for parsing JSON with Jackson and encoding to JSON with a simple and naive custom implementation.
@@ -99,8 +100,8 @@ object Json {
   /**
     * Parse a JSON string into either a generic type T, or a Throwable in the case of exception.
     */
-  def parseTo[T](input: String, klass: Class[T]): Either[Throwable, T] = {
-    try Right(mapper.readValue(input, klass))
+  def parseAs[T](input: String)(implicit tag: ClassTag[T]): Either[Throwable, T] = {
+    try Right(mapper.readValue(input, tag.runtimeClass).asInstanceOf[T])
     catch { case e: Throwable => Left(e)}
   }
 
diff --git a/core/src/main/scala/kafka/utils/ZkUtils.scala b/core/src/main/scala/kafka/utils/ZkUtils.scala
index c0e7d8e48e918..a395960cdb1d3 100644
--- a/core/src/main/scala/kafka/utils/ZkUtils.scala
+++ b/core/src/main/scala/kafka/utils/ZkUtils.scala
@@ -144,7 +144,7 @@ object ZkUtils {
     DeleteTopicsPath + "/" + topic
 
   def parsePartitionReassignmentData(jsonData: String): Map[TopicAndPartition, Seq[Int]] = {
-    val parseResult = Json.parseTo(jsonData, classOf[PartitionAssignment])
+    val parseResult = Json.parseAs[PartitionAssignment](jsonData)
 
     val assignments = parseResult match {
       case Left(throwable) => throw new ConfigException(s"Invalid reassignment config: $throwable")
diff --git a/core/src/test/scala/unit/kafka/utils/JsonTest.scala b/core/src/test/scala/unit/kafka/utils/JsonTest.scala
index 852c5b89d1f63..31f4be308da2f 100644
--- a/core/src/test/scala/unit/kafka/utils/JsonTest.scala
+++ b/core/src/test/scala/unit/kafka/utils/JsonTest.scala
@@ -138,7 +138,7 @@ class JsonTest {
     val foo = "baz"
     val bar = 1
 
-    val result = Json.parseTo(s"""{"foo": "$foo", "bar": $bar}""", classOf[TestObject])
+    val result = Json.parseAs[TestObject](s"""{"foo": "$foo", "bar": $bar}""")
 
     assertTrue(result.isRight)
     assertEquals(TestObject(foo, bar), result.right.get)
@@ -146,7 +146,7 @@ class JsonTest {
 
   @Test
   def testParseTo_invalidJson() = {
-    val result = Json.parseTo("{invalid json}", classOf[TestObject])
+    val result = Json.parseAs[TestObject]("{invalid json}")
 
     assertTrue(result.isLeft)
     assertEquals(classOf[JsonParseException], result.left.get.getClass)

From 8a1c7f2f6609b8c726f8ffa65a39acd40a7f79ad Mon Sep 17 00:00:00 2001
From: Ismael Juma <ismael@juma.me.uk>
Date: Mon, 27 Nov 2017 00:54:35 +0000
Subject: [PATCH 06/12] Use PartitionAssignment/ReplicaAssignment in
 ReassignPartitionsZNode

---
 core/src/main/scala/kafka/utils/Json.scala    | 33 +++++++++++--------
 core/src/main/scala/kafka/utils/ZkUtils.scala |  2 +-
 .../scala/unit/kafka/utils/JsonTest.scala     |  4 +--
 3 files changed, 23 insertions(+), 16 deletions(-)

diff --git a/core/src/main/scala/kafka/utils/Json.scala b/core/src/main/scala/kafka/utils/Json.scala
index cf35e8812a658..3a3377d496d64 100644
--- a/core/src/main/scala/kafka/utils/Json.scala
+++ b/core/src/main/scala/kafka/utils/Json.scala
@@ -46,6 +46,14 @@ object Json {
         catch { case _: JsonProcessingException => None }
     }
 
+  /**
+    * Parse a JSON string into either a generic type T, or a JsonProcessingException in the case of exception.
+    */
+  def parseStringAs[T](input: String)(implicit tag: ClassTag[T]): Either[Throwable, T] = {
+    try Right(mapper.readValue(input, tag.runtimeClass).asInstanceOf[T])
+    catch { case e: JsonProcessingException => Left(e) }
+  }
+
   /**
    * Parse a JSON byte array into a JsonValue if possible. `None` is returned if `input` is not valid JSON.
    */
@@ -57,6 +65,14 @@ object Json {
     try Right(mapper.readTree(input)).right.map(JsonValue(_))
     catch { case e: JsonProcessingException => Left(e) }
 
+  /**
+    * Parse a JSON string into either a generic type T, or a JsonProcessingException in the case of exception.
+    */
+  def parseBytesAs[T](input: Array[Byte])(implicit tag: ClassTag[T]): Either[JsonProcessingException, T] = {
+    try Right(mapper.readValue(input, tag.runtimeClass).asInstanceOf[T])
+    catch { case e: JsonProcessingException => Left(e) }
+  }
+
   /**
    * Encode an object into a JSON string. This method accepts any type T where
    *   T => null | Boolean | String | Number | Map[String, T] | Array[T] | Iterable[T]
@@ -91,18 +107,9 @@ object Json {
   def encodeAsString(obj: Any): String = mapper.writeValueAsString(obj)
 
   /**
-   * Encode an object into a JSON value in bytes. This method accepts any type supported by Jackson's ObjectMapper in
-   * the default configuration. That is, Java collections are supported, but Scala collections are not (to avoid
-   * a jackson-scala dependency).
-   */
-  def encodeAsBytes(obj: Any): Array[Byte] = mapper.writeValueAsBytes(obj)
-
-  /**
-    * Parse a JSON string into either a generic type T, or a Throwable in the case of exception.
+    * Encode an object into a JSON value in bytes. This method accepts any type supported by Jackson's ObjectMapper in
+    * the default configuration. That is, Java collections are supported, but Scala collections are not (to avoid
+    * a jackson-scala dependency).
     */
-  def parseAs[T](input: String)(implicit tag: ClassTag[T]): Either[Throwable, T] = {
-    try Right(mapper.readValue(input, tag.runtimeClass).asInstanceOf[T])
-    catch { case e: Throwable => Left(e)}
-  }
-
+  def encodeAsBytes(obj: Any): Array[Byte] = mapper.writeValueAsBytes(obj)
 }
diff --git a/core/src/main/scala/kafka/utils/ZkUtils.scala b/core/src/main/scala/kafka/utils/ZkUtils.scala
index a395960cdb1d3..e3ae473a0f006 100644
--- a/core/src/main/scala/kafka/utils/ZkUtils.scala
+++ b/core/src/main/scala/kafka/utils/ZkUtils.scala
@@ -144,7 +144,7 @@ object ZkUtils {
     DeleteTopicsPath + "/" + topic
 
   def parsePartitionReassignmentData(jsonData: String): Map[TopicAndPartition, Seq[Int]] = {
-    val parseResult = Json.parseAs[PartitionAssignment](jsonData)
+    val parseResult = Json.parseStringAs[PartitionAssignment](jsonData)
 
     val assignments = parseResult match {
       case Left(throwable) => throw new ConfigException(s"Invalid reassignment config: $throwable")
diff --git a/core/src/test/scala/unit/kafka/utils/JsonTest.scala b/core/src/test/scala/unit/kafka/utils/JsonTest.scala
index 31f4be308da2f..4d933fa99bcf2 100644
--- a/core/src/test/scala/unit/kafka/utils/JsonTest.scala
+++ b/core/src/test/scala/unit/kafka/utils/JsonTest.scala
@@ -138,7 +138,7 @@ class JsonTest {
     val foo = "baz"
     val bar = 1
 
-    val result = Json.parseAs[TestObject](s"""{"foo": "$foo", "bar": $bar}""")
+    val result = Json.parseStringAs[TestObject](s"""{"foo": "$foo", "bar": $bar}""")
 
     assertTrue(result.isRight)
     assertEquals(TestObject(foo, bar), result.right.get)
@@ -146,7 +146,7 @@ class JsonTest {
 
   @Test
   def testParseTo_invalidJson() = {
-    val result = Json.parseAs[TestObject]("{invalid json}")
+    val result = Json.parseStringAs[TestObject]("{invalid json}")
 
     assertTrue(result.isLeft)
     assertEquals(classOf[JsonParseException], result.left.get.getClass)

From 2f36f0bc2e43a95d6fadeb3aabd7e3f3ed7d4de8 Mon Sep 17 00:00:00 2001
From: Nicholas Travers <nickt@squareup.com>
Date: Sun, 17 Dec 2017 11:00:08 -0800
Subject: [PATCH 07/12] [rebase fixups]

---
 core/src/main/scala/kafka/utils/Json.scala    | 20 +++++++++----------
 .../scala/unit/kafka/utils/JsonTest.scala     |  1 -
 .../scala/unit/kafka/utils/ZkUtilsTest.scala  |  1 -
 3 files changed, 10 insertions(+), 12 deletions(-)

diff --git a/core/src/main/scala/kafka/utils/Json.scala b/core/src/main/scala/kafka/utils/Json.scala
index 3a3377d496d64..be45c98ca3b44 100644
--- a/core/src/main/scala/kafka/utils/Json.scala
+++ b/core/src/main/scala/kafka/utils/Json.scala
@@ -47,8 +47,8 @@ object Json {
     }
 
   /**
-    * Parse a JSON string into either a generic type T, or a JsonProcessingException in the case of exception.
-    */
+   * Parse a JSON string into either a generic type T, or a JsonProcessingException in the case of exception.
+   */
   def parseStringAs[T](input: String)(implicit tag: ClassTag[T]): Either[Throwable, T] = {
     try Right(mapper.readValue(input, tag.runtimeClass).asInstanceOf[T])
     catch { case e: JsonProcessingException => Left(e) }
@@ -66,8 +66,8 @@ object Json {
     catch { case e: JsonProcessingException => Left(e) }
 
   /**
-    * Parse a JSON string into either a generic type T, or a JsonProcessingException in the case of exception.
-    */
+   * Parse a JSON string into either a generic type T, or a JsonProcessingException in the case of exception.
+   */
   def parseBytesAs[T](input: Array[Byte])(implicit tag: ClassTag[T]): Either[JsonProcessingException, T] = {
     try Right(mapper.readValue(input, tag.runtimeClass).asInstanceOf[T])
     catch { case e: JsonProcessingException => Left(e) }
@@ -100,16 +100,16 @@ object Json {
   }
 
   /**
-    * Encode an object into a JSON string. This method accepts any type supported by Jackson's ObjectMapper in
+   * Encode an object into a JSON string. This method accepts any type supported by Jackson's ObjectMapper in
    * the default configuration. That is, Java collections are supported, but Scala collections are not (to avoid
    * a jackson-scala dependency).
-    */
+   */
   def encodeAsString(obj: Any): String = mapper.writeValueAsString(obj)
 
   /**
-    * Encode an object into a JSON value in bytes. This method accepts any type supported by Jackson's ObjectMapper in
-    * the default configuration. That is, Java collections are supported, but Scala collections are not (to avoid
-    * a jackson-scala dependency).
-    */
+   * Encode an object into a JSON value in bytes. This method accepts any type supported by Jackson's ObjectMapper in
+   * the default configuration. That is, Java collections are supported, but Scala collections are not (to avoid
+   * a jackson-scala dependency).
+   */
   def encodeAsBytes(obj: Any): Array[Byte] = mapper.writeValueAsBytes(obj)
 }
diff --git a/core/src/test/scala/unit/kafka/utils/JsonTest.scala b/core/src/test/scala/unit/kafka/utils/JsonTest.scala
index 4d933fa99bcf2..3df7040f1a124 100644
--- a/core/src/test/scala/unit/kafka/utils/JsonTest.scala
+++ b/core/src/test/scala/unit/kafka/utils/JsonTest.scala
@@ -151,5 +151,4 @@ class JsonTest {
     assertTrue(result.isLeft)
     assertEquals(classOf[JsonParseException], result.left.get.getClass)
   }
-
 }
diff --git a/core/src/test/scala/unit/kafka/utils/ZkUtilsTest.scala b/core/src/test/scala/unit/kafka/utils/ZkUtilsTest.scala
index 7f18186120559..fe48e7cd1abfb 100755
--- a/core/src/test/scala/unit/kafka/utils/ZkUtilsTest.scala
+++ b/core/src/test/scala/unit/kafka/utils/ZkUtilsTest.scala
@@ -168,5 +168,4 @@ class ZkUtilsTest extends ZooKeeperTestHarness {
     assertTrue(result.contains(TopicAndPartition(topic, partition1)))
     assertEquals(Seq(replica1, replica2), result(TopicAndPartition(topic, partition1)))
   }
-
 }

From ec7881ba726fb369a99a2600d75f43be348a85b7 Mon Sep 17 00:00:00 2001
From: Nicholas Travers <nickt@squareup.com>
Date: Fri, 5 Jan 2018 10:26:54 -0800
Subject: [PATCH 08/12] [feedback]

---
 core/src/main/scala/kafka/utils/Json.scala    |  5 +-
 core/src/main/scala/kafka/utils/ZkUtils.scala | 17 ++---
 .../main/scala/kafka/zk/KafkaZkClient.scala   | 10 ++-
 core/src/main/scala/kafka/zk/ZkData.scala     | 70 +++++++++----------
 .../scala/unit/kafka/utils/ZkUtilsTest.scala  | 38 ----------
 .../zk/ReassignPartitionsZNodeTest.scala      | 45 ++++++++++++
 6 files changed, 95 insertions(+), 90 deletions(-)
 create mode 100644 core/src/test/scala/unit/kafka/zk/ReassignPartitionsZNodeTest.scala

diff --git a/core/src/main/scala/kafka/utils/Json.scala b/core/src/main/scala/kafka/utils/Json.scala
index be45c98ca3b44..486b05a8ec25e 100644
--- a/core/src/main/scala/kafka/utils/Json.scala
+++ b/core/src/main/scala/kafka/utils/Json.scala
@@ -47,9 +47,10 @@ object Json {
     }
 
   /**
-   * Parse a JSON string into either a generic type T, or a JsonProcessingException in the case of exception.
+   * Parse a JSON string into either a generic type T, or a JsonProcessingException in the case of
+    * exception.
    */
-  def parseStringAs[T](input: String)(implicit tag: ClassTag[T]): Either[Throwable, T] = {
+  def parseStringAs[T](input: String)(implicit tag: ClassTag[T]): Either[JsonProcessingException, T] = {
     try Right(mapper.readValue(input, tag.runtimeClass).asInstanceOf[T])
     catch { case e: JsonProcessingException => Left(e) }
   }
diff --git a/core/src/main/scala/kafka/utils/ZkUtils.scala b/core/src/main/scala/kafka/utils/ZkUtils.scala
index e3ae473a0f006..2caafdcd18011 100644
--- a/core/src/main/scala/kafka/utils/ZkUtils.scala
+++ b/core/src/main/scala/kafka/utils/ZkUtils.scala
@@ -26,8 +26,7 @@ import kafka.cluster._
 import kafka.common.{KafkaException, NoEpochForPartitionException, TopicAndPartition}
 import kafka.consumer.{ConsumerThreadId, TopicCount}
 import kafka.controller.{LeaderIsrAndControllerEpoch, ReassignedPartitionsContext}
-import kafka.zk.{BrokerIdZNode, ZkData}
-import kafka.zk.PartitionAssignment
+import kafka.zk.{BrokerIdZNode, ReassignPartitionsZNode, ZkData}
 import org.I0Itec.zkclient.exception.{ZkBadVersionException, ZkException, ZkMarshallingError, ZkNoNodeException, ZkNodeExistsException}
 import org.I0Itec.zkclient.serialize.ZkSerializer
 import org.I0Itec.zkclient.{IZkChildListener, IZkDataListener, IZkStateListener, ZkClient, ZkConnection}
@@ -144,20 +143,12 @@ object ZkUtils {
     DeleteTopicsPath + "/" + topic
 
   def parsePartitionReassignmentData(jsonData: String): Map[TopicAndPartition, Seq[Int]] = {
-    val parseResult = Json.parseStringAs[PartitionAssignment](jsonData)
-
-    val assignments = parseResult match {
-      case Left(throwable) => throw new ConfigException(s"Invalid reassignment config: $throwable")
+    val assignments = ReassignPartitionsZNode.decode(jsonData.getBytes) match {
+      case Left(e) => throw e
       case Right(result) => result
     }
 
-    val seq = for {
-      assignment <- assignments.partitions.asScala
-    } yield {
-      (TopicAndPartition(assignment.topic, assignment.partitions), assignment.replicas.asScala)
-    }
-
-    seq.toMap
+    assignments.map { case (tp, p) => (new TopicAndPartition(tp), p) }
   }
 
   def parseTopicsData(jsonData: String): Seq[String] = {
diff --git a/core/src/main/scala/kafka/zk/KafkaZkClient.scala b/core/src/main/scala/kafka/zk/KafkaZkClient.scala
index 0a2d96a796a0d..9b58fc7cc4be7 100644
--- a/core/src/main/scala/kafka/zk/KafkaZkClient.scala
+++ b/core/src/main/scala/kafka/zk/KafkaZkClient.scala
@@ -666,11 +666,17 @@ class KafkaZkClient private (zooKeeperClient: ZooKeeperClient, isSecure: Boolean
    * Returns all reassignments.
    * @return the reassignments for each partition.
    */
-  def getPartitionReassignment: Map[TopicPartition, Seq[Int]] = {
+  def getPartitionReassignment: collection.Map[TopicPartition, Seq[Int]] = {
     val getDataRequest = GetDataRequest(ReassignPartitionsZNode.path)
     val getDataResponse = retryRequestUntilConnected(getDataRequest)
     getDataResponse.resultCode match {
-      case Code.OK => ReassignPartitionsZNode.decode(getDataResponse.data)
+      case Code.OK =>
+        ReassignPartitionsZNode.decode(getDataResponse.data) match {
+          case Left(e) =>
+            logger.warn(s"Ignoring partition reassignment due to invalid json: ${e.getMessage}", e)
+            Map.empty[TopicPartition, Seq[Int]]
+          case Right(assignments) => assignments
+        }
       case Code.NONODE => Map.empty
       case _ => throw getDataResponse.resultException.get
     }
diff --git a/core/src/main/scala/kafka/zk/ZkData.scala b/core/src/main/scala/kafka/zk/ZkData.scala
index eac08ad44fcac..fbed11dbe1598 100644
--- a/core/src/main/scala/kafka/zk/ZkData.scala
+++ b/core/src/main/scala/kafka/zk/ZkData.scala
@@ -20,6 +20,7 @@ import java.nio.charset.StandardCharsets.UTF_8
 import java.util.Properties
 
 import com.fasterxml.jackson.annotation.JsonProperty
+import com.fasterxml.jackson.core.JsonProcessingException
 import kafka.api.{ApiVersion, KAFKA_0_10_0_IV1, LeaderAndIsr}
 import kafka.cluster.{Broker, EndPoint}
 import kafka.common.KafkaException
@@ -36,9 +37,10 @@ import org.apache.kafka.common.utils.Time
 import org.apache.zookeeper.ZooDefs
 import org.apache.zookeeper.data.{ACL, Stat}
 
+import scala.beans.BeanProperty
 import scala.collection.JavaConverters._
-import scala.collection.Seq
 import scala.collection.mutable.ArrayBuffer
+import scala.collection.{Seq, breakOut}
 
 // This file contains objects for encoding/decoding data stored in ZooKeeper nodes (znodes).
 
@@ -368,26 +370,41 @@ object DeleteTopicsTopicZNode {
 }
 
 object ReassignPartitionsZNode {
+
+  /**
+    * The assignment of brokers for a `TopicPartition`.
+    *
+    * A replica assignment consists of a `topic`, `partition` and a list of `replicas`, which
+    * represent the broker ids that the `TopicPartition` is assigned to.
+    */
+  case class ReplicaAssignment(@BeanProperty @JsonProperty("topic") topic: String,
+                               @BeanProperty @JsonProperty("partition") partition: Int,
+                               @BeanProperty @JsonProperty("replicas") replicas: java.util.List[Int])
+
+  /**
+    * An assignment consists of a `version` and a list of `partitions`, which represent the
+    * assignment of topic-partitions to brokers.
+    */
+  case class PartitionAssignment(@BeanProperty @JsonProperty("version") version: Int,
+                                 @BeanProperty @JsonProperty("partitions") partitions: java.util.List[ReplicaAssignment])
+
   def path = s"${AdminZNode.path}/reassign_partitions"
-  def encode(reassignment: collection.Map[TopicPartition, Seq[Int]]): Array[Byte] = {
-    val reassignmentJson = reassignment.map { case (tp, replicas) =>
-      Map("topic" -> tp.topic, "partition" -> tp.partition, "replicas" -> replicas.asJava).asJava
-    }.asJava
-    Json.encodeAsBytes(Map("version" -> 1, "partitions" -> reassignmentJson).asJava)
+
+  def encode(reassignmentMap: collection.Map[TopicPartition, Seq[Int]]): Array[Byte] = {
+    val reassignment = PartitionAssignment(1,
+      reassignmentMap.toSeq.map { case (tp, replicas) =>
+        ReplicaAssignment(tp.topic, tp.partition, replicas.asJava)
+      }.asJava
+    )
+    Json.encodeAsBytes(reassignment)
   }
-  def decode(bytes: Array[Byte]): Map[TopicPartition, Seq[Int]] = Json.parseBytes(bytes).flatMap { js =>
-    val reassignmentJson = js.asJsonObject
-    val partitionsJsonOpt = reassignmentJson.get("partitions")
-    partitionsJsonOpt.map { partitionsJson =>
-      partitionsJson.asJsonArray.iterator.map { partitionFieldsJs =>
-        val partitionFields = partitionFieldsJs.asJsonObject
-        val topic = partitionFields("topic").to[String]
-        val partition = partitionFields("partition").to[Int]
-        val replicas = partitionFields("replicas").to[Seq[Int]]
-        new TopicPartition(topic, partition) -> replicas
-      }
+
+  def decode(bytes: Array[Byte]): Either[JsonProcessingException, collection.Map[TopicPartition, Seq[Int]]] =
+    Json.parseBytesAs[PartitionAssignment](bytes).right.map { partitionAssignment =>
+      partitionAssignment.partitions.asScala.map { replicaAssignment =>
+        new TopicPartition(replicaAssignment.topic, replicaAssignment.partition) -> replicaAssignment.replicas.asScala
+      }(breakOut)
     }
-  }.map(_.toMap).getOrElse(Map.empty)
 }
 
 object PreferredReplicaElectionZNode {
@@ -567,20 +584,3 @@ object ZkData {
     } else ZooDefs.Ids.OPEN_ACL_UNSAFE.asScala
   }
 }
-
-/**
-  * The assignment of brokers for a `TopicPartition`.
-  *
-  * A replica assignment consists of a `topic`, `partition` and a list of `replicas`, which
-  * represent the broker ids that the `TopicPartition` is assigned to.
-  */
-case class ReplicaAssignment(@JsonProperty("topic") topic: String,
-                             @JsonProperty("partition") partitions: Int,
-                             @JsonProperty("replicas") replicas: java.util.List[Int])
-
-/**
-  * An assignment consists of a `version` and a list of `partitions`, which represent the assignment
-  * of topic-partitions to brokers.
-  */
-case class PartitionAssignment(@JsonProperty("version") version: Int,
-                               @JsonProperty("partitions") partitions: java.util.List[ReplicaAssignment])
diff --git a/core/src/test/scala/unit/kafka/utils/ZkUtilsTest.scala b/core/src/test/scala/unit/kafka/utils/ZkUtilsTest.scala
index fe48e7cd1abfb..01a8ffa30ebd0 100755
--- a/core/src/test/scala/unit/kafka/utils/ZkUtilsTest.scala
+++ b/core/src/test/scala/unit/kafka/utils/ZkUtilsTest.scala
@@ -21,7 +21,6 @@ import kafka.api.LeaderAndIsr
 import kafka.common.TopicAndPartition
 import kafka.controller.LeaderIsrAndControllerEpoch
 import kafka.zk.ZooKeeperTestHarness
-import org.apache.kafka.common.config.ConfigException
 import org.apache.kafka.common.security.JaasUtils
 import org.junit.Assert._
 import org.junit.{After, Before, Test}
@@ -44,21 +43,6 @@ class ZkUtilsTest extends ZooKeeperTestHarness {
     super.tearDown
   }
 
-  val topic = "foo"
-  val partition1 = 0
-  val replica1 = 1
-  val replica2 = 2
-  val reassignmentJson =
-    s"""
-      |{
-      |  "version":1,
-      |  "partitions": [
-      |    { "topic": "$topic", "partition":$partition1, "replicas":[$replica1, $replica2]},
-      |    { "topic": "$topic", "partition":$partition1, "replicas":[$replica1, $replica2]}
-      |  ]
-      |}
-    """.stripMargin
-
   @Test
   def testSuccessfulConditionalDeletePath() {
     // Given an existing path
@@ -146,26 +130,4 @@ class ZkUtilsTest extends ZooKeeperTestHarness {
     }
   }
 
-  @Test
-  def testParsePartitionReassignmentDataWithoutDedup_invalidJson() = {
-    val jsonStr = "{invalid json}"
-
-    try {
-      ZkUtils.parsePartitionReassignmentData(jsonStr)
-      fail("Should have thrown ConfigException");
-    } catch {
-      case e: ConfigException =>
-        assertTrue(e.getMessage.contains("Invalid reassignment config"))
-    }
-  }
-
-  @Test
-  def testParsePartitionReassignmentData() = {
-    val result = ZkUtils.parsePartitionReassignmentData(reassignmentJson)
-
-    // Duplicates are removed
-    assertEquals(1, result.size)
-    assertTrue(result.contains(TopicAndPartition(topic, partition1)))
-    assertEquals(Seq(replica1, replica2), result(TopicAndPartition(topic, partition1)))
-  }
 }
diff --git a/core/src/test/scala/unit/kafka/zk/ReassignPartitionsZNodeTest.scala b/core/src/test/scala/unit/kafka/zk/ReassignPartitionsZNodeTest.scala
new file mode 100644
index 0000000000000..fbd7f3602980b
--- /dev/null
+++ b/core/src/test/scala/unit/kafka/zk/ReassignPartitionsZNodeTest.scala
@@ -0,0 +1,45 @@
+package unit.kafka.zk
+
+import com.fasterxml.jackson.core.JsonProcessingException
+import kafka.zk.ReassignPartitionsZNode
+import org.apache.kafka.common.TopicPartition
+import org.junit.Assert._
+import org.junit.Test
+
+class ReassignPartitionsZNodeTest {
+
+  val topic = "foo"
+  val partition1 = 0
+  val replica1 = 1
+  val replica2 = 2
+
+  private val reassignPartitionData = Map(
+    new TopicPartition(topic, partition1) -> Seq(replica1, replica2))
+  private val reassignmentJson = "{\"version\":1,\"partitions\":[{\"topic\":\"foo\",\"partition\":0,\"replicas\":[1,2]}]}"
+
+  @Test
+  def testEncode() {
+    val encodedJsonString = ReassignPartitionsZNode.encode(reassignPartitionData)
+      .map(_.toChar)
+      .mkString
+
+    assertEquals(reassignmentJson, encodedJsonString)
+  }
+
+  @Test
+  def testDecodeInvalidJson() {
+    val result = ReassignPartitionsZNode.decode("invalid json".getBytes)
+
+    assertTrue(result.isLeft)
+    assertTrue(result.left.get.isInstanceOf[JsonProcessingException])
+  }
+
+  @Test
+  def testDecodeValidJson() {
+    val result = ReassignPartitionsZNode.decode(reassignmentJson.getBytes)
+
+    assertTrue(result.isRight)
+    val assignmentMap = result.right.get
+    assertEquals(Seq(replica1, replica2), assignmentMap(new TopicPartition(topic, partition1)))
+  }
+}

From fae8341adbabf68ed2261d40e5f1b08d2de2c843 Mon Sep 17 00:00:00 2001
From: Nicholas Travers <nickt@squareup.com>
Date: Fri, 5 Jan 2018 13:48:30 -0800
Subject: [PATCH 09/12] [fixup] Encode bytes as UTF-8

---
 core/src/main/scala/kafka/utils/ZkUtils.scala | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/core/src/main/scala/kafka/utils/ZkUtils.scala b/core/src/main/scala/kafka/utils/ZkUtils.scala
index 2caafdcd18011..0c16243454198 100644
--- a/core/src/main/scala/kafka/utils/ZkUtils.scala
+++ b/core/src/main/scala/kafka/utils/ZkUtils.scala
@@ -143,7 +143,8 @@ object ZkUtils {
     DeleteTopicsPath + "/" + topic
 
   def parsePartitionReassignmentData(jsonData: String): Map[TopicAndPartition, Seq[Int]] = {
-    val assignments = ReassignPartitionsZNode.decode(jsonData.getBytes) match {
+    val utf8Bytes = jsonData.getBytes(StandardCharsets.UTF_8)
+    val assignments = ReassignPartitionsZNode.decode(utf8Bytes) match {
       case Left(e) => throw e
       case Right(result) => result
     }

From 5b2c54cd53c46f5b91e6a1102b70508d22b9607c Mon Sep 17 00:00:00 2001
From: Nicholas Travers <nickt@squareup.com>
Date: Fri, 5 Jan 2018 14:45:38 -0800
Subject: [PATCH 10/12] [fixup] Add missing license

---
 .../kafka/zk/ReassignPartitionsZNodeTest.scala   | 16 ++++++++++++++++
 1 file changed, 16 insertions(+)

diff --git a/core/src/test/scala/unit/kafka/zk/ReassignPartitionsZNodeTest.scala b/core/src/test/scala/unit/kafka/zk/ReassignPartitionsZNodeTest.scala
index fbd7f3602980b..94ed9fed73cd9 100644
--- a/core/src/test/scala/unit/kafka/zk/ReassignPartitionsZNodeTest.scala
+++ b/core/src/test/scala/unit/kafka/zk/ReassignPartitionsZNodeTest.scala
@@ -1,3 +1,19 @@
+/**
+  * Licensed to the Apache Software Foundation (ASF) under one or more
+  * contributor license agreements.  See the NOTICE file distributed with
+  * this work for additional information regarding copyright ownership.
+  * The ASF licenses this file to You under the Apache License, Version 2.0
+  * (the "License"); you may not use this file except in compliance with
+  * the License.  You may obtain a copy of the License at
+  *
+  *    http://www.apache.org/licenses/LICENSE-2.0
+  *
+  * Unless required by applicable law or agreed to in writing, software
+  * distributed under the License is distributed on an "AS IS" BASIS,
+  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  * See the License for the specific language governing permissions and
+  * limitations under the License.
+  */
 package unit.kafka.zk
 
 import com.fasterxml.jackson.core.JsonProcessingException

From ec7f36d94f07d5004633660fcf3fb81400f3ae02 Mon Sep 17 00:00:00 2001
From: Nicholas Travers <nickt@squareup.com>
Date: Mon, 19 Mar 2018 20:33:42 -0700
Subject: [PATCH 11/12] [fixup] Revert unnecessary newline removal

---
 core/src/test/scala/unit/kafka/utils/ZkUtilsTest.scala | 1 -
 1 file changed, 1 deletion(-)

diff --git a/core/src/test/scala/unit/kafka/utils/ZkUtilsTest.scala b/core/src/test/scala/unit/kafka/utils/ZkUtilsTest.scala
index 01a8ffa30ebd0..292db8b88958e 100755
--- a/core/src/test/scala/unit/kafka/utils/ZkUtilsTest.scala
+++ b/core/src/test/scala/unit/kafka/utils/ZkUtilsTest.scala
@@ -129,5 +129,4 @@ class ZkUtilsTest extends ZooKeeperTestHarness {
       assertEquals(seqid, zkUtils.getSequenceId(path))
     }
   }
-
 }

From 019ab707506fcb178d2b595bb67eb258da1d1bff Mon Sep 17 00:00:00 2001
From: Ismael Juma <ismael@juma.me.uk>
Date: Sat, 31 Mar 2018 00:21:30 -0700
Subject: [PATCH 12/12] Minor fixes

---
 core/src/main/scala/kafka/utils/Json.scala    |  4 +--
 .../scala/unit/kafka/utils/JsonTest.scala     |  2 +-
 .../zk/ReassignPartitionsZNodeTest.scala      | 25 ++++++++-----------
 3 files changed, 13 insertions(+), 18 deletions(-)

diff --git a/core/src/main/scala/kafka/utils/Json.scala b/core/src/main/scala/kafka/utils/Json.scala
index 486b05a8ec25e..c61e1490448b1 100644
--- a/core/src/main/scala/kafka/utils/Json.scala
+++ b/core/src/main/scala/kafka/utils/Json.scala
@@ -48,7 +48,7 @@ object Json {
 
   /**
    * Parse a JSON string into either a generic type T, or a JsonProcessingException in the case of
-    * exception.
+   * exception.
    */
   def parseStringAs[T](input: String)(implicit tag: ClassTag[T]): Either[JsonProcessingException, T] = {
     try Right(mapper.readValue(input, tag.runtimeClass).asInstanceOf[T])
@@ -67,7 +67,7 @@ object Json {
     catch { case e: JsonProcessingException => Left(e) }
 
   /**
-   * Parse a JSON string into either a generic type T, or a JsonProcessingException in the case of exception.
+   * Parse a JSON byte array into either a generic type T, or a JsonProcessingException in the case of exception.
    */
   def parseBytesAs[T](input: Array[Byte])(implicit tag: ClassTag[T]): Either[JsonProcessingException, T] = {
     try Right(mapper.readValue(input, tag.runtimeClass).asInstanceOf[T])
diff --git a/core/src/test/scala/unit/kafka/utils/JsonTest.scala b/core/src/test/scala/unit/kafka/utils/JsonTest.scala
index 3df7040f1a124..209fdee2edc31 100644
--- a/core/src/test/scala/unit/kafka/utils/JsonTest.scala
+++ b/core/src/test/scala/unit/kafka/utils/JsonTest.scala
@@ -145,7 +145,7 @@ class JsonTest {
   }
 
   @Test
-  def testParseTo_invalidJson() = {
+  def testParseToWithInvalidJson() = {
     val result = Json.parseStringAs[TestObject]("{invalid json}")
 
     assertTrue(result.isLeft)
diff --git a/core/src/test/scala/unit/kafka/zk/ReassignPartitionsZNodeTest.scala b/core/src/test/scala/unit/kafka/zk/ReassignPartitionsZNodeTest.scala
index 94ed9fed73cd9..2f3456f4b0d0b 100644
--- a/core/src/test/scala/unit/kafka/zk/ReassignPartitionsZNodeTest.scala
+++ b/core/src/test/scala/unit/kafka/zk/ReassignPartitionsZNodeTest.scala
@@ -14,38 +14,34 @@
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
-package unit.kafka.zk
+package kafka.zk
+
+import java.nio.charset.StandardCharsets
 
 import com.fasterxml.jackson.core.JsonProcessingException
-import kafka.zk.ReassignPartitionsZNode
 import org.apache.kafka.common.TopicPartition
 import org.junit.Assert._
 import org.junit.Test
 
 class ReassignPartitionsZNodeTest {
 
-  val topic = "foo"
-  val partition1 = 0
-  val replica1 = 1
-  val replica2 = 2
+  private val topic = "foo"
+  private val partition1 = 0
+  private val replica1 = 1
+  private val replica2 = 2
 
-  private val reassignPartitionData = Map(
-    new TopicPartition(topic, partition1) -> Seq(replica1, replica2))
-  private val reassignmentJson = "{\"version\":1,\"partitions\":[{\"topic\":\"foo\",\"partition\":0,\"replicas\":[1,2]}]}"
+  private val reassignPartitionData = Map(new TopicPartition(topic, partition1) -> Seq(replica1, replica2))
+  private val reassignmentJson = """{"version":1,"partitions":[{"topic":"foo","partition":0,"replicas":[1,2]}]}"""
 
   @Test
   def testEncode() {
-    val encodedJsonString = ReassignPartitionsZNode.encode(reassignPartitionData)
-      .map(_.toChar)
-      .mkString
-
+    val encodedJsonString = new String(ReassignPartitionsZNode.encode(reassignPartitionData), StandardCharsets.UTF_8)
     assertEquals(reassignmentJson, encodedJsonString)
   }
 
   @Test
   def testDecodeInvalidJson() {
     val result = ReassignPartitionsZNode.decode("invalid json".getBytes)
-
     assertTrue(result.isLeft)
     assertTrue(result.left.get.isInstanceOf[JsonProcessingException])
   }
@@ -53,7 +49,6 @@ class ReassignPartitionsZNodeTest {
   @Test
   def testDecodeValidJson() {
     val result = ReassignPartitionsZNode.decode(reassignmentJson.getBytes)
-
     assertTrue(result.isRight)
     val assignmentMap = result.right.get
     assertEquals(Seq(replica1, replica2), assignmentMap(new TopicPartition(topic, partition1)))