Skip to content

KAFKA-5647: Use KafkaZkClient in ReassignPartitionsCommand and PreferredReplicaLeaderElectionCommand #4260

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
omkreddy wants to merge 3 commits into from
Closed

KAFKA-5647: Use KafkaZkClient in ReassignPartitionsCommand and PreferredReplicaLeaderElectionCommand #4260

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
fd5c172
KAFKA-5647: Use KafkaZkClient in ReassignPartitionsCommand and Prefer…
omkreddy Nov 17, 2017
29a2e1a
Address review comments
omkreddy Dec 8, 2017
4c15081
Remove TopicAndPartition usage from ReassignPartitionsCommand
omkreddy Dec 8, 2017
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
2 changes: 1 addition & 1 deletion checkstyle/import-control.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -223,10 +223,10 @@
<allow pkg="kafka.tools" />
<allow pkg="kafka.utils" />
<allow pkg="kafka.zk" />
<allow pkg="kafka.zookeeper" />
<allow pkg="kafka.log" />
<allow pkg="scala" />
<allow pkg="scala.collection" />
<allow pkg="org.I0Itec.zkclient" />
</subpackage>

<subpackage name="test">
Expand Down
56 changes: 27 additions & 29 deletions core/src/main/scala/kafka/admin/PreferredReplicaLeaderElectionCommand.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,12 +18,15 @@ package kafka.admin

import joptsimple.OptionParser
import kafka.utils._
import org.I0Itec.zkclient.ZkClient
import org.I0Itec.zkclient.exception.ZkNodeExistsException
import kafka.common.{TopicAndPartition, AdminCommandFailedException}
import kafka.common.AdminCommandFailedException
import kafka.zk.KafkaZkClient
import kafka.zookeeper.ZooKeeperClient

import collection._
import org.apache.kafka.common.utils.Utils
import org.apache.kafka.common.utils.{Time, Utils}
import org.apache.kafka.common.security.JaasUtils
import org.apache.kafka.common.TopicPartition
import org.apache.zookeeper.KeeperException.NodeExistsException

object PreferredReplicaLeaderElectionCommand extends Logging {

Expand Down Expand Up @@ -51,20 +54,19 @@ object PreferredReplicaLeaderElectionCommand extends Logging {
CommandLineUtils.checkRequiredArgs(parser, options, zkConnectOpt)

val zkConnect = options.valueOf(zkConnectOpt)
var zkClient: ZkClient = null
var zkUtils: ZkUtils = null
var zooKeeperClient: ZooKeeperClient = null
var zkClient: KafkaZkClient = null
try {
zkClient = ZkUtils.createZkClient(zkConnect, 30000, 30000)
zkUtils = ZkUtils(zkConnect,
30000,
30000,
JaasUtils.isZkSecurityEnabled())
val time = Time.SYSTEM
zooKeeperClient = new ZooKeeperClient(zkConnect, 30000, 30000, Int.MaxValue, time)
zkClient = new KafkaZkClient(zooKeeperClient, JaasUtils.isZkSecurityEnabled, time)

val partitionsForPreferredReplicaElection =
if (!options.has(jsonFileOpt))
zkUtils.getAllPartitions()
zkClient.getAllPartitions()
else
parsePreferredReplicaElectionData(Utils.readFileAsString(options.valueOf(jsonFileOpt)))
val preferredReplicaElectionCommand = new PreferredReplicaLeaderElectionCommand(zkUtils, partitionsForPreferredReplicaElection)
val preferredReplicaElectionCommand = new PreferredReplicaLeaderElectionCommand(zkClient, partitionsForPreferredReplicaElection)

preferredReplicaElectionCommand.moveLeaderToPreferredReplica()
} catch {
Expand All @@ -77,7 +79,7 @@ object PreferredReplicaLeaderElectionCommand extends Logging {
}
}

def parsePreferredReplicaElectionData(jsonString: String): immutable.Set[TopicAndPartition] = {
def parsePreferredReplicaElectionData(jsonString: String): immutable.Set[TopicPartition] = {
Json.parseFull(jsonString) match {
case Some(js) =>
js.asJsonObject.get("partitions") match {
Expand All @@ -86,7 +88,7 @@ object PreferredReplicaLeaderElectionCommand extends Logging {
val partitions = partitionsRaw.map { p =>
val topic = p("topic").to[String]
val partition = p("partition").to[Int]
TopicAndPartition(topic, partition)
new TopicPartition(topic, partition)
}.toBuffer
val duplicatePartitions = CoreUtils.duplicates(partitions)
if (duplicatePartitions.nonEmpty)
Expand All @@ -98,34 +100,30 @@ object PreferredReplicaLeaderElectionCommand extends Logging {
}
}

def writePreferredReplicaElectionData(zkUtils: ZkUtils,
partitionsUndergoingPreferredReplicaElection: scala.collection.Set[TopicAndPartition]) {
val zkPath = ZkUtils.PreferredReplicaLeaderElectionPath
val jsonData = ZkUtils.preferredReplicaLeaderElectionZkData(partitionsUndergoingPreferredReplicaElection)
def writePreferredReplicaElectionData(zkClient: KafkaZkClient,
partitionsUndergoingPreferredReplicaElection: Set[TopicPartition]) {
try {
zkUtils.createPersistentPath(zkPath, jsonData)
println("Created preferred replica election path with %s".format(jsonData))
zkClient.createPreferredReplicaElection(partitionsUndergoingPreferredReplicaElection.toSet)
println("Created preferred replica election path with %s".format(partitionsUndergoingPreferredReplicaElection.mkString(",")))
} catch {
case _: ZkNodeExistsException =>
val partitionsUndergoingPreferredReplicaElection =
PreferredReplicaLeaderElectionCommand.parsePreferredReplicaElectionData(zkUtils.readData(zkPath)._1)
case _: NodeExistsException =>
throw new AdminOperationException("Preferred replica leader election currently in progress for " +
"%s. Aborting operation".format(partitionsUndergoingPreferredReplicaElection))
"%s. Aborting operation".format(zkClient.getPreferredReplicaElection.mkString(",")))
case e2: Throwable => throw new AdminOperationException(e2.toString)
}
}
}

class PreferredReplicaLeaderElectionCommand(zkUtils: ZkUtils, partitionsFromUser: scala.collection.Set[TopicAndPartition]) {
class PreferredReplicaLeaderElectionCommand(zkClient: KafkaZkClient, partitionsFromUser: scala.collection.Set[TopicPartition]) {
def moveLeaderToPreferredReplica() = {
try {
val topics = partitionsFromUser.map(_.topic).toSet
val partitionsFromZk = zkUtils.getPartitionsForTopics(topics.toSeq).flatMap{ case (topic, partitions) =>
partitions.map(TopicAndPartition(topic, _))
val partitionsFromZk = zkClient.getPartitionsForTopics(topics).flatMap{ case (topic, partitions) =>
partitions.map(new TopicPartition(topic, _))
}.toSet

val (validPartitions, invalidPartitions) = partitionsFromUser.partition(partitionsFromZk.contains)
PreferredReplicaLeaderElectionCommand.writePreferredReplicaElectionData(zkUtils, validPartitions)
PreferredReplicaLeaderElectionCommand.writePreferredReplicaElectionData(zkClient, validPartitions)

println("Successfully started preferred replica election for partitions %s".format(validPartitions))
invalidPartitions.foreach(p => println("Skipping preferred replica leader election for partition %s since it doesn't exist.".format(p)))
Expand Down
Loading