KAFKA-19518 Remove the usage of KafkaMetricsGroup(Class<?> klass) (#20399)

the constructor is error-prone when migrating code, since metrics could
get unintentionally changed. We should remove the constructor and use
constant strings instead to avoid issues like KAFKA-17876, KAFKA-19150,
and others.

Reviewers: Ken Huang <[email protected]>, Jhen-Yung Hsu
 <[email protected]>, KuoChe <[email protected]>, Chia-Ping
 Tsai <[email protected]>

Author: jim0987795064

core/src/main/scala/kafka/cluster/Partition.scala

@@ -116,7 +116,10 @@ class DelayedOperations(topicId: Option[Uuid],
 }
 
 object Partition {
-  private val metricsGroup = new KafkaMetricsGroup(classOf[Partition])
+  // Changing the package or class name may cause incompatibility with existing code and metrics configuration
+  private val metricsPackage = "kafka.cluster"
+  private val metricsClassName = "Partition"
+  private val metricsGroup = new KafkaMetricsGroup(metricsPackage, metricsClassName)
 
   def apply(topicIdPartition: TopicIdPartition,
             time: Time,

core/src/main/scala/kafka/coordinator/transaction/TransactionMarkerChannelManager.scala

@@ -166,8 +166,10 @@ class TransactionMarkerChannelManager(
   time: Time
 ) extends InterBrokerSendThread("TxnMarkerSenderThread-" + config.brokerId, networkClient, config.requestTimeoutMs, time)
   with Logging {
-
-  private val metricsGroup = new KafkaMetricsGroup(this.getClass)
+  // Changing the package or class name may cause incompatibility with existing code and metrics configuration
+  private val metricsPackage = "kafka.coordinator.transaction"
+  private val metricsClassName = "TransactionMarkerChannelManager"
+  private val metricsGroup = new KafkaMetricsGroup(metricsPackage, metricsClassName)
 
   this.logIdent = "[Transaction Marker Channel Manager " + config.brokerId + "]: "
 

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

@@ -81,8 +81,10 @@ class LogManager(logDirs: Seq[File],
                  cleanerFactory: (CleanerConfig, util.List[File], ConcurrentMap[TopicPartition, UnifiedLog], LogDirFailureChannel, Time) => LogCleaner =
                   (cleanerConfig, files, map, logDirFailureChannel, time) => new LogCleaner(cleanerConfig, files, map, logDirFailureChannel, time)
                 ) extends Logging {
-
-  private val metricsGroup = new KafkaMetricsGroup(this.getClass)
+  // Changing the package or class name may cause incompatibility with existing code and metrics configuration
+  private val metricsPackage = "kafka.log"
+  private val metricsClassName = "LogManager"
+  private val metricsGroup = new KafkaMetricsGroup(metricsPackage, metricsClassName)
 
   private val logCreationOrDeletionLock = new Object
   private val currentLogs = new util.concurrent.ConcurrentHashMap[TopicPartition, UnifiedLog]()

core/src/main/scala/kafka/network/RequestChannel.scala

@@ -346,7 +346,10 @@ class RequestChannel(val queueSize: Int,
                      val metrics: RequestChannelMetrics) {
   import RequestChannel._
 
-  private val metricsGroup = new KafkaMetricsGroup(this.getClass)
+  // Changing the package or class name may cause incompatibility with existing code and metrics configuration
+  private val metricsPackage = "kafka.network"
+  private val metricsClassName = "RequestChannel"
+  private val metricsGroup = new KafkaMetricsGroup(metricsPackage, metricsClassName)
 
   private val requestQueue = new ArrayBlockingQueue[BaseRequest](queueSize)
   private val processors = new ConcurrentHashMap[Int, Processor]()

core/src/main/scala/kafka/network/SocketServer.scala

@@ -78,8 +78,10 @@ class SocketServer(
   val socketFactory: ServerSocketFactory = ServerSocketFactory.INSTANCE,
   val connectionDisconnectListeners: Seq[ConnectionDisconnectListener] = Seq.empty
 ) extends Logging with BrokerReconfigurable {
-
-  private val metricsGroup = new KafkaMetricsGroup(this.getClass)
+  // Changing the package or class name may cause incompatibility with existing code and metrics configuration
+  private val metricsPackage = "kafka.network"
+  private val metricsClassName = "SocketServer"
+  private val metricsGroup = new KafkaMetricsGroup(metricsPackage, metricsClassName)
 
   private val maxQueuedRequests = config.queuedMaxRequests
 
@@ -485,9 +487,6 @@ private[kafka] abstract class Acceptor(val socketServer: SocketServer,
                                        memoryPool: MemoryPool,
                                        apiVersionManager: ApiVersionManager)
   extends Runnable with Logging {
-
-  private val metricsGroup = new KafkaMetricsGroup(this.getClass)
-
   val shouldRun = new AtomicBoolean(true)
 
   private val sendBufferSize = config.socketSendBufferBytes
@@ -516,7 +515,7 @@ private[kafka] abstract class Acceptor(val socketServer: SocketServer,
   private val blockedPercentMeterMetricName = backwardCompatibilityMetricGroup.metricName(
     "AcceptorBlockedPercent",
     Map(ListenerMetricTag -> endPoint.listener).asJava)
-  private val blockedPercentMeter = metricsGroup.newMeter(blockedPercentMeterMetricName,"blocked time", TimeUnit.NANOSECONDS)
+  private val blockedPercentMeter = backwardCompatibilityMetricGroup.newMeter(blockedPercentMeterMetricName,"blocked time", TimeUnit.NANOSECONDS)
   private var currentProcessorIndex = 0
   private[network] val throttledSockets = new mutable.PriorityQueue[DelayedCloseSocket]()
   private val started = new AtomicBoolean()
@@ -834,7 +833,9 @@ private[kafka] class Processor(
   threadName: String,
   connectionDisconnectListeners: Seq[ConnectionDisconnectListener]
 ) extends Runnable with Logging {
-  private val metricsGroup = new KafkaMetricsGroup(this.getClass)
+  private val metricsPackage = "kafka.server"
+  private val metricsClassName = "Processor"
+  private val metricsGroup = new KafkaMetricsGroup(metricsPackage, metricsClassName)
 
   val shouldRun: AtomicBoolean = new AtomicBoolean(true)
   private val started: AtomicBoolean = new AtomicBoolean()

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

@@ -30,7 +30,10 @@ import scala.jdk.OptionConverters._
 
 abstract class AbstractFetcherManager[T <: AbstractFetcherThread](val name: String, clientId: String, numFetchers: Int)
   extends Logging {
-  private val metricsGroup = new KafkaMetricsGroup(this.getClass)
+  // Changing the package or class name may cause incompatibility with existing code and metrics configuration
+  private val metricsPackage = "kafka.server"
+  private val metricsClassName = this.getClass.getSimpleName
+  private val metricsGroup = new KafkaMetricsGroup(metricsPackage, metricsClassName)
 
   // map of (source broker_id, fetcher_id per source broker) => fetcher.
   // package private for test

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

@@ -888,7 +888,10 @@ object FetcherMetrics {
 }
 
 class FetcherLagMetrics(metricId: ClientIdTopicPartition) {
-  private val metricsGroup = new KafkaMetricsGroup(this.getClass)
+  // Changing the package or class name may cause incompatibility with existing code and metrics configuration
+  private val metricsPackage = "kafka.server"
+  private val metricsClassName = "FetcherLagMetrics"
+  private val metricsGroup = new KafkaMetricsGroup(metricsPackage, metricsClassName)
 
   private[this] val lagVal = new AtomicLong(-1L)
   private[this] val tags = Map(
@@ -927,7 +930,10 @@ class FetcherLagStats(metricId: ClientIdAndBroker) {
 }
 
 class FetcherStats(metricId: ClientIdAndBroker) {
-  private val metricsGroup = new KafkaMetricsGroup(this.getClass)
+  // Changing the package or class name may cause incompatibility with existing code and metrics configuration
+  private val metricsPackage = "kafka.server"
+  private val metricsClassName = "FetcherStats"
+  private val metricsGroup = new KafkaMetricsGroup(metricsPackage, metricsClassName)
 
   val tags: util.Map[String, String] = Map("clientId" -> metricId.clientId,
     "brokerHost" -> metricId.brokerHost,

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

@@ -71,7 +71,10 @@ class ControllerServer(
 
   import kafka.server.Server._
 
-  private val metricsGroup = new KafkaMetricsGroup(this.getClass)
+  // Changing the package or class name may cause incompatibility with existing code and metrics configuration
+  private val metricsPackage = "kafka.server"
+  private val metricsClassName = "ControllerServer"
+  private val metricsGroup = new KafkaMetricsGroup(metricsPackage, metricsClassName)
 
   val config = sharedServer.controllerConfig
   val logContext = new LogContext(s"[ControllerServer id=${config.nodeId}] ")

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

@@ -187,7 +187,10 @@ class DelayedFetch(
 }
 
 object DelayedFetchMetrics {
-  private val metricsGroup = new KafkaMetricsGroup(DelayedFetchMetrics.getClass)
+  // Changing the package or class name may cause incompatibility with existing code and metrics configuration
+  private val metricsPackage = "kafka.server"
+  private val metricsClassName = "DelayedFetchMetrics"
+  private val metricsGroup = new KafkaMetricsGroup(metricsPackage, metricsClassName)
   private val FetcherTypeKey = "fetcherType"
   val followerExpiredRequestMeter: Meter = metricsGroup.newMeter("ExpiresPerSec", "requests", TimeUnit.SECONDS, Map(FetcherTypeKey -> "follower").asJava)
   val consumerExpiredRequestMeter: Meter = metricsGroup.newMeter("ExpiresPerSec", "requests", TimeUnit.SECONDS, Map(FetcherTypeKey -> "consumer").asJava)

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

@@ -135,7 +135,10 @@ class DelayedProduce(delayMs: Long,
 }
 
 object DelayedProduceMetrics {
-  private val metricsGroup = new KafkaMetricsGroup(DelayedProduceMetrics.getClass)
+  // Changing the package or class name may cause incompatibility with existing code and metrics configuration
+  private val metricsPackage = "kafka.server"
+  private val metricsClassName = "DelayedProduceMetrics"
+  private val metricsGroup = new KafkaMetricsGroup(metricsPackage, metricsClassName)
 
   private val aggregateExpirationMeter = metricsGroup.newMeter("ExpiresPerSec", "requests", TimeUnit.SECONDS)