KAFKA-6320: move ZK metrics in KafkaHealthCheck to ZookeeperClient #4351

junrao · 2017-12-21T22:44:37Z

Moved metrics in KafkaHealthCheck to ZookeeperClient.
Converted remaining ZkUtils usage in KafkaServer to ZookeeperClient and removed ZkUtils from KafkaServer.
Made the re-creation of ZooKeeper during ZK session expiration with infinite retries.
Added unit tests for all new methods in KafkaZkClient.

Committer Checklist (excluded from commit message)

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

junrao · 2017-12-21T22:47:39Z

Ping @ijuma @onurkaraman for review.

ijuma

Thanks for the PR, looks good overall, just a few minor questions/comments.

ijuma · 2017-12-21T22:58:26Z

core/src/main/scala/kafka/zookeeper/ZooKeeperClient.scala

-              now = System.currentTimeMillis()
-            }
+          case e: Exception =>
+            debug("Error when recreating ZooKeeper", e)


Would we want this to be info? It seems like it should be rare, right?

ijuma · 2017-12-21T23:01:06Z

core/src/main/scala/kafka/zookeeper/ZooKeeperClient.scala

+    * This is added to preserve the original metric name in JMX
+    */
+  override def metricName(name: String, metricTags: scala.collection.Map[String, String]): MetricName = {
+    explicitMetricName("kafka.server", "KafkaHealthcheck", name, metricTags)


Nit: maybe we can take a metricGroup and metricTypeName as parameters in the constructor to keep ZooKeeperClient generic.

Good point. Added.

ijuma · 2017-12-21T23:01:35Z

core/src/main/scala/kafka/zookeeper/ZooKeeperClient.scala

            error("Auth failed.")
            stateChangeHandlers.foreach {case (name, handler) => handler.onAuthFailure()}
-          } else if (event.getState == KeeperState.Expired) {
+          } else if (state == KeeperState.Expired) {
            inWriteLock(initializationLock) {
              info("Session expired.")
              stateChangeHandlers.foreach {case (name, handler) => handler.beforeInitializingSession()}


Nit: we can use values.foreach to simplify this. There are a few other cases where we do something similar in this method and initialize.

ijuma · 2017-12-21T23:04:34Z

core/src/test/scala/integration/kafka/api/MetricsTest.scala

-    assertEquals(s"Unexpected ZK state ${server.zkUtils.zkConnection.getZookeeperState}",
-        "CONNECTED", yammerMetricValue("SessionState"))
+    // Latency is rounded to milliseconds, so check the count instead.
+    val initialCount = yammerHistogramCount("kafka.server:type=ZooKeeperClientMetrics,name=ZooKeeperRequestLatencyMs").asInstanceOf[Long]


So, we don't actually need to wait until true here?

Right, since the ZK write in between is blocking and is guaranteed to update the count in the histogram.

Ah, I had missed that you're using the count now. Makes sense.

omkreddy · 2017-12-22T07:07:44Z

core/src/main/scala/kafka/coordinator/transaction/ProducerIdManager.scala

@@ -20,7 +20,7 @@ import java.nio.charset.StandardCharsets

 import kafka.common.KafkaException
 import kafka.utils.{Json, Logging, ZkUtils}


unused ZKUtils import

omkreddy · 2017-12-22T07:15:17Z

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

+  }
+
+  /**
+    * Generate a borker id by updating the broker sequence id path in ZK and return the version of the path.


nit: typo "borker"

omkreddy · 2017-12-22T08:26:10Z

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

@@ -31,6 +32,7 @@ import org.apache.kafka.common.serialization.{IntegerSerializer, StringSerialize
 import org.I0Itec.zkclient.exception.ZkException


unused import

ijuma

Thanks for the PR. LGTM. Will submit a separate PR for the comments below.

ijuma · 2017-12-22T09:51:50Z

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

@@ -455,9 +439,9 @@ class KafkaServer(val config: KafkaConfig, time: Time = Time.SYSTEM, threadNameP

          // Get the current controller info. This is to ensure we use the most recent info to issue the
          // controlled shutdown request
-          val controllerId = zkUtils.getController()
+          val controllerId = zkClient.getControllerId.getOrElse(throw new KafkaException("Controller doesn't exist"))


Should we ignore the missing controller and retry (like we ignore the missing broker info)?

ijuma · 2017-12-22T10:11:36Z

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

+    TopicsZNode.path,
+    ConfigEntityChangeNotificationZNode.path,
+    ConfigEntityTypeZNode.path(ConfigType.Topic),
+    ConfigEntityTypeZNode.path(ConfigType.Client),


Shouldn't we do this for Broker and User as well?

ijuma · 2017-12-22T10:13:06Z

core/src/main/scala/kafka/zookeeper/ZooKeeperClient.scala

-                      time: Time) extends Logging {
+                      time: Time,
+                      metricGroup: String = "kafka.server",
+                      metricType: String = "KafkaHealthcheck") extends Logging with KafkaMetricsGroup {


I think it would be better not to have the default at all.

ijuma · 2017-12-22T10:14:18Z

core/src/main/scala/kafka/zookeeper/ZooKeeperClient.scala

+      state -> newMeter(name, eventType.toLowerCase(Locale.ROOT), TimeUnit.SECONDS)
+    }
+  }
+
  info(s"Initializing a new session to $connectString.")
  @volatile private var zooKeeper = new ZooKeeper(connectString, sessionTimeoutMs, ZooKeeperClientWatcher)


Worth adding a comment stating that we intend to fail-fast during construction, but that we retry forever if this is reinitialised.

ijuma · 2017-12-22T10:15:36Z

core/src/main/scala/kafka/zookeeper/ZooKeeperClient.scala

  waitUntilConnected(connectionTimeoutMs, TimeUnit.MILLISECONDS)

+
+  /**
+    * This is added to preserve the original metric name in JMX


Since we're receiving this as a parameter now, we should tweak this comment.

ijuma · 2017-12-22T10:16:59Z

core/src/main/scala/kafka/zookeeper/ZooKeeperClient.scala

        }
      }
      info(s"Timed out waiting for connection during session initialization while in state: ${zooKeeper.getState}")
-      stateChangeHandlers.foreach {case (name, handler) => handler.onReconnectionTimeout()}
+      stateChangeHandlers.values.foreach(_.onReconnectionTimeout())


I guess this code will never run now?

Good point. We can just remove the onReconnectionTimeout() callback.

ijuma · 2017-12-22T10:17:25Z

core/src/test/scala/integration/kafka/api/MetricsTest.scala

-    assertEquals(s"Unexpected ZK state ${server.zkUtils.zkConnection.getZookeeperState}",
-        "CONNECTED", yammerMetricValue("SessionState"))
+    // Latency is rounded to milliseconds, so check the count instead.
+    val initialCount = yammerHistogramCount("kafka.server:type=ZooKeeperClientMetrics,name=ZooKeeperRequestLatencyMs").asInstanceOf[Long]


Ah, I had missed that you're using the count now. Makes sense.

ijuma · 2017-12-22T10:18:24Z

core/src/test/scala/integration/kafka/api/MetricsTest.scala

@@ -276,6 +274,16 @@ class MetricsTest extends IntegrationTestHarness with SaslSetup {
    }
  }

+  private def yammerHistogramCount(name: String): Any = {


I think we can return a Long here.

ijuma · 2017-12-22T10:21:45Z

core/src/test/scala/unit/kafka/zk/KafkaZkClientTest.scala

+    val clusterId = CoreUtils.generateUuidAsBase64
+
+    zkClient.createOrGetClusterId(clusterId)
+    assertEquals(clusterId, zkClient.getClusterId.getOrElse(fail("No cluster id found")))


Nit: we can assert that that the cluster id is Some(clusterId).

ijuma · 2017-12-22T10:22:26Z

core/src/test/scala/unit/kafka/zk/KafkaZkClientTest.scala

+    zkClient.createTopLevelPaths()
+
+    ZkData.PersistentZkPaths.foreach {
+      path => assertTrue(zkClient.pathExists(path))


Nit: path is usually in the previous line.

kafka-6320: move ZK metrics in KafkaHealthCheck to ZookeeperClient

6893752

junrao changed the title ~~kafka-6320: move ZK metrics in KafkaHealthCheck to ZookeeperClient~~ KAFKA-6320: move ZK metrics in KafkaHealthCheck to ZookeeperClient Dec 21, 2017

ijuma reviewed Dec 21, 2017

View reviewed changes

ijuma self-assigned this Dec 22, 2017

ijuma added this to the 1.1.0 milestone Dec 22, 2017

junrao added 2 commits December 21, 2017 18:38

Fix unit test bug; address review comments

203bee5

fix test compilation

d10b765

omkreddy reviewed Dec 22, 2017

View reviewed changes

ijuma approved these changes Dec 22, 2017

View reviewed changes

ijuma merged commit 27c2c81 into apache:trunk Dec 22, 2017

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-6320: move ZK metrics in KafkaHealthCheck to ZookeeperClient #4351

KAFKA-6320: move ZK metrics in KafkaHealthCheck to ZookeeperClient #4351

junrao commented Dec 21, 2017 •

edited by ijuma

Loading

junrao commented Dec 21, 2017

ijuma left a comment

ijuma Dec 21, 2017

junrao Dec 22, 2017

ijuma Dec 21, 2017

junrao Dec 22, 2017

ijuma Dec 21, 2017

junrao Dec 22, 2017

ijuma Dec 21, 2017

junrao Dec 22, 2017

ijuma Dec 22, 2017

omkreddy Dec 22, 2017

omkreddy Dec 22, 2017

omkreddy Dec 22, 2017

ijuma left a comment

ijuma Dec 22, 2017

ijuma Dec 22, 2017

ijuma Dec 22, 2017

ijuma Dec 22, 2017

ijuma Dec 22, 2017

ijuma Dec 22, 2017

junrao Dec 22, 2017

ijuma Dec 22, 2017

ijuma Dec 22, 2017

ijuma Dec 22, 2017

ijuma Dec 22, 2017

		@@ -20,7 +20,7 @@ import java.nio.charset.StandardCharsets

		import kafka.common.KafkaException
		import kafka.utils.{Json, Logging, ZkUtils}

		@@ -31,6 +32,7 @@ import org.apache.kafka.common.serialization.{IntegerSerializer, StringSerialize
		import org.I0Itec.zkclient.exception.ZkException

KAFKA-6320: move ZK metrics in KafkaHealthCheck to ZookeeperClient #4351

KAFKA-6320: move ZK metrics in KafkaHealthCheck to ZookeeperClient #4351

Conversation

junrao commented Dec 21, 2017 • edited by ijuma Loading

Committer Checklist (excluded from commit message)

junrao commented Dec 21, 2017

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

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

junrao commented Dec 21, 2017 •

edited by ijuma

Loading