Skip to content
Open
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion core/src/main/scala/kafka/network/RequestChannel.scala
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
}

Expand Down
10 changes: 5 additions & 5 deletions core/src/test/scala/unit/kafka/network/SocketServerTest.scala
Original file line number Diff line number Diff line change
Expand Up @@ -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)))
Expand Down Expand Up @@ -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

Expand All @@ -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,
Expand Down
8 changes: 4 additions & 4 deletions core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
Original file line number Diff line number Diff line change
Expand Up @@ -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()
}
Expand Down Expand Up @@ -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()
}
Expand Down
2 changes: 1 addition & 1 deletion server/src/main/java/org/apache/kafka/network/Request.java
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,7 @@ public RequestChannelMetrics(ApiMessageType.ListenerType scope) {
this(ApiKeys.apisForListener(scope));
}

public RequestMetrics apply(String metricName) {
public RequestMetrics get(String metricName) {
Comment thread
dino2895 marked this conversation as resolved.
RequestMetrics requestMetrics = metricsMap.get(metricName);
if (requestMetrics == null) {
throw new NoSuchElementException("No RequestMetrics for " + metricName);
Expand Down
Loading