KAFKA-6074 Use ZookeeperClient in ReplicaManager and Partition #4166
Conversation
|
See if changes in ReassignPartitionsCommand are on right track. Alternative is to use ZkUtils and KafkaZkClient in ReassignPartitionsCommand. |
|
I suggest to change ReplicaManager.scala, LogDirUtils.scala, ReplicationUtils.scala classes as part of this PR. ReassignPartitionsCommand can be done in another JIRA/PR. I am doing few changes to AdminUtils as part of KAFKA-5646. We can reuse those changes for Admin Operations/KAFKA-5647. |
|
FAILURE |
2 similar comments
|
FAILURE |
|
FAILURE |
|
FAILURE |
2 similar comments
|
FAILURE |
|
FAILURE |
|
40 tests extend ZooKeeperTestHarness where {{var zkUtils: ZkUtils}} is declared. Suggestion on how to phase out ZkUtils is welcome. |
| def createSequentialPersistentPath(path: String, data: String = ""): String = { | ||
| val createRequest = CreateRequest(path, data.getBytes("UTF-8"), acls(path), CreateMode.PERSISTENT_SEQUENTIAL) | ||
| val createResponse = retryRequestUntilConnected(createRequest) | ||
| createResponse.path |
There was a problem hiding this comment.
We should throw an exception if the result code is not OK.
| def leaderAndIsrZkData(leaderAndIsr: LeaderAndIsr, controllerEpoch: Int): String = { | ||
| Json.encode(Map("version" -> 1, "leader" -> leaderAndIsr.leader, "leader_epoch" -> leaderAndIsr.leaderEpoch, | ||
| "controller_epoch" -> controllerEpoch, "isr" -> leaderAndIsr.isr)) | ||
| } |
There was a problem hiding this comment.
This doesn't access zookeeperClient and should be moved to ZkData.scala.
| } else { | ||
| val data = new String(getDataResponse.data, UTF_8) | ||
| (Some(data), getDataResponse.stat) | ||
| } |
There was a problem hiding this comment.
We should throw an exception in the result code is not OK.
| "controller_epoch" -> controllerEpoch, "isr" -> leaderAndIsr.isr)) | ||
| } | ||
|
|
||
| def readDataMaybeNull(path: String): (Option[String], Stat) = { |
There was a problem hiding this comment.
With this, methods like getPreferredReplicaElection() and getControllerId, etc could potentially be simplified by reusing this method.
| @@ -48,6 +48,29 @@ import scala.collection.mutable.ArrayBuffer | |||
| class KafkaZkClient(zooKeeperClient: ZooKeeperClient, isSecure: Boolean) extends Logging { | |||
| import KafkaZkClient._ | |||
|
|
|||
| def createSequentialPersistentPath(path: String, data: String = ""): String = { | |||
There was a problem hiding this comment.
Could we add a unit test for each new public method introduced in this class?
|
@junrao I was waiting for some advice on how the related tests should be handled so that compilation passes. |
|
@tedyu I am migrating few utility methods as part of KAFKA-5646 and 5647. |
|
@tedyu : KAFKA-5646 is now committed. You can rebase your patch now. |
No description provided.