KAFKA-6074 Use ZookeeperClient in ReplicaManager and Partition #4254

tedyu · 2017-11-22T22:30:08Z

Replace ZkUtils with KafkaZkClient in ReplicaManager and Partition

Utilize existing unit tests

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

junrao

@tedyu : Thanks for the patch. A few comments below.

junrao · 2017-11-23T00:20:00Z

core/src/main/scala/kafka/zk/KafkaZkClient.scala

+  def createSequentialPersistentPath(path: String, data: String = ""): String = {
+    val createRequest = CreateRequest(path, data.getBytes("UTF-8"), acls(path), CreateMode.PERSISTENT_SEQUENTIAL)
+    val createResponse = retryRequestUntilConnected(createRequest)
+    if (createResponse.resultCode != Code.OK) {


We don't really need the if statement.

I see similar code in createConfigChangeNotification() on line 272.
Can you clarify how the if statement can be skipped ?

We just need createResponse.resultException.foreach(e => throw e) since createResponse.resultException is None only when createResponse.resultCode is OK.

Mani will have a separate PR to make existing usage consistent.

junrao · 2017-11-23T00:23:30Z

core/src/main/scala/kafka/zk/KafkaZkClient.scala

+    createResponse.path
+  }
+
+  def leaderAndIsrZkData(leaderAndIsr: LeaderAndIsr, controllerEpoch: Int): String = {


This method doesn't access ZK. So, it should be in ZkData.scala.

I looked at ZkData.scala where there are bunch of objects.
Not sure whether this method fits there.

We could just add a new LeaderAndIsr object and put the method there.

junrao · 2017-11-23T00:28:56Z

core/src/main/scala/kafka/admin/AdminUtils.scala

+   * Read the entity (topic, broker, client, user or <user, client>) config (if any) from zk
+   * sanitizedEntityName is <topic>, <broker>, <client-id>, <user> or <user>/clients/<client-id>.
+   */
+  def fetchEntityConfigZkClient(zkClient: KafkaZkClient, rootEntityType: String, sanitizedEntityName: String): Properties = {


This already exists in AdminZkClient.fetchEntityConfig().

That method doesn't take KafkaZkClient
Let me take another look.

AdminZkClient.fetchEntityConfig() take zkUtils.
I want to see if KafkaZkClient can be used.

Trying to figure out how to make ReplicaManagerTest pass.

AdminZkClient takes KafkaZkClient in the constructor.

Pardon.
AdminZkClient is not referenced by Partition or ReplicaManagerTest.
Do you mind explaining a bit more ?

As Jun said, there's no need to add this method. AdminZkClient.fetchEntityConfig should be used instead. You said that the method did not take a KafkaZkClient, so I explained that the constructor of the class did. Can you clarify what is unclear?

junrao · 2017-11-23T00:36:27Z

core/src/main/scala/kafka/utils/ReplicationUtils.scala

    try {
-      val writtenLeaderAndIsrInfo = zkUtils.readDataMaybeNull(path)
+      val writtenLeaderAndIsrInfo = zkClient.readDataMaybeNull(path)


This can be val (writtenLeaderOpt, writtenStat) = zkClient.readDataMaybeNull(path)

junrao · 2017-11-23T00:38:14Z

core/src/main/scala/kafka/utils/LogDirUtils.scala

 import scala.collection.Map

 object LogDirUtils extends Logging {

  private val LogDirEventNotificationPrefix = "log_dir_event_"
  val LogDirFailureEvent = 1

-  def propagateLogDirEvent(zkUtils: ZkUtils, brokerId: Int) {
-    val logDirEventNotificationPath: String = zkUtils.createSequentialPersistentPath(
+  def propagateLogDirEvent(zkClient: KafkaZkClient, brokerId: Int) {


Perhaps we could delete the other two methods in this class since they don't seem to be used.

junrao · 2017-11-23T00:40:15Z

core/src/main/scala/kafka/zk/KafkaZkClient.scala

@@ -49,6 +49,32 @@ import scala.collection.{Seq, mutable}
 class KafkaZkClient(zooKeeperClient: ZooKeeperClient, isSecure: Boolean) extends Logging {
  import KafkaZkClient._

+  def createSequentialPersistentPath(path: String, data: String = ""): String = {


Could we add a unit test for the new methods added in this class in KafkaZkClientTest?

Seems like this still needs to be addressed.

tedyu · 2017-11-23T05:22:40Z

I am thinking of passing zkClient to TopicPartition ctor.
But TopicPartition is public:

public final class TopicPartition implements Serializable {

ijuma

Thanks for the PR, a couple of comments.

ijuma · 2017-11-23T14:02:27Z

core/src/main/scala/kafka/zk/KafkaZkClient.scala

+    createResponse.path
+  }
+
+  def readDataMaybeNull(path: String): (Option[String], Stat) = {


This looks similar to getDataAndStat.

Are you going to remove this? You can just use getDataAndStat.

ijuma · 2017-11-23T14:03:17Z

core/src/main/scala/kafka/zk/ZkData.scala

@@ -30,6 +30,13 @@ import org.apache.zookeeper.data.Stat

 // This file contains objects for encoding/decoding data stored in ZooKeeper nodes (znodes).

+object LeaderAndIsrZNode {
+  def encode(leaderAndIsr: LeaderAndIsr, controllerEpoch: Int): String = {


Other encode methods return Array[Byte], why does this return a String?

This method calls Json.encode().
The String return type was from zkUtils.leaderAndIsrZkData().

The point is that we should aim to be consistent and return Array[Byte] from all the encode methods.

String is expected for argument 2 of:

def conditionalUpdatePath(path: String, data: String, expectVersion: Int,

If return value is changed to Array[Byte], wouldn't that require converting the output from Json.encode() to Array[Byte] and back to String ?
Please confirm.

You should change data in that method to be of type Array[Byte].

The parameter meanders thru checker and lands as 1st parameter here:

private def parseLeaderAndIsr(leaderAndIsrStr: String, path: String, stat: Stat): Option[LeaderIsrAndControllerEpoch] = { Json.parseFull(leaderAndIsrStr).flatMap { js =>

This would involve back and forth conversion which seems unnecessary.

That code can be changed to be Json.parseBytes. You can take a look at the other code, it's all following the same pattern.

When making the change, I found I need to overcome the following:

https://pastebin.com/rphaH3eD

Can you confirm this is desirable ?

I think it's desirable, but maybe we can leave it for another PR in that case.

Yeah. The scope is bigger than I thought.

omkreddy · 2017-11-23T15:02:06Z

core/src/main/scala/kafka/cluster/Partition.scala

@@ -55,7 +55,7 @@ class Partition(val topic: String,
  // Do not use replicaManager if this partition is ReplicaManager.OfflinePartition
  private val localBrokerId = if (!isOffline) replicaManager.config.brokerId else -1
  private val logManager = if (!isOffline) replicaManager.logManager else null
-  private val zkUtils = if (!isOffline) replicaManager.zkUtils else null
+  private val zkUtils = if (!isOffline) replicaManager.zkClient else null
  // allReplicasMap includes both assigned replicas and the future replica if there is ongoing replica movement


We can create AdminZkClient instance and use here.
Similar usage for reference:
https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/server/AdminManager.scala#L50

omkreddy · 2017-11-23T15:43:33Z

core/src/main/scala/kafka/utils/LogDirUtils.scala

-  def propagateLogDirEvent(zkUtils: ZkUtils, brokerId: Int) {
-    val logDirEventNotificationPath: String = zkUtils.createSequentialPersistentPath(
+  def propagateLogDirEvent(zkClient: KafkaZkClient, brokerId: Int) {
+    val logDirEventNotificationPath: String = zkClient.createSequentialPersistentPath(


we need to add this method to KafkaZkClient. This is similar imilar to KafkaZkClient.createAclChangeNotification method.
Related zk nodes/paths are : ZkData.LogDirEventNotificationZNode, ZkData.LogDirEventNotificationSequenceZNode

ijuma · 2017-11-23T15:46:08Z

core/src/main/scala/kafka/cluster/Partition.scala

@@ -55,7 +56,7 @@ class Partition(val topic: String,
  // Do not use replicaManager if this partition is ReplicaManager.OfflinePartition
  private val localBrokerId = if (!isOffline) replicaManager.config.brokerId else -1
  private val logManager = if (!isOffline) replicaManager.logManager else null
-  private val zkUtils = if (!isOffline) replicaManager.zkUtils else null
+  private val zkUtils = if (!isOffline) replicaManager.zkClient else null


This should be renamed to zkClient

omkreddy · 2017-11-23T15:49:10Z

core/src/test/scala/unit/kafka/admin/TestAdminUtils.scala

@@ -18,6 +18,7 @@ package kafka.admin

 import java.util.Properties
 import kafka.utils.ZkUtils
+import kafka.zk.KafkaZkClient



unused import?

omkreddy · 2017-11-23T15:50:06Z

core/src/test/scala/unit/kafka/server/HighwatermarkPersistenceTest.scala

@@ -35,6 +36,7 @@ class HighwatermarkPersistenceTest {
  val configs = TestUtils.createBrokerConfigs(2, TestUtils.MockZkConnect).map(KafkaConfig.fromProps)
  val topic = "foo"
  val zkUtils = EasyMock.createMock(classOf[ZkUtils])


Do we need zkUtils now?

omkreddy · 2017-11-23T15:51:45Z

core/src/test/scala/unit/kafka/server/ReplicaManagerQuotasTest.scala

@@ -149,6 +150,7 @@ class ReplicaManagerQuotasTest {

  def setUpMocks(fetchInfo: Seq[(TopicPartition, PartitionData)], record: SimpleRecord = this.record, bothReplicasInSync: Boolean = false) {
    val zkUtils = createNiceMock(classOf[ZkUtils])


Do we need zkUtils now?

omkreddy · 2017-11-23T15:52:24Z

core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala

@@ -49,10 +50,14 @@ class ReplicaManagerTest {
  val metrics = new Metrics
  var zkClient: ZkClient = _
  var zkUtils: ZkUtils = _


Do we need zkUtils now?

omkreddy · 2017-11-23T15:53:10Z

core/src/test/scala/unit/kafka/server/SimpleFetchTest.scala

@@ -70,6 +71,7 @@ class SimpleFetchTest {
  def setUp() {
    // create nice mock since we don't particularly care about zkclient calls
    val zkUtils = EasyMock.createNiceMock(classOf[ZkUtils])


tedyu · 2017-11-23T15:56:34Z

Thanks for the reviews.

Update coming soon.

I would be out of town next week. So trying to get some work done.

tedyu · 2017-11-23T17:33:47Z

Failure of SaslSslAdminClientIntegrationTest.testLogStartOffsetCheckpoint was not related to the PR.

omkreddy · 2017-11-23T17:36:43Z

core/src/main/scala/kafka/admin/AdminUtils.scala

@@ -21,6 +21,7 @@ import kafka.log.LogConfig
 import kafka.server.{ConfigEntityName, ConfigType, DynamicConfig}
 import kafka.utils._
 import kafka.utils.ZkUtils._
+import kafka.zk.KafkaZkClient


unused import?

omkreddy · 2017-11-23T17:39:33Z

core/src/main/scala/kafka/server/ReplicaFetcherThread.scala

@@ -26,6 +26,7 @@ import kafka.cluster.{BrokerEndPoint, Replica}
 import kafka.log.LogConfig


unused import?

LogConfig is used

omkreddy · 2017-11-23T18:03:00Z

core/src/main/scala/kafka/zk/KafkaZkClient.scala

+    Json.encode(Map("version" -> LogDirEventNotificationHandler.Version, "broker" -> brokerId, "event" -> LogDirFailureEvent))
+  }
+
+  def propagateLogDirEvent(brokerId: Int) {


hmm..This method should be like createConfigChangeNotification, createAclChangeNotification methods.
and should use existing ZkData.LogDirEventNotificationSequenceZNode object.

omkreddy · 2017-11-23T18:05:20Z

core/src/main/scala/kafka/zk/KafkaZkClient.scala

+
+  def propagateLogDirEvent(brokerId: Int) {
+    val logDirEventNotificationPath: String = createSequentialPersistentPath(
+      ZkUtils.LogDirEventNotificationPath + "/log_dir_event_", logDirFailureEventZkData(brokerId))


we should use LogDirEventNotificationZNode.path, LogDirEventNotificationSequenceZNode.SequenceNumberPrefix constants here

omkreddy · 2017-11-23T18:06:53Z

core/src/main/scala/kafka/zk/KafkaZkClient.scala

@@ -933,6 +953,17 @@ class KafkaZkClient(zooKeeperClient: ZooKeeperClient, isSecure: Boolean) extends
    createResponse.resultException.foreach(e => throw e)
  }

+  private def logDirFailureEventZkData(brokerId: Int): String = {
+    val LogDirFailureEvent = 1
+    Json.encode(Map("version" -> LogDirEventNotificationHandler.Version, "broker" -> brokerId, "event" -> LogDirFailureEvent))


This is already available in ZkData.LogDirEventNotificationSequenceZNode.encode method

tedyu · 2017-11-23T23:59:49Z

Java 8 build failed with:

org.apache.kafka.connect.data.SchemaBuilderTest > testFloatBuilder STARTED
FATAL: command execution failed

Not related to my PR.

ijuma

LGTM from a correctness perspective, but it's missing a test and can be simplified. Will merge to trunk and submit a follow-up.

ijuma · 2017-11-24T11:13:01Z

core/src/main/scala/kafka/zk/KafkaZkClient.scala

@@ -1159,7 +1174,7 @@ class KafkaZkClient(zooKeeperClient: ZooKeeperClient, isSecure: Boolean) extends
    }
  }

-  private[zk] def pathExists(path: String): Boolean = {
+  def pathExists(path: String): Boolean = {


This can be reverted.

tedyu · 2017-11-24T11:33:25Z

Thanks for the review, @ijuma @omkreddy @junrao

Happy Holiday

tedyu added 3 commits November 22, 2017 14:27

KAFKA-6074 Use ZookeeperClient in ReplicaManager and Partition

f4991b2

Try to fix compilation error in tests

19d19e0

Add fetchEntityConfigZkClient to TestAdminUtils

c716bf6

junrao reviewed Nov 23, 2017

View reviewed changes

tedyu added 2 commits November 22, 2017 17:38

Address Jun's review comments

e833069

Address Jun's comments

980f145

ijuma reviewed Nov 23, 2017

View reviewed changes

omkreddy reviewed Nov 23, 2017

View reviewed changes

tedyu added 2 commits November 23, 2017 07:07

Remove fetchEntityConfigZkClient

5c394c5

use EasyMock for kafkaZkClient

46eff85

omkreddy reviewed Nov 23, 2017

View reviewed changes

ijuma reviewed Nov 23, 2017

View reviewed changes

omkreddy reviewed Nov 23, 2017

View reviewed changes

Address Manikumar's comments

eec90d6

omkreddy reviewed Nov 23, 2017

View reviewed changes

Address Manikumar's comments

da39abf

tedyu added 2 commits November 23, 2017 16:07

Fix failed LogDirFailureTest

19dfe00

Drop readDataMaybeNull

d54f03b

ijuma approved these changes Nov 24, 2017

View reviewed changes

asfgit closed this in 0bc2d0e Nov 24, 2017

ijuma mentioned this pull request Dec 28, 2017

KAFKA-6074 Use ZookeeperClient in ReplicaManager and Partition #4124

Closed

		@@ -149,6 +150,7 @@ class ReplicaManagerQuotasTest {

		def setUpMocks(fetchInfo: Seq[(TopicPartition, PartitionData)], record: SimpleRecord = this.record, bothReplicasInSync: Boolean = false) {
		val zkUtils = createNiceMock(classOf[ZkUtils])

		@@ -26,6 +26,7 @@ import kafka.cluster.{BrokerEndPoint, Replica}
		import kafka.log.LogConfig

KAFKA-6074 Use ZookeeperClient in ReplicaManager and Partition #4254

KAFKA-6074 Use ZookeeperClient in ReplicaManager and Partition #4254

Conversation

tedyu commented Nov 22, 2017

Committer Checklist (excluded from commit message)

junrao left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tedyu Nov 23, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tedyu commented Nov 23, 2017 • edited Loading

ijuma left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tedyu commented Nov 23, 2017

tedyu commented Nov 23, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tedyu commented Nov 23, 2017

ijuma left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tedyu commented Nov 24, 2017

tedyu Nov 23, 2017 •

edited

Loading

tedyu commented Nov 23, 2017 •

edited

Loading