showuon commented on code in PR #15133:
URL: https://github.com/apache/kafka/pull/15133#discussion_r1461839764
##########
core/src/main/scala/kafka/server/KafkaRequestHandler.scala:
##########
@@ -659,6 +618,132 @@ class BrokerTopicStats(configOpt:
java.util.Optional[KafkaConfig] = java.util.Op
}
}
+ // Update the broker-level all topic metric values so that we have a sample
right for all topics metric after update of partition.
+ def recordRemoteCopyLagBytes(topic: String, partition: Int, value: Long):
Unit = {
+ val topicMetric = topicStats(topic)
+
topicMetric.remoteCopyLagBytesAggrMetric().setValue(String.valueOf(partition),
value)
+ allTopicsStats.remoteCopyLagBytesAggrMetric().setValue(topic,
topicMetric.remoteCopyLagBytes)
+ }
+
+ // Update the broker-level all topic metric values so that we have a sample
right for all topics metric after removal of partition.
+ def removeRemoteCopyLagBytes(topic: String, partition: Int): Unit = {
+ val topicMetric = topicStats(topic)
+
topicMetric.remoteCopyLagBytesAggrMetric().removeKey(String.valueOf(partition))
+ allTopicsStats.remoteCopyLagBytesAggrMetric().setValue(topic,
topicMetric.remoteCopyLagBytes)
+ }
Review Comment:
After refactor, the per-topic and per-broker metric will be updated within
one method to avoid forgetting about it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
