This is an automated email from the ASF dual-hosted git repository.
showuon pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new a640a81040f KAFKA-16323: fix testRemoteFetchExpiresPerSecMetric
(#15463)
a640a81040f is described below
commit a640a81040f6ef6f85819b60194f0394f5f2194e
Author: Johnny Hsu <44309740+johnnych...@users.noreply.github.com>
AuthorDate: Mon Apr 1 10:52:53 2024 +0800
KAFKA-16323: fix testRemoteFetchExpiresPerSecMetric (#15463)
The variable of metrics item
(kafka.server:type=DelayedRemoteFetchMetrics,name=ExpiresPerSec) is singleton
object and it could be removed by other tests which are running in same JVM
(and it is not recreated). Hence, verifying the metrics value is not stable to
this test case.
Reviewers: Luke Chen <show...@gmail.com>, Chia-Ping Tsai
<chia7...@gmail.com>, Kamal Chandraprakash <kamal.chandraprak...@gmail.com>
---
.../unit/kafka/server/ReplicaManagerTest.scala | 23 +++-------------------
1 file changed, 3 insertions(+), 20 deletions(-)
diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
index 35a5d9cc7fb..c7587e4d615 100644
--- a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
@@ -4185,16 +4185,13 @@ class ReplicaManagerTest {
mock(classOf[FetchDataInfo])
}).when(spyRLM).read(any())
- // Get the current type=DelayedRemoteFetchMetrics,name=ExpiresPerSec
metric value before fetching
- val curExpiresPerSec =
safeYammerMetricValue("type=DelayedRemoteFetchMetrics,name=ExpiresPerSec").asInstanceOf[Long]
+ val curExpiresPerSec =
DelayedRemoteFetchMetrics.expiredRequestMeter.count()
replicaManager.fetchMessages(params, Seq(tidp0 -> new
PartitionData(topicId, fetchOffset, 0, 100000,
Optional.of[Integer](leaderEpoch), Optional.of[Integer](leaderEpoch))),
UnboundedQuota, fetchCallback)
// advancing the clock to expire the delayed remote fetch
timer.advanceClock(2000L)
- // verify the metric value is incremented since the delayed remote fetch
is expired
- TestUtils.waitUntilTrue(() => curExpiresPerSec + 1 ==
safeYammerMetricValue("type=DelayedRemoteFetchMetrics,name=ExpiresPerSec").asInstanceOf[Long],
- "The ExpiresPerSec value is not incremented. Current value is: " +
-
safeYammerMetricValue("type=DelayedRemoteFetchMetrics,name=ExpiresPerSec").asInstanceOf[Long])
+ // verify the DelayedRemoteFetchMetrics.expiredRequestMeter.mark is
called since the delayed remote fetch is expired
+ TestUtils.waitUntilTrue(() => (curExpiresPerSec + 1) ==
DelayedRemoteFetchMetrics.expiredRequestMeter.count(),
"DelayedRemoteFetchMetrics.expiredRequestMeter.count() should be 1, but got: "
+ DelayedRemoteFetchMetrics.expiredRequestMeter.count(), 10000L)
latch.countDown()
} finally {
Utils.tryAll(util.Arrays.asList[Callable[Void]](
@@ -4210,20 +4207,6 @@ class ReplicaManagerTest {
}
}
- private def safeYammerMetricValue(name: String): Any = {
- val allMetrics = KafkaYammerMetrics.defaultRegistry.allMetrics.asScala
- val opt = allMetrics.find { case (n, _) => n.getMBeanName.endsWith(name) }
- if (opt.isEmpty)
- 0L
- else {
- opt.get._2 match {
- case m: Gauge[_] => m.value
- case m: Meter => m.count()
- case m => fail(s"Unexpected broker metric of class ${m.getClass}")
- }
- }
- }
-
private def yammerMetricValue(name: String): Any = {
val allMetrics = KafkaYammerMetrics.defaultRegistry.allMetrics.asScala
val (_, metric) = allMetrics.find { case (n, _) =>
n.getMBeanName.endsWith(name) }
