diff --git a/core/src/main/scala/kafka/network/RequestChannel.scala b/core/src/main/scala/kafka/network/RequestChannel.scala
index ce776a1ee9f73..cc77de75d3e8f 100644
--- a/core/src/main/scala/kafka/network/RequestChannel.scala
+++ b/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)
     }
   }
 
diff --git a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
index 4e0de77c24713..6f745c5f35c6e 100644
--- a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
+++ b/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,
diff --git a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
index 97b458e27b370..1ba179b46f8e5 100644
--- a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
+++ b/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()
     }
diff --git a/server/src/main/java/org/apache/kafka/network/Request.java b/server/src/main/java/org/apache/kafka/network/Request.java
index 3b3104eb667ae..c9144889496d2 100644
--- a/server/src/main/java/org/apache/kafka/network/Request.java
+++ b/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);
diff --git a/server/src/main/java/org/apache/kafka/network/metrics/RequestChannelMetrics.java b/server/src/main/java/org/apache/kafka/network/metrics/RequestChannelMetrics.java
index 8d43107e8d825..f880e97a647f5 100644
--- a/server/src/main/java/org/apache/kafka/network/metrics/RequestChannelMetrics.java
+++ b/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,7 +48,7 @@ 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);
@@ -56,6 +56,7 @@ public RequestMetrics apply(String metricName) {
         return requestMetrics;
     }
 
+    @Override
     public void close() {
         for (RequestMetrics requestMetrics : metricsMap.values()) {
             requestMetrics.removeMetrics();