Skip to content

Kafka-14743: update request metrics after callback #13297

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
showuon merged 4 commits into from
Feb 26, 2023
Merged

Kafka-14743: update request metrics after callback #13297

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
820e0bd
KAFKA-14743: update request metrics after callback
showuon Feb 23, 2023
321e64d
KAFKA-14743: refactor
showuon Feb 24, 2023
7e7d2d3
KAFKA-14743: refactor
showuon Feb 24, 2023
6515b91
KAFKA-14743: include TemporaryMemoryBytes metric validation
showuon Feb 25, 2023
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
7 changes: 4 additions & 3 deletions core/src/main/scala/kafka/network/SocketServer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1175,10 +1175,11 @@ private[kafka] class Processor(
val response = inflightResponses.remove(send.destinationId).getOrElse {
throw new IllegalStateException(s"Send for ${send.destinationId} completed, but not in `inflightResponses`")
}
updateRequestMetrics(response)

// Invoke send completion callback

// Invoke send completion callback, and then update request metrics since there might be some
// request metrics got updated during callback
response.onComplete.foreach(onComplete => onComplete(send))
updateRequestMetrics(response)

// Try unmuting the channel. If there was no quota violation and the channel has not been throttled,
// it will be unmuted immediately. If the channel has been throttled, it will unmuted only if the throttling
Expand Down
8 changes: 8 additions & 0 deletions core/src/test/scala/unit/kafka/server/FetchRequestDownConversionConfigTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ package kafka.server

import java.util
import java.util.{Optional, Properties}
import kafka.network.RequestMetrics.MessageConversionsTimeMs
import kafka.utils.{TestInfoUtils, TestUtils}
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.kafka.common.config.TopicConfig
Expand Down Expand Up @@ -163,7 +164,9 @@ class FetchRequestDownConversionConfigTest extends BaseRequestTest {
}

def testV1Fetch(isFollowerFetch: Boolean): Unit = {
val fetchMessageConversionsTimeMsMetricName = s"$MessageConversionsTimeMs,request=Fetch"

@divijvaidya divijvaidya Feb 24, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit

Is it possible to generate this string with some constants? e.g. we can use RequestMetrics.MessageConversionsTimeMs

@showuon showuon Feb 25, 2023

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, we already used the constants from RequestMetrics.MessageConversionsTimeMs. It's just the fetch/produce request is another tag name that needed to be filtered out. Thanks.

val initialFetchMessageConversionsPerSec = TestUtils.metersCount(BrokerTopicStats.FetchMessageConversionsPerSec)
val initialFetchMessageConversionsTimeMs = TestUtils.metersCount(fetchMessageConversionsTimeMsMetricName)
val topicWithDownConversionEnabled = "foo"
val topicWithDownConversionDisabled = "bar"
val replicaIds = brokers.map(_.config.brokerId)
Expand Down Expand Up @@ -222,9 +225,14 @@ class FetchRequestDownConversionConfigTest extends BaseRequestTest {
} else {
assertEquals(Errors.UNSUPPORTED_VERSION, error(partitionWithDownConversionDisabled))
}

TestUtils.waitUntilTrue(() => TestUtils.metersCount(BrokerTopicStats.FetchMessageConversionsPerSec) > initialFetchMessageConversionsPerSec,
s"The `FetchMessageConversionsPerSec` metric count is not incremented after 5 seconds. " +
s"init: $initialFetchMessageConversionsPerSec final: ${TestUtils.metersCount(BrokerTopicStats.FetchMessageConversionsPerSec)}", 5000)

TestUtils.waitUntilTrue(() => TestUtils.metersCount(fetchMessageConversionsTimeMsMetricName) > initialFetchMessageConversionsTimeMs,
s"The `MessageConversionsTimeMs` in fetch request metric count is not incremented after 5 seconds. " +
s"init: $initialFetchMessageConversionsTimeMs final: ${TestUtils.metersCount(fetchMessageConversionsTimeMsMetricName)}", 5000)

@chia7712 chia7712 Feb 24, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could you add tea for TemporaryMemoryBytes? It seems to me TemporaryMemoryBytes also get fixed by this patch .

@showuon showuon Feb 25, 2023

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good point. Updated.

}

private def sendFetch(
Expand Down
10 changes: 7 additions & 3 deletions core/src/test/scala/unit/kafka/utils/TestUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ import java.util
import java.util.concurrent.atomic.{AtomicBoolean, AtomicInteger}
import java.util.concurrent.{Callable, CompletableFuture, ExecutionException, Executors, TimeUnit}
import java.util.{Arrays, Collections, Optional, Properties}
import com.yammer.metrics.core.{Gauge, Meter}
import com.yammer.metrics.core.{Gauge, Histogram, Meter}

import javax.net.ssl.X509TrustManager
import kafka.api._
Expand Down Expand Up @@ -2101,8 +2101,12 @@ object TestUtils extends Logging {

def metersCount(metricName: String): Long = {
KafkaYammerMetrics.defaultRegistry.allMetrics.asScala
.filter { case (k, _) => k.getMBeanName.endsWith(metricName)}
.values.map(_.asInstanceOf[Meter].count()).sum
.filter { case (k, _) => k.getMBeanName.endsWith(metricName) }
.values.map {
case histogram: Histogram => histogram.count()
case meter: Meter => meter.count()
case _ => 0
}.sum
}

def clearYammerMetrics(): Unit = {
Expand Down