Flaky Test kafka.api.ClientIdQuotaTest.testQuotaOverrideDelete
Invoke `waitForQuotaUpdate` after the quotas are removed. It also changes
the default request quota to `Long.MaxValue`.
Reviewers: Anna Povzner Ismael Juma
@@ -57,9 +57,9 @@ abstract class BaseQuotaTest extends IntegrationTestHarness {
this.consumerConfig.setProperty(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, "0")
// Low enough quota that a producer sending a small payload in a tight loop should get throttled
- val defaultProducerQuota = 8000
- val defaultConsumerQuota = 2500
- val defaultRequestQuota = Int.MaxValue
+ val defaultProducerQuota: Long = 8000
+ val defaultConsumerQuota: Long = 2500
+ val defaultRequestQuota: Double = Long.MaxValue.toDouble
val topic1 = "topic-1"
var leaderNode: KafkaServer = _
@@ -95,8 +95,8 @@ abstract class BaseQuotaTest extends IntegrationTestHarness {
props.put(DynamicConfig.Client.ProducerByteRateOverrideProp, Long.MaxValue.toString)
props.put(DynamicConfig.Client.ConsumerByteRateOverrideProp, Long.MaxValue.toString)
- quotaTestClients.overrideQuotas(Long.MaxValue, Long.MaxValue, Int.MaxValue)
- quotaTestClients.waitForQuotaUpdate(Long.MaxValue, Long.MaxValue, Int.MaxValue)
+ quotaTestClients.overrideQuotas(Long.MaxValue, Long.MaxValue, Long.MaxValue.toDouble)
+ quotaTestClients.waitForQuotaUpdate(Long.MaxValue, Long.MaxValue, Long.MaxValue.toDouble)
val numRecords = 1000
assertEquals(numRecords, quotaTestClients.produceUntilThrottled(numRecords))
@@ -112,8 +112,8 @@ abstract class BaseQuotaTest extends IntegrationTestHarness {
// consumer quota is set such that consumer quota * default quota window (10 seconds) is less than
// MAX_PARTITION_FETCH_BYTES_CONFIG, so that we can test consumer ability to fetch in this case
// In this case, 250 * 10 < 4096
- quotaTestClients.overrideQuotas(2000, 250, Int.MaxValue)
- quotaTestClients.waitForQuotaUpdate(2000, 250, Int.MaxValue)
+ quotaTestClients.overrideQuotas(2000, 250, Long.MaxValue.toDouble)
+ quotaTestClients.waitForQuotaUpdate(2000, 250, Long.MaxValue.toDouble)
val numRecords = 1000
val produced = quotaTestClients.produceUntilThrottled(numRecords)
@@ -127,8 +127,8 @@ abstract class BaseQuotaTest extends IntegrationTestHarness {
@Test
def testQuotaOverrideDelete(): Unit = {
// Override producer and consumer quotas to unlimited
- quotaTestClients.overrideQuotas(Long.MaxValue, Long.MaxValue, Int.MaxValue)
- quotaTestClients.waitForQuotaUpdate(Long.MaxValue, Long.MaxValue, Int.MaxValue)
+ quotaTestClients.overrideQuotas(Long.MaxValue, Long.MaxValue, Long.MaxValue.toDouble)
+ quotaTestClients.waitForQuotaUpdate(Long.MaxValue, Long.MaxValue, Long.MaxValue.toDouble)
val numRecords = 1000
assertEquals(numRecords, quotaTestClients.produceUntilThrottled(numRecords))
@@ -139,6 +139,7 @@ abstract class BaseQuotaTest extends IntegrationTestHarness {
// Delete producer and consumer quota overrides. Consumer and producer should now be
// throttled since broker defaults are very small
quotaTestClients.removeQuotaOverrides()
+ quotaTestClients.waitForQuotaUpdate(defaultProducerQuota, defaultConsumerQuota, defaultRequestQuota)
val produced = quotaTestClients.produceUntilThrottled(numRecords)
quotaTestClients.verifyProduceThrottle(expectThrottle = true)
