MINOR: Fix flaky ControllerMutationQuotaTest.testQuotaMetric
`ClientQuotaManager.updateQuota` updates the in-memory quota before updating the configuration of the metric. Therefore, `quotaLimit` can return the updated value while the metric's configuration still returns the previous one. This patch updates the test to be resilient to this case.
Reviewers: David Jacot
@@ -41,6 +41,7 @@ import org.apache.kafka.common.requests.DeleteTopicsResponse
import org.apache.kafka.common.security.auth.AuthenticationContext
import org.apache.kafka.common.security.auth.KafkaPrincipal
import org.apache.kafka.common.security.auth.KafkaPrincipalBuilder
+import org.apache.kafka.test.{TestUtils => JTestUtils}
import org.junit.Assert.assertEquals
import org.junit.Assert.assertTrue
import org.junit.Assert.fail
@@ -140,14 +141,14 @@ class ControllerMutationQuotaTest extends BaseRequestTest {
assertEquals(Set(Errors.NONE), errors.values.toSet)
// Metric must be there with the correct config
- verifyQuotaMetric(principal.getName, ControllerMutationRate)
+ waitQuotaMetric(principal.getName, ControllerMutationRate)
// Update quota
defineUserQuota(ThrottledPrincipal.getName, Some(ControllerMutationRate * 2))
waitUserQuota(ThrottledPrincipal.getName, ControllerMutationRate * 2)
// Metric must be there with the updated config
- verifyQuotaMetric(principal.getName, ControllerMutationRate * 2)
+ waitQuotaMetric(principal.getName, ControllerMutationRate * 2)
}
}
@@ -381,7 +382,8 @@ class ControllerMutationQuotaTest extends BaseRequestTest {
Option(servers.head.metrics.metric(metricName))
}
- private def verifyQuotaMetric(user: String, expectedQuota: Double): Unit = {
+ private def waitQuotaMetric(user: String, expectedQuota: Double): Unit = {
+ TestUtils.retry(JTestUtils.DEFAULT_MAX_WAIT_MS) {
quotaMetric(user) match {
case Some(metric) =>
val config = metric.config()
@@ -393,6 +395,7 @@ class ControllerMutationQuotaTest extends BaseRequestTest {
fail(s"Quota metric of $user is not defined")
}
}
+ }
private def alterClientQuotas(request: Map[ClientQuotaEntity, Map[String, Option[Double]]]): Map[ClientQuotaEntity, KafkaFutureImpl[Void]] = {
val entries = request.map { case (entity, alter) =>
