showuon commented on code in PR #15463:
URL: https://github.com/apache/kafka/pull/15463#discussion_r1538452997
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -4164,16 +4164,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
+ assertEquals(curExpiresPerSec + 1,
DelayedRemoteFetchMetrics.expiredRequestMeter.count())
Review Comment:
I think we should use use `TestUtils.waitUntilTrue` here because the
`DelayedRemoteFetchMetrics` is marked in a separate thread, we can't make sure
it will be triggered immediately.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
