KAFKA-5194: KIP-153: Include only client traffic in BytesOutPerSec me… #3003

mimaison · 2017-05-09T14:04:49Z

…tric

Also added 2 new metrics to account for incoming/outgoing traffic due to internal replication

ReplicationBytesInPerSec
ReplicationBytesOutPerSec

We had a look at creating an integration test to verify the Bytes/Replication In/Out metrics but had issues.
When running using IntegrationTestHarness, all the brokers were using the same Yammer Metrics registry and metrics name collided.

Also although we only update the global value for the 2 new metrics, we can see an inner node for each topic (that stays 0).
eg: kafka.server:type=BrokerTopicMetrics,name=ReplicationBytesOutPerSec,topic=testtopic

It looks like something forced by BrokerTopicMetrics. If there's no way to remove them, then we might want to update them accordingly to give them meaningful values

…tric Also added 2 new metrics to account for incoming/outgoing traffic due to internal replication - ReplicationBytesInPerSec - ReplicationBytesOutPerSec

mimaison · 2017-05-09T14:06:02Z

@ijuma @junrao This is a first pass, can you have a look ?

ijuma

Thanks for the PR, left one suggestion and one question.

ijuma · 2017-05-09T14:08:33Z

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

+          BrokerTopicStats.getBrokerTopicStats(topicPartition.topic).bytesOutRate.mark(data.records.sizeInBytes)
+          BrokerTopicStats.getBrokerAllTopicsStats.bytesOutRate.mark(data.records.sizeInBytes)
+        } else {
+          BrokerTopicStats.getBrokerAllTopicsStats.replicationBytesOutRate.mark(data.records.sizeInBytes)


We should have a method in BrokerTopicStats that takes a topic and isFromFollower that does this logic.

ijuma · 2017-05-09T14:09:07Z

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

@@ -119,6 +119,8 @@ class BrokerTopicMetrics(name: Option[String]) extends KafkaMetricsGroup {
  val failedFetchRequestRate = newMeter(BrokerTopicStats.FailedFetchRequestsPerSec, "requests", TimeUnit.SECONDS, tags)
  val totalProduceRequestRate = newMeter(BrokerTopicStats.TotalProduceRequestsPerSec, "requests", TimeUnit.SECONDS, tags)
  val totalFetchRequestRate = newMeter(BrokerTopicStats.TotalFetchRequestsPerSec, "requests", TimeUnit.SECONDS, tags)
+  val replicationBytesInRate = newMeter(BrokerTopicStats.ReplicationBytesInPerSec, "bytes", TimeUnit.SECONDS, tags)
+  val replicationBytesOutRate = newMeter(BrokerTopicStats.ReplicationBytesOutPerSec, "bytes", TimeUnit.SECONDS, tags)


Probably good to move this higher, next to the other bytes metrics.

We should only create these meters if name is None.

ijuma · 2017-05-09T14:25:01Z

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

@@ -112,6 +112,7 @@ class ReplicaFetcherThread(name: String,
        trace(s"Follower ${replica.brokerId} set replica high watermark for partition $topicPartition to $followerHighWatermark")
      if (quota.isThrottled(topicPartition))
        quota.record(records.sizeInBytes)
+      BrokerTopicStats.getBrokerAllTopicsStats.replicationBytesInRate.mark(records.sizeInBytes)


For bytesOut, we track everything in the same place (KafkaApis).

For bytesIn, it's a bit scattered. For the leader we do the check in ReplicaManager, which means that we count the bytes from GroupCoordinator.doSyncGroup, but we don't count the bytes from GroupMetadataManager.cleanupGroupMetadata. Not sure if this is by design. @junrao probably knows.

asfbot · 2017-05-09T15:25:18Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/3678/
Test PASSed (JDK 7 and Scala 2.10).

asfbot · 2017-05-09T15:35:38Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3674/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-05-09T15:50:29Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/3684/
Test FAILed (JDK 8 and Scala 2.11).

mimaison · 2017-05-10T10:51:43Z

Thanks for the feedback !

Instead of guarding the creation of the 2 new replication metrics in BrokerTopicMetrics, we decided to put them in a new class: BrokerReplicationMetrics. It feels cleaner but we might need to update the KIP to reflect that. What do you think ?

ijuma · 2017-05-10T10:56:36Z

@mimaison, with the latest change, the metric names are:

kafka.server:type=BrokerReplicationMetrics,name= ReplicationBytesInPerSec
kafka.server:type=BrokerReplicationMetrics,name= ReplicationBytesInPerSec

Did you really intend to do that?

asfbot · 2017-05-10T11:54:48Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/3705/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-05-10T11:57:06Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/3699/
Test PASSed (JDK 7 and Scala 2.10).

asfbot · 2017-05-10T12:34:50Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3695/
Test FAILed (JDK 8 and Scala 2.12).

mimaison · 2017-05-10T12:35:46Z

@ijuma have you seen #3003 (comment) ?

It seemed cleaner to have them in a new class even if they are under a new name. As these are new metrics I thought that was acceptable.

If you'd prefer to have them under BrokerTopicStats, I can move them back.
It felt unsafe to have these 2 new metrics uninitialized in case there was a topic. I was looking at something like:

 val replicationBytesInRate: Meter = name match {
    case None => newMeter(BrokerTopicStats.ReplicationBytesInPerSec, "bytes", TimeUnit.SECONDS, tags)
    case _ => null
  }

Maybe there's a better way in Scala (with Option ?)

ijuma · 2017-05-10T13:48:26Z

@mimaison I saw the comment, but it wasn't clear if you realised what the metric names would be. Metric names are public and their consistency is more important than internal aspects.

Yeah, using Option sounds good.

private[server] val replicationBytesInRate = if (name.isEmpty) Some(newMeter(...)) else None

And maybe introduce methods in BrokerTopicStats to update the metric so that callers can only use the global version.

@mpburg

With @mpburg

asfbot · 2017-05-11T10:51:58Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/3756/
Test PASSed (JDK 7 and Scala 2.11).

ijuma

A couple of minor comments, LGTM otherwise.

ijuma · 2017-05-11T11:21:47Z

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

@@ -115,6 +115,8 @@ class BrokerTopicMetrics(name: Option[String]) extends KafkaMetricsGroup {
  val bytesInRate = newMeter(BrokerTopicStats.BytesInPerSec, "bytes", TimeUnit.SECONDS, tags)
  val bytesOutRate = newMeter(BrokerTopicStats.BytesOutPerSec, "bytes", TimeUnit.SECONDS, tags)
  val bytesRejectedRate = newMeter(BrokerTopicStats.BytesRejectedPerSec, "bytes", TimeUnit.SECONDS, tags)
+  private[server] val replicationBytesInRate = if (name.isEmpty) Some(newMeter(BrokerTopicStats.ReplicationBytesInPerSec, "bytes", TimeUnit.SECONDS, tags)) else None
+  private[server] val replicationBytesOutRate = if (name.isEmpty) Some(newMeter(BrokerTopicStats.ReplicationBytesOutPerSec, "bytes", TimeUnit.SECONDS, tags)) else None


These lines are a bit long.

ijuma · 2017-05-11T11:23:48Z

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

@@ -152,9 +159,30 @@ object BrokerTopicStats extends Logging {
    stats.getAndMaybePut(topic)
  }

+  def updateReplicationBytesIn(value: Long) {
+    if (getBrokerAllTopicsStats.replicationBytesInRate.isDefined) {
+      getBrokerAllTopicsStats.replicationBytesInRate.get.mark(value)


You should use foreach instead of isDefined/get. Same for the other methods where this applies.

ijuma · 2017-05-11T11:25:51Z

I had forgotten. About the test, can we simply have a test that verifies that the metrics are updated? That should be OK even with the existing limitations right?

asfbot · 2017-05-11T11:31:42Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3746/
Test PASSed (JDK 8 and Scala 2.12).

@mpburg

With @mpburg

mimaison · 2017-05-11T15:32:09Z

Thanks for the feedback.

We've added a test that verifies metrics are updated when producing/consuming. Hopefully it's what you had in mind.

asfbot · 2017-05-11T16:30:54Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3758/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-05-11T16:53:28Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/3768/
Test FAILed (JDK 7 and Scala 2.11).

asfbot · 2017-05-11T19:09:20Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/3777/
Test PASSed (JDK 7 and Scala 2.11).

asfbot · 2017-05-11T19:32:53Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3767/
Test FAILed (JDK 8 and Scala 2.12).

junrao

@mimaison : Thanks for the patch. Looks good. Just a few minor comments.

junrao · 2017-05-11T22:48:57Z

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

+    }
+  }
+
+  def updateReplicationBytesOut(value: Long) {


This can be private?

junrao · 2017-05-11T22:49:08Z

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

    removeMetric(BrokerTopicStats.FailedProduceRequestsPerSec, tags)
    removeMetric(BrokerTopicStats.FailedFetchRequestsPerSec, tags)
    removeMetric(BrokerTopicStats.TotalProduceRequestsPerSec, tags)
    removeMetric(BrokerTopicStats.TotalFetchRequestsPerSec, tags)
+


Remove the extra new line.

junrao · 2017-05-11T22:57:10Z

core/src/test/scala/unit/kafka/metrics/MetricsTest.scala

+  @Test
+  def testBrokerTopicMetrics() {
+    val replicationBytesIn = "ReplicationBytesInPerSec"
+    val replicationBytesOut = "ReplicationBytesOutPerSec"


Could we just use the constants in BrokerTopicStats?

junrao · 2017-05-11T23:08:13Z

core/src/test/scala/unit/kafka/metrics/MetricsTest.scala

+    createTopic(zkUtils, topic, 1, numNodes, servers)
+    // Topic metrics can take a moment to be created
+    TestUtils.waitUntilTrue(
+        () => Metrics.defaultRegistry.allMetrics.asScala.filterKeys(k => k.getMBeanName.endsWith(topic)).size > 1,


Hmm, the BytesIn metric is only created when there is data produced to that topic. So, perhaps we should publish some messages first?

@mpburg

With @mpburg

ijuma

Thanks for the updates, LGTM. I included a few changes before merging as per my comments below.

ijuma · 2017-05-12T11:34:18Z

core/src/test/scala/unit/kafka/metrics/MetricsTest.scala

+      .getOrElse(fail(s"Unable to find metric ${metricName}"))
+      ._2.asInstanceOf[Meter]
+      .count
+  }


I changed this to:

Metrics.defaultRegistry.allMetrics.asScala .filterKeys(_.getMBeanName.endsWith(metricName)) .values .headOption .getOrElse(fail(s"Unable to find metric $metricName")) .asInstanceOf[Meter] .count

ijuma · 2017-05-12T11:35:11Z

core/src/test/scala/unit/kafka/metrics/MetricsTest.scala

+    assertTrue(getMeterCount(bytesOut) > initialBytesOut)
+  }
+
+  private def getMeterCount(metricName: String): Long = {


I renamed this to meterCount, we generally avoid the get prefix in Kafka.

ijuma · 2017-05-12T11:35:23Z

core/src/test/scala/unit/kafka/metrics/MetricsTest.scala

+    val replicationBytesIn = BrokerTopicStats.ReplicationBytesInPerSec
+    val replicationBytesOut = BrokerTopicStats.ReplicationBytesOutPerSec
+    val bytesIn = BrokerTopicStats.BytesInPerSec + ",topic=" + topic
+    val bytesOut = BrokerTopicStats.BytesOutPerSec + ",topic=" + topic


I changed this to use String interpolation.

ijuma · 2017-05-12T11:36:17Z

core/src/test/scala/unit/kafka/utils/TestUtils.scala

+      for (record <- consumer.poll(50).asScala)
+        records.add(record)
+    } while (records.size < numMessages)
+    records.asScala.toList


I refactored this into 2 separate methods, made sure the consumer was closed, checked the exact number of messages and added a timeout.

asfbot · 2017-05-12T11:44:22Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/3813/
Test PASSed (JDK 7 and Scala 2.11).

asfbot · 2017-05-12T12:06:56Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3801/
Test PASSed (JDK 8 and Scala 2.12).

ijuma · 2017-05-12T12:34:08Z

We had a transient test failure after this was merged:

https://builds.apache.org/blue/organizations/jenkins/kafka-trunk-jdk8/detail/kafka-trunk-jdk8/1509/tests

@mimaison, would you be able to investigate?

mimaison · 2017-05-12T12:58:21Z

Yes I'm having a look

ijuma · 2017-05-12T13:15:47Z

@mimaison, it's not clear to me how this can happen. Since we are producing messages with acks=-1, the producer only gets a response after the data has been replicated. And we need at least two fetch requests from the follower for the leader to know that the data has been replicated. The follower initialises ReplicationBytesIn while processing the response from first fetch request. Hmm.

ijuma · 2017-05-12T13:32:49Z

Oh, it's probably min.isr.

mimaison · 2017-05-12T13:39:59Z

Yes that's probably it, sending a new PR now

ijuma · 2017-05-12T14:45:37Z

I was holding off submitting my PR, but since it's been a while and we need to fix this ASAP, I did it now:

#3033

mimaison · 2017-05-12T14:48:23Z

@ijuma I just submitted a PR too #3034
Your changes are a lot more contained whereas we updated the settings for the whole class. Looking at the other tests, we made a couple of tweaks to fix/improve them as changing the ISR setting exposed a few issues

KAFKA-5194: KIP-153: Include only client traffic in BytesOutPerSec me…

ce34c59

…tric Also added 2 new metrics to account for incoming/outgoing traffic due to internal replication - ReplicationBytesInPerSec - ReplicationBytesOutPerSec

ijuma reviewed May 9, 2017

View reviewed changes

Updates following reviews

b2cb7b9

Moved back the replication metrics in BrokerTopicMetrics

3d73a6f

With @mpburg

ijuma reviewed May 11, 2017

View reviewed changes

Updates following reviews + added a test

51e1a80

With @mpburg

mimaison force-pushed the KAFKA-5194 branch from f22fa2b to 51e1a80 Compare May 11, 2017 15:30

Moved the test into MetricsTest

e5b06f8

junrao reviewed May 11, 2017

View reviewed changes

More updates

19f181e

With @mpburg

ijuma approved these changes May 12, 2017

View reviewed changes

asfgit closed this in d968662 May 12, 2017

mimaison deleted the KAFKA-5194 branch April 18, 2018 13:31

KAFKA-5194: KIP-153: Include only client traffic in BytesOutPerSec me… #3003

KAFKA-5194: KIP-153: Include only client traffic in BytesOutPerSec me… #3003

Conversation

mimaison commented May 9, 2017

mimaison commented May 9, 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

asfbot commented May 9, 2017

asfbot commented May 9, 2017

asfbot commented May 9, 2017

mimaison commented May 10, 2017

ijuma commented May 10, 2017

asfbot commented May 10, 2017

asfbot commented May 10, 2017

asfbot commented May 10, 2017

mimaison commented May 10, 2017

ijuma commented May 10, 2017

asfbot commented May 11, 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

ijuma commented May 11, 2017

asfbot commented May 11, 2017

mimaison commented May 11, 2017

asfbot commented May 11, 2017

asfbot commented May 11, 2017

asfbot commented May 11, 2017

asfbot commented May 11, 2017

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

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

asfbot commented May 12, 2017

asfbot commented May 12, 2017

ijuma commented May 12, 2017

mimaison commented May 12, 2017

ijuma commented May 12, 2017 • edited Loading

ijuma commented May 12, 2017

mimaison commented May 12, 2017

ijuma commented May 12, 2017

mimaison commented May 12, 2017

ijuma commented May 12, 2017 •

edited

Loading