Skip to content

KAFKA-10163; Define controller_mutation_rate as a Double instead of a Long #9092

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
merged 2 commits into from
Aug 3, 2020
Merged

KAFKA-10163; Define controller_mutation_rate as a Double instead of a Long #9092

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4ceb9d5
KAFKA-10163; Make `controller_mutation_rate` a Double instead of a Long
dajac Jul 28, 2020
784f215
fixup
dajac Jul 30, 2020
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
2 changes: 1 addition & 1 deletion core/src/main/scala/kafka/server/ConfigHandler.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -146,7 +146,7 @@ class QuotaConfigHandler(private val quotaManagers: QuotaManagers) {
quotaManagers.request.updateQuota(sanitizedUser, clientId, sanitizedClientId, requestQuota)
val controllerMutationQuota =
if (config.containsKey(DynamicConfig.Client.ControllerMutationOverrideProp))
Some(new Quota(config.getProperty(DynamicConfig.Client.ControllerMutationOverrideProp).toLong.toDouble, true))
Some(new Quota(config.getProperty(DynamicConfig.Client.ControllerMutationOverrideProp).toDouble, true))
else
None
quotaManagers.controllerMutation.updateQuota(sanitizedUser, clientId, sanitizedClientId, controllerMutationQuota)
Expand Down
4 changes: 4 additions & 0 deletions core/src/main/scala/kafka/server/ControllerMutationQuotaManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -129,6 +129,10 @@ class PermissiveControllerMutationQuota(private val time: Time,
}
}

object ControllerMutationQuotaManager {
val QuotaControllerMutationDefault = Int.MaxValue.toDouble
}

/**
* The ControllerMutationQuotaManager is a specialized ClientQuotaManager used in the context
* of throttling controller's operations/mutations.
Expand Down
6 changes: 3 additions & 3 deletions core/src/main/scala/kafka/server/DynamicConfig.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,7 +89,7 @@ object DynamicConfig {
val DefaultProducerOverride = ClientQuotaManagerConfig.QuotaDefault
val DefaultConsumerOverride = ClientQuotaManagerConfig.QuotaDefault
val DefaultRequestOverride = ClientRequestQuotaManager.QuotaRequestPercentDefault
val DefaultControllerMutationOverride = ClientQuotaManagerConfig.QuotaDefault
val DefaultControllerMutationOverride = ControllerMutationQuotaManager.QuotaControllerMutationDefault

// Documentation
val ProducerOverrideDoc = "A rate representing the upper bound (bytes/sec) for producer traffic."
Expand All @@ -103,7 +103,7 @@ object DynamicConfig {
.define(ProducerByteRateOverrideProp, LONG, DefaultProducerOverride, MEDIUM, ProducerOverrideDoc)
.define(ConsumerByteRateOverrideProp, LONG, DefaultConsumerOverride, MEDIUM, ConsumerOverrideDoc)
.define(RequestPercentageOverrideProp, DOUBLE, DefaultRequestOverride, MEDIUM, RequestOverrideDoc)
.define(ControllerMutationOverrideProp, LONG, DefaultConsumerOverride, MEDIUM, ControllerMutationOverrideDoc)
.define(ControllerMutationOverrideProp, DOUBLE, DefaultControllerMutationOverride, MEDIUM, ControllerMutationOverrideDoc)

def configKeys = clientConfigs.configKeys

Expand All @@ -118,7 +118,7 @@ object DynamicConfig {
.define(Client.ProducerByteRateOverrideProp, LONG, Client.DefaultProducerOverride, MEDIUM, Client.ProducerOverrideDoc)
.define(Client.ConsumerByteRateOverrideProp, LONG, Client.DefaultConsumerOverride, MEDIUM, Client.ConsumerOverrideDoc)
.define(Client.RequestPercentageOverrideProp, DOUBLE, Client.DefaultRequestOverride, MEDIUM, Client.RequestOverrideDoc)
.define(Client.ControllerMutationOverrideProp, LONG, Client.DefaultConsumerOverride, MEDIUM, Client.ControllerMutationOverrideDoc)
.define(Client.ControllerMutationOverrideProp, DOUBLE, Client.DefaultControllerMutationOverride, MEDIUM, Client.ControllerMutationOverrideDoc)

def configKeys = userConfigs.configKeys

Expand Down
5 changes: 3 additions & 2 deletions core/src/test/scala/unit/kafka/server/ControllerMutationQuotaTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,13 +110,14 @@ class ControllerMutationQuotaTest extends BaseRequestTest {

@Test
def testSetUnsetQuota(): Unit = {
val rate = 1.5
val principal = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "User")
// Default Value
waitUserQuota(principal.getName, Long.MaxValue)
// Define a new quota
defineUserQuota(principal.getName, Some(ControllerMutationRate))
defineUserQuota(principal.getName, Some(rate))
// Check it
waitUserQuota(principal.getName, ControllerMutationRate)
waitUserQuota(principal.getName, rate)
// Remove it
defineUserQuota(principal.getName, None)
// Back to the default
Expand Down