MINOR: Rename RequestChannelMetrics.apply to get

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

@@ -216,7 +216,7 @@ class RequestChannel(val queueSize: Int,
 
   def updateErrorMetrics(apiKey: ApiKeys, errors: collection.Map[Errors, Integer]): Unit = {
     errors.foreachEntry { (error, count) =>
-      metrics(apiKey.name).markErrorMeter(error, count)
+      metrics.get(apiKey.name).markErrorMeter(error, count)
     }
   }
 

core/src/test/scala/unit/kafka/network/SocketServerTest.scala

@@ -1098,7 +1098,7 @@ class SocketServerTest {
       val channel = overrideServer.dataPlaneRequestChannel
       val request = receiveRequest(channel)
 
-      val requestMetrics = channel.metrics(request.header.apiKey.name)
+      val requestMetrics = channel.metrics.get(request.header.apiKey.name)
       def totalTimeHistCount(): Long = requestMetrics.totalTimeHist.count
       val expectedTotalTimeCount = totalTimeHistCount() + 1
       val send = new NetworkSend(request.context.connectionId, ByteBufferSend.sizePrefixed(ByteBuffer.allocate(responseBufferSize)))
@@ -1170,7 +1170,7 @@ class SocketServerTest {
       TestUtils.waitUntilTrue(() => overrideServer.dataPlaneAcceptor(listener).get.processors(request.processor).channel(request.context.connectionId).isEmpty,
         s"Idle connection `${request.context.connectionId}` was not closed by selector")
 
-      val requestMetrics = channel.metrics(request.header.apiKey.name)
+      val requestMetrics = channel.metrics.get(request.header.apiKey.name)
       def totalTimeHistCount(): Long = requestMetrics.totalTimeHist.count
       val expectedTotalTimeCount = totalTimeHistCount() + 1
 
@@ -1189,9 +1189,9 @@ class SocketServerTest {
     server.stopProcessingRequests()
     val version = ApiKeys.PRODUCE.latestVersion
     val version2 = (version - 1).toShort
-    for (_ <- 0 to 1) server.dataPlaneRequestChannel.metrics(ApiKeys.PRODUCE.name).requestRate(version).mark()
-    server.dataPlaneRequestChannel.metrics(ApiKeys.PRODUCE.name).requestRate(version2).mark()
-    assertEquals(2, server.dataPlaneRequestChannel.metrics(ApiKeys.PRODUCE.name).requestRate(version).count())
+    for (_ <- 0 to 1) server.dataPlaneRequestChannel.metrics.get(ApiKeys.PRODUCE.name).requestRate(version).mark()
+    server.dataPlaneRequestChannel.metrics.get(ApiKeys.PRODUCE.name).requestRate(version2).mark()
+    assertEquals(2, server.dataPlaneRequestChannel.metrics.get(ApiKeys.PRODUCE.name).requestRate(version).count())
     server.dataPlaneRequestChannel.updateErrorMetrics(ApiKeys.PRODUCE, Map(Errors.NONE -> 1))
     val nonZeroMeters = Map(s"kafka.network:type=RequestMetrics,name=RequestsPerSec,request=Produce,version=$version" -> 2,
       s"kafka.network:type=RequestMetrics,name=RequestsPerSec,request=Produce,version=$version2" -> 1,

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

@@ -12309,8 +12309,8 @@ class KafkaApisTest extends Logging {
       verify(metadataCache, never).getAllTopics
       verify(groupConfigManager, never).groupIds
       verify(metadataCache, never).getBrokerNodes(any)
-      assertTrue(requestMetrics.apply(ApiKeys.LIST_CONFIG_RESOURCES.name).requestQueueTimeHist.count > 0)
-      assertTrue(requestMetrics.apply(RequestMetrics.LIST_CLIENT_METRICS_RESOURCES_METRIC_NAME).requestQueueTimeHist.count > 0)
+      assertTrue(requestMetrics.get(ApiKeys.LIST_CONFIG_RESOURCES.name).requestQueueTimeHist.count > 0)
+      assertTrue(requestMetrics.get(RequestMetrics.LIST_CLIENT_METRICS_RESOURCES_METRIC_NAME).requestQueueTimeHist.count > 0)
     } finally {
       requestMetrics.close()
     }
@@ -12357,8 +12357,8 @@ class KafkaApisTest extends Logging {
             ).toList
           ).flatMap(s => s.stream).collect(util.stream.Collectors.toList[ListConfigResourcesResponseData.ConfigResource]))
       assertEquals(expectedResponseData, response.data)
-      assertTrue(requestMetrics.apply(ApiKeys.LIST_CONFIG_RESOURCES.name).requestQueueTimeHist.count > 0)
-      assertEquals(0, requestMetrics.apply(RequestMetrics.LIST_CLIENT_METRICS_RESOURCES_METRIC_NAME).requestQueueTimeHist.count)
+      assertTrue(requestMetrics.get(ApiKeys.LIST_CONFIG_RESOURCES.name).requestQueueTimeHist.count > 0)
+      assertEquals(0, requestMetrics.get(RequestMetrics.LIST_CLIENT_METRICS_RESOURCES_METRIC_NAME).requestQueueTimeHist.count)
     } finally {
       requestMetrics.close()
     }

server/src/main/java/org/apache/kafka/network/Request.java

@@ -356,7 +356,7 @@ public void updateRequestMetrics(long networkThreadTimeNanos, Optional<JsonNode>
         };
 
         for (String metricName : overrideMetricNames) {
-            RequestMetrics m = metrics.apply(metricName);
+            RequestMetrics m = metrics.get(metricName);
             m.requestRate(header().apiVersion()).mark();
             m.deprecatedRequestRate(header().apiKey(), header().apiVersion(), context.clientInformation)
                 .ifPresent(Meter::mark);

server/src/main/java/org/apache/kafka/network/metrics/RequestChannelMetrics.java

@@ -25,7 +25,7 @@
 import java.util.NoSuchElementException;
 import java.util.Set;
 
-public class RequestChannelMetrics {
+public class RequestChannelMetrics implements AutoCloseable {
 
     private final Map<String, RequestMetrics> metricsMap;
 
@@ -48,14 +48,15 @@ public RequestChannelMetrics(ApiMessageType.ListenerType scope) {
         this(ApiKeys.apisForListener(scope));
     }
 
-    public RequestMetrics apply(String metricName) {
+    public RequestMetrics get(String metricName) {
         RequestMetrics requestMetrics = metricsMap.get(metricName);
         if (requestMetrics == null) {
             throw new NoSuchElementException("No RequestMetrics for " + metricName);
         }
         return requestMetrics;
     }
 
+    @Override
     public void close() {
         for (RequestMetrics requestMetrics : metricsMap.values()) {
             requestMetrics.removeMetrics();