KAFKA-9730 add tag "partition" to BrokerTopicMetrics so as to observe…

core/src/main/scala/kafka/log/Log.scala

@@ -1088,7 +1088,7 @@ class Log(@volatile var dir: File,
               if (batch.sizeInBytes > config.maxMessageSize) {
                 // we record the original message set size instead of the trimmed size
                 // to be consistent with pre-compression bytesRejectedRate recording
-                brokerTopicStats.topicStats(topicPartition.topic).bytesRejectedRate.mark(records.sizeInBytes)
+                brokerTopicStats. topicStats(topicPartition).bytesRejectedRate.mark(records.sizeInBytes)
                 brokerTopicStats.allTopicsStats.bytesRejectedRate.mark(records.sizeInBytes)
                 throw new RecordTooLargeException(s"Message batch size is ${batch.sizeInBytes} bytes in append to" +
                   s"partition $topicPartition which exceeds the maximum configured size of ${config.maxMessageSize}.")
@@ -1362,7 +1362,7 @@ class Log(@volatile var dir: File,
       // Check if the message sizes are valid.
       val batchSize = batch.sizeInBytes
       if (batchSize > config.maxMessageSize) {
-        brokerTopicStats.topicStats(topicPartition.topic).bytesRejectedRate.mark(records.sizeInBytes)
+        brokerTopicStats.topicStats(topicPartition).bytesRejectedRate.mark(records.sizeInBytes)
         brokerTopicStats.allTopicsStats.bytesRejectedRate.mark(records.sizeInBytes)
         throw new RecordTooLargeException(s"The record batch size in the append to $topicPartition is $batchSize bytes " +
           s"which exceeds the maximum configured value of ${config.maxMessageSize}.")

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

@@ -746,7 +746,7 @@ class KafkaApis(val requestChannel: RequestChannel,
           unconvertedFetchResponse.sessionId)
         // record the bytes out metrics only when the response is being sent
         response.responseData.asScala.foreach { case (tp, data) =>
-          brokerTopicStats.updateBytesOut(tp.topic, fetchRequest.isFromFollower, reassigningPartitions.contains(tp), data.records.sizeInBytes)
+          brokerTopicStats.updateBytesOut(tp, fetchRequest.isFromFollower, reassigningPartitions.contains(tp), data.records.sizeInBytes)
         }
         response
       }
@@ -2915,10 +2915,10 @@ class KafkaApis(val requestChannel: RequestChannel,
     if (conversionCount > 0) {
       request.header.apiKey match {
         case ApiKeys.PRODUCE =>
-          brokerTopicStats.topicStats(tp.topic).produceMessageConversionsRate.mark(conversionCount)
+          brokerTopicStats.topicStats(tp).produceMessageConversionsRate.mark(conversionCount)
           brokerTopicStats.allTopicsStats.produceMessageConversionsRate.mark(conversionCount)
         case ApiKeys.FETCH =>
-          brokerTopicStats.topicStats(tp.topic).fetchMessageConversionsRate.mark(conversionCount)
+          brokerTopicStats.topicStats(tp).fetchMessageConversionsRate.mark(conversionCount)
           brokerTopicStats.allTopicsStats.fetchMessageConversionsRate.mark(conversionCount)
         case _ =>
           throw new IllegalStateException("Message conversion info is recorded only for Produce/Fetch requests")

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

@@ -24,6 +24,7 @@ import java.util.concurrent.{CountDownLatch, TimeUnit}
 import java.util.concurrent.atomic.AtomicInteger
 
 import com.yammer.metrics.core.Meter
+import org.apache.kafka.common.TopicPartition
 import org.apache.kafka.common.internals.FatalExitError
 import org.apache.kafka.common.utils.{KafkaThread, Time}
 
@@ -141,10 +142,10 @@ class KafkaRequestHandlerPool(val brokerId: Int,
   }
 }
 
-class BrokerTopicMetrics(name: Option[String]) extends KafkaMetricsGroup {
+class BrokerTopicMetrics(name: Option[TopicPartition]) extends KafkaMetricsGroup {
   val tags: scala.collection.Map[String, String] = name match {
     case None => Map.empty
-    case Some(topic) => Map("topic" -> topic)
+    case Some(topic) => Map("topic" -> topic.topic(), "partition" -> topic.partition().toString)
   }
 
   case class MeterWrapper(metricType: String, eventType: String) {
@@ -279,16 +280,16 @@ object BrokerTopicStats {
   val InvalidMessageCrcRecordsPerSec = "InvalidMessageCrcRecordsPerSec"
   val InvalidOffsetOrSequenceRecordsPerSec = "InvalidOffsetOrSequenceRecordsPerSec"
 
-  private val valueFactory = (k: String) => new BrokerTopicMetrics(Some(k))
+  private val valueFactory = (k: TopicPartition) => new BrokerTopicMetrics(Some(k))
 }
 
 class BrokerTopicStats {
   import BrokerTopicStats._
 
-  private val stats = new Pool[String, BrokerTopicMetrics](Some(valueFactory))
+  private val stats = new Pool[TopicPartition, BrokerTopicMetrics](Some(valueFactory))
   val allTopicsStats = new BrokerTopicMetrics(None)
 
-  def topicStats(topic: String): BrokerTopicMetrics =
+  def topicStats(topic: TopicPartition): BrokerTopicMetrics =
     stats.getAndMaybePut(topic)
 
   def updateReplicationBytesIn(value: Long): Unit = {
@@ -316,7 +317,7 @@ class BrokerTopicStats {
   }
 
   // This method only removes metrics only used for leader
-  def removeOldLeaderMetrics(topic: String): Unit = {
+  def removeOldLeaderMetrics(topic: TopicPartition): Unit = {
     val topicMetrics = topicStats(topic)
     if (topicMetrics != null) {
       topicMetrics.closeMetric(BrokerTopicStats.MessagesInPerSec)
@@ -331,21 +332,21 @@ class BrokerTopicStats {
   }
 
   // This method only removes metrics only used for follower
-  def removeOldFollowerMetrics(topic: String): Unit = {
+  def removeOldFollowerMetrics(topic: TopicPartition): Unit = {
     val topicMetrics = topicStats(topic)
     if (topicMetrics != null) {
       topicMetrics.closeMetric(BrokerTopicStats.ReplicationBytesInPerSec)
       topicMetrics.closeMetric(BrokerTopicStats.ReassignmentBytesInPerSec)
     }
   }
 
-  def removeMetrics(topic: String): Unit = {
+  def removeMetrics(topic: TopicPartition): Unit = {
     val metrics = stats.remove(topic)
     if (metrics != null)
       metrics.close()
   }
 
-  def updateBytesOut(topic: String, isFollower: Boolean, isReassignment: Boolean, value: Long): Unit = {
+  def updateBytesOut(topic: TopicPartition, isFollower: Boolean, isReassignment: Boolean, value: Long): Unit = {
     if (isFollower) {
       if (isReassignment)
         updateReassignmentBytesOut(value)

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

@@ -315,9 +315,9 @@ class ReplicaManager(val config: KafkaConfig,
     logDirFailureHandler.start()
   }
 
-  private def maybeRemoveTopicMetrics(topic: String): Unit = {
+  private def maybeRemoveTopicMetrics(topic: TopicPartition): Unit = {
     val topicHasOnlinePartition = allPartitions.values.exists {
-      case HostedPartition.Online(partition) => topic == partition.topic
+      case HostedPartition.Online(partition) => topic == partition.topicPartition
       case HostedPartition.None | HostedPartition.Offline => false
     }
     if (!topicHasOnlinePartition)
@@ -334,7 +334,7 @@ class ReplicaManager(val config: KafkaConfig,
 
         case hostedPartition @ HostedPartition.Online(removedPartition) =>
           if (allPartitions.remove(topicPartition, hostedPartition)) {
-            maybeRemoveTopicMetrics(topicPartition.topic)
+            maybeRemoveTopicMetrics(topicPartition)
             // this will delete the local log. This call may throw exception if the log is on offline directory
             removedPartition.delete()
           }
@@ -771,7 +771,7 @@ class ReplicaManager(val config: KafkaConfig,
         case HostedPartition.Online(partition) => partition.logStartOffset
         case HostedPartition.None | HostedPartition.Offline => -1L
       }
-      brokerTopicStats.topicStats(topicPartition.topic).failedProduceRequestRate.mark()
+      brokerTopicStats.topicStats(topicPartition).failedProduceRequestRate.mark()
       brokerTopicStats.allTopicsStats.failedProduceRequestRate.mark()
       error(s"Error processing append operation on partition $topicPartition", t)
 
@@ -780,7 +780,7 @@ class ReplicaManager(val config: KafkaConfig,
 
     trace(s"Append [$entriesPerPartition] to local log")
     entriesPerPartition.map { case (topicPartition, records) =>
-      brokerTopicStats.topicStats(topicPartition.topic).totalProduceRequestRate.mark()
+      brokerTopicStats.topicStats(topicPartition).totalProduceRequestRate.mark()
       brokerTopicStats.allTopicsStats.totalProduceRequestRate.mark()
 
       // reject appending to internal topics if it is not allowed
@@ -795,9 +795,9 @@ class ReplicaManager(val config: KafkaConfig,
           val numAppendedMessages = info.numMessages
 
           // update stats for successfully appended bytes and messages as bytesInRate and messageInRate
-          brokerTopicStats.topicStats(topicPartition.topic).bytesInRate.mark(records.sizeInBytes)
+          brokerTopicStats.topicStats(topicPartition).bytesInRate.mark(records.sizeInBytes)
           brokerTopicStats.allTopicsStats.bytesInRate.mark(records.sizeInBytes)
-          brokerTopicStats.topicStats(topicPartition.topic).messagesInRate.mark(numAppendedMessages)
+          brokerTopicStats.topicStats(topicPartition).messagesInRate.mark(numAppendedMessages)
           brokerTopicStats.allTopicsStats.messagesInRate.mark(numAppendedMessages)
 
           trace(s"${records.sizeInBytes} written to log $topicPartition beginning at offset " +
@@ -953,7 +953,7 @@ class ReplicaManager(val config: KafkaConfig,
       val partitionFetchSize = fetchInfo.maxBytes
       val followerLogStartOffset = fetchInfo.logStartOffset
 
-      brokerTopicStats.topicStats(tp.topic).totalFetchRequestRate.mark()
+      brokerTopicStats.topicStats(tp).totalFetchRequestRate.mark()
       brokerTopicStats.allTopicsStats.totalFetchRequestRate.mark()
 
       val adjustedMaxBytes = math.min(fetchInfo.maxBytes, limitBytes)
@@ -1043,7 +1043,7 @@ class ReplicaManager(val config: KafkaConfig,
             lastStableOffset = None,
             exception = Some(e))
         case e: Throwable =>
-          brokerTopicStats.topicStats(tp.topic).failedFetchRequestRate.mark()
+          brokerTopicStats.topicStats(tp).failedFetchRequestRate.mark()
           brokerTopicStats.allTopicsStats.failedFetchRequestRate.mark()
 
           val fetchSource = Request.describeReplicaId(replicaId)
@@ -1258,8 +1258,8 @@ class ReplicaManager(val config: KafkaConfig,
          * those topics. Note that this means the broker stops being either a replica or a leader of
          * partitions of said topics
          */
-        val leaderTopicSet = leaderPartitionsIterator.map(_.topic).toSet
-        val followerTopicSet = partitionsBecomeFollower.map(_.topic).toSet
+        val leaderTopicSet = leaderPartitionsIterator.map(_.topicPartition).toSet
+        val followerTopicSet = partitionsBecomeFollower.map(_.topicPartition).toSet
         followerTopicSet.diff(leaderTopicSet).foreach(brokerTopicStats.removeOldLeaderMetrics)
 
         // remove metrics for brokers which are not followers of a topic
@@ -1632,9 +1632,7 @@ class ReplicaManager(val config: KafkaConfig,
       newOfflinePartitions.foreach { topicPartition =>
         markPartitionOffline(topicPartition)
       }
-      newOfflinePartitions.map(_.topic).foreach { topic: String =>
-        maybeRemoveTopicMetrics(topic)
-      }
+      newOfflinePartitions.foreach(maybeRemoveTopicMetrics)
       highWatermarkCheckpoints = highWatermarkCheckpoints.filter { case (checkpointDir, _) => checkpointDir != dir }
 
       warn(s"Broker $localBrokerId stopped fetcher for partitions ${newOfflinePartitions.mkString(",")} and stopped moving logs " +

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

@@ -64,7 +64,7 @@ class MetricsTest extends KafkaServerTestHarness with Logging {
     // Don't consume messages as it may cause metrics to be re-created causing the test to fail, see KAFKA-5238
     TestUtils.generateAndProduceMessages(servers, topic, nMessages)
     assertTrue("Topic metrics don't exist", topicMetricGroups(topic).nonEmpty)
-    servers.foreach(s => assertNotNull(s.brokerTopicStats.topicStats(topic)))
+    servers.foreach(s => assertNotNull(s.brokerTopicStats.topicStats(new TopicPartition(topic, 0))))
     adminZkClient.deleteTopic(topic)
     TestUtils.verifyTopicDeletion(zkClient, topic, 1, servers)
     assertEquals("Topic metrics exists after deleteTopic", Set.empty, topicMetricGroups(topic))
@@ -126,8 +126,8 @@ class MetricsTest extends KafkaServerTestHarness with Logging {
     val topic = "test-bytes-in-out"
     val replicationBytesIn = BrokerTopicStats.ReplicationBytesInPerSec
     val replicationBytesOut = BrokerTopicStats.ReplicationBytesOutPerSec
-    val bytesIn = s"${BrokerTopicStats.BytesInPerSec},topic=$topic"
-    val bytesOut = s"${BrokerTopicStats.BytesOutPerSec},topic=$topic"
+    val bytesIn = s"${BrokerTopicStats.BytesInPerSec},topic=$topic,partition=0"
+    val bytesOut = s"${BrokerTopicStats.BytesOutPerSec},topic=$topic,partition=0"
 
     val topicConfig = new Properties
     topicConfig.setProperty(LogConfig.MinInSyncReplicasProp, "2")

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

@@ -1508,17 +1508,18 @@ class ReplicaManagerTest {
     val mockTopicStats1: BrokerTopicStats = EasyMock.mock(classOf[BrokerTopicStats])
     val (rm0, rm1) = prepareDifferentReplicaManagers(EasyMock.mock(classOf[BrokerTopicStats]), mockTopicStats1)
 
-    EasyMock.expect(mockTopicStats1.removeOldLeaderMetrics(topic)).andVoid.once
+    // make broker 0 the leader of partition 0 and
+    // make broker 1 the leader of partition 1
+    val tp0 = new TopicPartition(topic, 0)
+    val tp1 = new TopicPartition(topic, 1)
+    EasyMock.expect(mockTopicStats1.removeOldLeaderMetrics(tp0)).andVoid.times(2)
+    EasyMock.expect(mockTopicStats1.removeOldLeaderMetrics(tp1)).andVoid.once
+    EasyMock.expect(mockTopicStats1.removeOldFollowerMetrics(tp1)).andVoid.once
     EasyMock.replay(mockTopicStats1)
 
     try {
-      // make broker 0 the leader of partition 0 and
-      // make broker 1 the leader of partition 1
-      val tp0 = new TopicPartition(topic, 0)
-      val tp1 = new TopicPartition(topic, 1)
       val partition0Replicas = Seq[Integer](0, 1).asJava
       val partition1Replicas = Seq[Integer](1, 0).asJava
-
       val leaderAndIsrRequest1 = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion,
         controllerId, 0, brokerEpoch,
         Seq(
@@ -1596,15 +1597,17 @@ class ReplicaManagerTest {
     val mockTopicStats1: BrokerTopicStats = EasyMock.mock(classOf[BrokerTopicStats])
     val (rm0, rm1) = prepareDifferentReplicaManagers(EasyMock.mock(classOf[BrokerTopicStats]), mockTopicStats1)
 
-    EasyMock.expect(mockTopicStats1.removeOldLeaderMetrics(topic)).andVoid.once
-    EasyMock.expect(mockTopicStats1.removeOldFollowerMetrics(topic)).andVoid.once
+    // make broker 0 the leader of partition 0 and
+    // make broker 1 the leader of partition 1
+    val tp0 = new TopicPartition(topic, 0)
+    val tp1 = new TopicPartition(topic, 1)
+    EasyMock.expect(mockTopicStats1.removeOldFollowerMetrics(tp0)).andVoid.once
+    EasyMock.expect(mockTopicStats1.removeOldFollowerMetrics(tp1)).andVoid.once
+    EasyMock.expect(mockTopicStats1.removeOldLeaderMetrics(tp0)).andVoid.once
+    EasyMock.expect(mockTopicStats1.removeOldLeaderMetrics(tp1)).andVoid.once
     EasyMock.replay(mockTopicStats1)
 
     try {
-      // make broker 0 the leader of partition 0 and
-      // make broker 1 the leader of partition 1
-      val tp0 = new TopicPartition(topic, 0)
-      val tp1 = new TopicPartition(topic, 1)
       val partition0Replicas = Seq[Integer](1, 0).asJava
       val partition1Replicas = Seq[Integer](1, 0).asJava
 

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

@@ -170,7 +170,7 @@ class SimpleFetchTest {
   @Test
   def testReadFromLog(): Unit = {
     val brokerTopicStats = new BrokerTopicStats
-    val initialTopicCount = brokerTopicStats.topicStats(topic).totalFetchRequestRate.count()
+    val initialTopicCount = brokerTopicStats.topicStats(topicPartition).totalFetchRequestRate.count()
     val initialAllTopicsCount = brokerTopicStats.allTopicsStats.totalFetchRequestRate.count()
 
     val readCommittedRecords = replicaManager.readFromLocalLog(
@@ -200,7 +200,7 @@ class SimpleFetchTest {
     assertEquals("Reading any data can return messages up to the end of the log", recordToLEO,
       new SimpleRecord(firstRecord))
 
-    assertEquals("Counts should increment after fetch", initialTopicCount+2, brokerTopicStats.topicStats(topic).totalFetchRequestRate.count())
+    assertEquals("Counts should increment after fetch", initialTopicCount+2, brokerTopicStats.topicStats(topicPartition).totalFetchRequestRate.count())
     assertEquals("Counts should increment after fetch", initialAllTopicsCount+2, brokerTopicStats.allTopicsStats.totalFetchRequestRate.count())
   }
 }