Skip to content

KAFKA-5919: Adding checks on "version" field for tools using it #5126

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
ppatierno wants to merge 3 commits into from
Closed

KAFKA-5919: Adding checks on "version" field for tools using it #5126

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e5edd21
Adding checks on "version" field for tools using it
ppatierno Jun 2, 2018
9aebf42
Removed blank lines
ppatierno Jun 3, 2018
264d90b
Removed brackets
ppatierno Jun 4, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 31 additions & 9 deletions core/src/main/scala/kafka/admin/DeleteRecordsCommand.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,31 +22,53 @@ import java.util.Properties

import kafka.admin.AdminClient.DeleteRecordsResult
import kafka.common.AdminCommandFailedException
import kafka.utils.{CoreUtils, Json, CommandLineUtils}
import kafka.utils.{CommandLineUtils, CoreUtils, Json}
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.utils.Utils
import org.apache.kafka.clients.CommonClientConfigs
import joptsimple._
import kafka.utils.json.JsonValue

/**
* A command for delete records of the given partitions down to the specified offset.
*/
object DeleteRecordsCommand {

private[admin] val EarliestVersion = 1

def main(args: Array[String]): Unit = {
execute(args, System.out)
}

def parseOffsetJsonStringWithoutDedup(jsonData: String): Seq[(TopicPartition, Long)] = {
Json.parseFull(jsonData).toSeq.flatMap { js =>
js.asJsonObject.get("partitions").toSeq.flatMap { partitionsJs =>
partitionsJs.asJsonArray.iterator.map(_.asJsonObject).map { partitionJs =>
val topic = partitionJs("topic").to[String]
val partition = partitionJs("partition").to[Int]
val offset = partitionJs("offset").to[Long]
new TopicPartition(topic, partition) -> offset
}.toBuffer
Json.parseFull(jsonData) match {

@lindong28 lindong28 Jun 3, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nits: can you remove the { here?

@ppatierno ppatierno Jun 3, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I searched for other places where match is used at it seems that in those places, the { is always used. Or am I wrong?

@lindong28 lindong28 Jun 3, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah, I comment in the wrong line. I actually mean the line below.

More specifically,

      case Some(js) => {
        val version = js.asJsonObject.get("version") match {
          case Some(jsonValue) => jsonValue.to[Int]
          case None => EarliestVersion
        }
        parseJsonData(version, js)
      }

can be:

      case Some(js) =>
        val version = js.asJsonObject.get("version") match {
          case Some(jsonValue) => jsonValue.to[Int]
          case None => EarliestVersion
        }
        parseJsonData(version, js)

@lindong28 lindong28 Jun 3, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For example, we don't use { after all usage of case in LogManager.scala.

@ppatierno ppatierno Jun 4, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah you are right. Done!

case Some(js) => {

@lindong28 lindong28 Jun 3, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nits: can you remove the { here so that we following the existing code style? There are a few other places where we can reduce { and } similarly. Also, could we also follow the existing code style and remove some empty lines?

val version = js.asJsonObject.get("version") match {
case Some(jsonValue) => jsonValue.to[Int]
case None => EarliestVersion
}
parseJsonData(version, js)
}
case None => throw new AdminOperationException("The input string is not a valid JSON")
}
}

def parseJsonData(version: Int, js: JsonValue): Seq[(TopicPartition, Long)] = {
version match {
case 1 => {
js.asJsonObject.get("partitions") match {
case Some(partitions) => {
partitions.asJsonArray.iterator.map(_.asJsonObject).map { partitionJs =>
val topic = partitionJs("topic").to[String]
val partition = partitionJs("partition").to[Int]
val offset = partitionJs("offset").to[Long]
new TopicPartition(topic, partition) -> offset
}.toBuffer
}
case _ => throw new AdminOperationException("Missing partitions field");
}
}
case _ => throw new AdminOperationException(s"Not supported version field value $version")
}
}

Expand Down
93 changes: 69 additions & 24 deletions core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ import kafka.log.LogConfig
import kafka.log.LogConfig._
import kafka.server.{ConfigType, DynamicConfig}
import kafka.utils._
import kafka.utils.json.JsonValue
import kafka.zk.{AdminZkClient, KafkaZkClient}
import org.apache.kafka.clients.admin.DescribeReplicaLogDirsResult.ReplicaLogDirInfo
import org.apache.kafka.clients.admin.{AdminClientConfig, AlterReplicaLogDirsOptions, AdminClient => JAdminClient}
Expand All @@ -45,6 +46,8 @@ object ReassignPartitionsCommand extends Logging {
private[admin] val NoThrottle = Throttle(-1, -1)
private[admin] val AnyLogDir = "any"

private[admin] val EarliestVersion = 1

def main(args: Array[String]): Unit = {
val opts = validateAndParseArgs(args)
val zkConnect = opts.options.valueOf(opts.zkConnectOpt)
Expand Down Expand Up @@ -168,7 +171,7 @@ object ReassignPartitionsCommand extends Logging {
}

def generateAssignment(zkClient: KafkaZkClient, brokerListToReassign: Seq[Int], topicsToMoveJsonString: String, disableRackAware: Boolean): (Map[TopicPartition, Seq[Int]], Map[TopicPartition, Seq[Int]]) = {
val topicsToReassign = ZkUtils.parseTopicsData(topicsToMoveJsonString)
val topicsToReassign = parseTopicsData(topicsToMoveJsonString)
val duplicateTopicsToReassign = CoreUtils.duplicates(topicsToReassign)
if (duplicateTopicsToReassign.nonEmpty)
throw new AdminCommandFailedException("List of topics to reassign contains duplicate entries: %s".format(duplicateTopicsToReassign.mkString(",")))
Expand Down Expand Up @@ -241,32 +244,74 @@ object ReassignPartitionsCommand extends Logging {
).asJava)
}

// Parses without deduplicating keys so the data can be checked before allowing reassignment to proceed
def parseTopicsData(jsonData: String): Seq[String] = {
Json.parseFull(jsonData) match {
case Some(js) => {
val version = js.asJsonObject.get("version") match {
case Some(jsonValue) => jsonValue.to[Int]
case None => EarliestVersion
}
parseTopicsData(version, js)
}
case None => throw new AdminOperationException("The input string is not a valid JSON")
}
}

def parseTopicsData(version: Int, js: JsonValue): Seq[String] = {
version match {
case 1 => {
for {
partitionsSeq <- js.asJsonObject.get("topics").toSeq
p <- partitionsSeq.asJsonArray.iterator
} yield p.asJsonObject("topic").to[String]
}
case _ => throw new AdminOperationException(s"Not supported version field value $version")
}
}

def parsePartitionReassignmentData(jsonData: String): (Seq[(TopicPartition, Seq[Int])], Map[TopicPartitionReplica, String]) = {
val partitionAssignment = mutable.ListBuffer.empty[(TopicPartition, Seq[Int])]
val replicaAssignment = mutable.Map.empty[TopicPartitionReplica, String]
for {
js <- Json.parseFull(jsonData).toSeq
partitionsSeq <- js.asJsonObject.get("partitions").toSeq
p <- partitionsSeq.asJsonArray.iterator
} {
val partitionFields = p.asJsonObject
val topic = partitionFields("topic").to[String]
val partition = partitionFields("partition").to[Int]
val newReplicas = partitionFields("replicas").to[Seq[Int]]
val newLogDirs = partitionFields.get("log_dirs") match {
case Some(jsonValue) => jsonValue.to[Seq[String]]
case None => newReplicas.map(_ => AnyLogDir)
Json.parseFull(jsonData) match {
case Some(js) => {
val version = js.asJsonObject.get("version") match {
case Some(jsonValue) => jsonValue.to[Int]
case None => EarliestVersion
}
parsePartitionReassignmentData(version, js)
}
case None => throw new AdminOperationException("The input string is not a valid JSON")
}
}

// Parses without deduplicating keys so the data can be checked before allowing reassignment to proceed
def parsePartitionReassignmentData(version:Int, jsonData: JsonValue): (Seq[(TopicPartition, Seq[Int])], Map[TopicPartitionReplica, String]) = {
version match {
case 1 => {
val partitionAssignment = mutable.ListBuffer.empty[(TopicPartition, Seq[Int])]
val replicaAssignment = mutable.Map.empty[TopicPartitionReplica, String]
for {
partitionsSeq <- jsonData.asJsonObject.get("partitions").toSeq
p <- partitionsSeq.asJsonArray.iterator
} {
val partitionFields = p.asJsonObject
val topic = partitionFields("topic").to[String]
val partition = partitionFields("partition").to[Int]
val newReplicas = partitionFields("replicas").to[Seq[Int]]
val newLogDirs = partitionFields.get("log_dirs") match {
case Some(jsonValue) => jsonValue.to[Seq[String]]
case None => newReplicas.map(_ => AnyLogDir)
}
if (newReplicas.size != newLogDirs.size)
throw new AdminCommandFailedException(s"Size of replicas list $newReplicas is different from " +
s"size of log dirs list $newLogDirs for partition ${new TopicPartition(topic, partition)}")
partitionAssignment += (new TopicPartition(topic, partition) -> newReplicas)
replicaAssignment ++= newReplicas.zip(newLogDirs).map { case (replica, logDir) =>
new TopicPartitionReplica(topic, partition, replica) -> logDir
}.filter(_._2 != AnyLogDir)
}
(partitionAssignment, replicaAssignment)
}
if (newReplicas.size != newLogDirs.size)
throw new AdminCommandFailedException(s"Size of replicas list $newReplicas is different from " +
s"size of log dirs list $newLogDirs for partition ${new TopicPartition(topic, partition)}")
partitionAssignment += (new TopicPartition(topic, partition) -> newReplicas)
replicaAssignment ++= newReplicas.zip(newLogDirs).map { case (replica, logDir) =>
new TopicPartitionReplica(topic, partition, replica) -> logDir
}.filter(_._2 != AnyLogDir)
case _ => throw new AdminOperationException(s"Not supported version field value $version")
}
(partitionAssignment, replicaAssignment)
}

def parseAndValidate(zkClient: KafkaZkClient, reassignmentJsonString: String): (Seq[(TopicPartition, Seq[Int])], Map[TopicPartitionReplica, String]) = {
Expand Down
8 changes: 0 additions & 8 deletions core/src/main/scala/kafka/utils/ZkUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -152,14 +152,6 @@ object ZkUtils {
assignments.map { case (tp, p) => (new TopicAndPartition(tp), p) }
}

def parseTopicsData(jsonData: String): Seq[String] = {
for {
js <- Json.parseFull(jsonData).toSeq
partitionsSeq <- js.asJsonObject.get("topics").toSeq
p <- partitionsSeq.asJsonArray.iterator
} yield p.asJsonObject("topic").to[String]
}

def controllerZkData(brokerId: Int, timestamp: Long): String = {
Json.legacyEncodeAsString(Map("version" -> 1, "brokerid" -> brokerId, "timestamp" -> timestamp.toString))
}
Expand Down