From 5bab5b3f6b9192bcafd361337c024f7d865d385f Mon Sep 17 00:00:00 2001 From: Chia-Ping Tsai Date: Tue, 17 Mar 2020 18:35:36 +0800 Subject: [PATCH] KAFKA-9730 add tag "partition" to BrokerTopicMetrics so as to observe the partition metrics on the same broker --- core/src/main/scala/kafka/log/Log.scala | 4 +-- .../main/scala/kafka/server/KafkaApis.scala | 6 ++--- .../kafka/server/KafkaRequestHandler.scala | 19 ++++++------- .../scala/kafka/server/ReplicaManager.scala | 26 +++++++++--------- .../unit/kafka/metrics/MetricsTest.scala | 6 ++--- .../kafka/server/ReplicaManagerTest.scala | 27 ++++++++++--------- .../unit/kafka/server/SimpleFetchTest.scala | 4 +-- 7 files changed, 47 insertions(+), 45 deletions(-) diff --git a/core/src/main/scala/kafka/log/Log.scala b/core/src/main/scala/kafka/log/Log.scala index c6bce27164dec..843b85a7e810b 100644 --- a/core/src/main/scala/kafka/log/Log.scala +++ b/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}.") diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala index 1049c5c53a343..52f9d2b2b9ca9 100644 --- a/core/src/main/scala/kafka/server/KafkaApis.scala +++ b/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") diff --git a/core/src/main/scala/kafka/server/KafkaRequestHandler.scala b/core/src/main/scala/kafka/server/KafkaRequestHandler.scala index e4583fda9263b..fc81417b3ba44 100755 --- a/core/src/main/scala/kafka/server/KafkaRequestHandler.scala +++ b/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,7 +332,7 @@ 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) @@ -339,13 +340,13 @@ class BrokerTopicStats { } } - 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) diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala index 008f422234f58..0e397b82c8c19 100644 --- a/core/src/main/scala/kafka/server/ReplicaManager.scala +++ b/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 " + diff --git a/core/src/test/scala/unit/kafka/metrics/MetricsTest.scala b/core/src/test/scala/unit/kafka/metrics/MetricsTest.scala index 80bcbc71034dd..6476bd50c6f35 100644 --- a/core/src/test/scala/unit/kafka/metrics/MetricsTest.scala +++ b/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") diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala index a7c59ea03e135..34ea92b4a19b4 100644 --- a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala +++ b/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 diff --git a/core/src/test/scala/unit/kafka/server/SimpleFetchTest.scala b/core/src/test/scala/unit/kafka/server/SimpleFetchTest.scala index b32054667d584..7043ca6d7b000 100644 --- a/core/src/test/scala/unit/kafka/server/SimpleFetchTest.scala +++ b/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()) } }