[kafka] branch 1.1 updated: KAFKA-6853: ZooKeeperRequestLatencyMs is incorrect (#4961)

[ij...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

ijuma pushed a commit to branch 1.1
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/1.1 by this push:
     new 22d6e21  KAFKA-6853: ZooKeeperRequestLatencyMs is incorrect (#4961)
22d6e21 is described below

commit 22d6e214635ea4017472765e9edcf07adfb2cd6b
Author: Fedor Bobin <fu...@gmail.com>
AuthorDate: Thu May 3 19:46:30 2018 +0300

    KAFKA-6853: ZooKeeperRequestLatencyMs is incorrect (#4961)
    
    ResponseMetadata.responseTimeMs is always 0 or negative.
    
    Reviewers: Rajini Sivaram <ra...@gmail.com>, Ismael Juma <is...@juma.me.uk>
---
 core/src/main/scala/kafka/zookeeper/ZooKeeperClient.scala  |  2 +-
 .../src/test/scala/integration/kafka/api/MetricsTest.scala | 14 +++++++++-----
 2 files changed, 10 insertions(+), 6 deletions(-)

diff --git a/core/src/main/scala/kafka/zookeeper/ZooKeeperClient.scala b/core/src/main/scala/kafka/zookeeper/ZooKeeperClient.scala
old mode 100644
new mode 100755
index 74a3a2d..5c4cd68
--- a/core/src/main/scala/kafka/zookeeper/ZooKeeperClient.scala
+++ b/core/src/main/scala/kafka/zookeeper/ZooKeeperClient.scala
@@ -476,7 +476,7 @@ sealed abstract class AsyncResponse {
 }
 
 case class ResponseMetadata(sendTimeMs: Long, receivedTimeMs: Long) {
-  def responseTimeMs: Long = sendTimeMs - receivedTimeMs
+  def responseTimeMs: Long = receivedTimeMs - sendTimeMs
 }
 
 case class CreateResponse(resultCode: Code, path: String, ctx: Option[Any], name: String, metadata: ResponseMetadata) extends AsyncResponse
diff --git a/core/src/test/scala/integration/kafka/api/MetricsTest.scala b/core/src/test/scala/integration/kafka/api/MetricsTest.scala
index baadd66..cea3d27 100644
--- a/core/src/test/scala/integration/kafka/api/MetricsTest.scala
+++ b/core/src/test/scala/integration/kafka/api/MetricsTest.scala
@@ -217,12 +217,16 @@ class MetricsTest extends IntegrationTestHarness with SaslSetup {
   }
 
   private def verifyBrokerZkMetrics(server: KafkaServer, topic: String): Unit = {
-    // Latency is rounded to milliseconds, so check the count instead.
-    val initialCount = yammerHistogramCount("kafka.server:type=ZooKeeperClientMetrics,name=ZooKeeperRequestLatencyMs")
+    val histogram = yammerHistogram("kafka.server:type=ZooKeeperClientMetrics,name=ZooKeeperRequestLatencyMs")
+    // Latency is rounded to milliseconds, so check the count instead
+    val initialCount = histogram.count
     servers.head.zkClient.getLeaderForPartition(new TopicPartition(topic, 0))
-    val newCount = yammerHistogramCount("kafka.server:type=ZooKeeperClientMetrics,name=ZooKeeperRequestLatencyMs")
+    val newCount = histogram.count
     assertTrue("ZooKeeper latency not recorded",  newCount > initialCount)
 
+    val min = histogram.min
+    assertTrue(s"Min latency should not be negative: $min", min >= 0)
+
     assertEquals(s"Unexpected ZK state", "CONNECTED", yammerMetricValue("SessionState"))
   }
 
@@ -286,12 +290,12 @@ class MetricsTest extends IntegrationTestHarness with SaslSetup {
     }
   }
 
-  private def yammerHistogramCount(name: String): Long = {
+  private def yammerHistogram(name: String): Histogram = {
     val allMetrics = Metrics.defaultRegistry.allMetrics.asScala
     val (_, metric) = allMetrics.find { case (n, _) => n.getMBeanName.endsWith(name) }
       .getOrElse(fail(s"Unable to find broker metric $name: allMetrics: ${allMetrics.keySet.map(_.getMBeanName)}"))
     metric match {
-      case m: Histogram => m.count
+      case m: Histogram => m
       case m => fail(s"Unexpected broker metric of class ${m.getClass}")
     }
   }

-- 
To stop receiving notification emails like this one, please contact
ijuma@apache.org.