dajac commented on a change in pull request #9628:
URL: https://github.com/apache/kafka/pull/9628#discussion_r539138004
##########
File path: core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala
##########
@@ -223,8 +279,34 @@ class ClientQuotasRequestTest extends BaseRequestTest {
alterEntityQuotas(entity, Map((ProducerByteRateProp -> Some(10000.5))),
validateOnly = true)
}
+ @Test
+ def testAlterClientQuotasInvalidEntityCombination(): Unit = {
Review comment:
For these three tests, would it make sense to also verify that the error
message? `InvalidRequestException` are thrown all over the place so we may get
one but not for the expected cause. Your last commit suggests that you got
caught by this already, is it?
https://github.com/apache/kafka/pull/9628/commits/1b0ec2a3ca79dc8da5908c41afc4ef87d2288d84#diff-f02b619b5cd14e83f7e21dbe211b3f88336f825e1ee5c630fee32b8a0fbe3d20R294
##########
File path: core/src/main/scala/kafka/server/AdminManager.scala
##########
@@ -920,32 +954,57 @@ class AdminManager(val config: KafkaConfig,
!name.isDefined || !strict
}
- def fromProps(props: Map[String, String]): Map[String, Double] = {
- props.map { case (key, value) =>
- val doubleValue = try value.toDouble catch {
- case _: NumberFormatException =>
- throw new IllegalStateException(s"Unexpected client quota
configuration value: $key -> $value")
- }
- key -> doubleValue
- }
- }
-
- (userEntries ++ clientIdEntries ++ bothEntries).map { case ((u, c), p) =>
+ (userEntries ++ clientIdEntries ++ bothEntries).flatMap { case ((u, c), p)
=>
val quotaProps = p.asScala.filter { case (key, _) =>
QuotaConfigs.isQuotaConfig(key) }
if (quotaProps.nonEmpty && matches(userComponent, u) &&
matches(clientIdComponent, c))
Some(userClientIdToEntity(u, c) -> fromProps(quotaProps))
else
None
- }.flatten.toMap
+ }.toMap
+ }
+
+ def handleDescribeIpQuotas(ipComponent: Option[ClientQuotaFilterComponent],
strict: Boolean): Map[ClientQuotaEntity, Map[String, Double]] = {
+ val ip = ipComponent.flatMap(c => toOption(c.`match`))
+ val exactIp = wantExact(ipComponent)
+ val allIps = ipComponent.exists(_.`match` == null) || (ipComponent.isEmpty
&& !strict)
+ val ipEntries = if (exactIp)
+ Map(Some(ip.get) -> adminZkClient.fetchEntityConfig(ConfigType.Ip,
sanitized(ip)))
+ else if (allIps)
+ adminZkClient.fetchAllEntityConfigs(ConfigType.Ip).map { case (name,
props) =>
+ Some(desanitizeEntityName(name)) -> props
+ }
+ else
+ Map.empty
+
+ def ipToQuotaEntity(ip: Option[String]): ClientQuotaEntity = {
+ new ClientQuotaEntity(ip.map(ipName => ClientQuotaEntity.IP ->
ipName).toMap.asJava)
+ }
+
+ ipEntries.flatMap { case (ip, props) =>
+ val ipQuotaProps = props.asScala.filter { case (key, _) =>
DynamicConfig.Ip.names.contains(key) }
+ if (ipQuotaProps.nonEmpty)
+ Some(ipToQuotaEntity(ip) -> fromProps(ipQuotaProps))
+ else
+ None
+ }
}
def alterClientQuotas(entries: Seq[ClientQuotaAlteration], validateOnly:
Boolean): Map[ClientQuotaEntity, ApiError] = {
def alterEntityQuotas(entity: ClientQuotaEntity, ops:
Iterable[ClientQuotaAlteration.Op]): Unit = {
- val (path, configType, configKeys) =
entityToSanitizedUserClientId(entity) match {
- case (Some(user), Some(clientId)) => (user + "/clients/" + clientId,
ConfigType.User, DynamicConfig.User.configKeys)
- case (Some(user), None) => (user, ConfigType.User,
DynamicConfig.User.configKeys)
- case (None, Some(clientId)) => (clientId, ConfigType.Client,
DynamicConfig.Client.configKeys)
- case _ => throw new InvalidRequestException("Invalid empty client
quota entity")
+ val (path, configType, configKeys) = parseAndSanitizeQuotaEntity(entity)
match {
+ case (Some(user), Some(clientId), None) => (user + "/clients/" +
clientId, ConfigType.User, DynamicConfig.User.configKeys)
+ case (Some(user), None, None) => (user, ConfigType.User,
DynamicConfig.User.configKeys)
+ case (None, Some(clientId), None) => (clientId, ConfigType.Client,
DynamicConfig.Client.configKeys)
+ case (None, None, Some(ip)) =>
+ try {
+ DynamicConfig.Ip.validateIpOrHost(ip)
+ } catch {
+ case e: IllegalArgumentException => throw new
InvalidRequestException(e.getMessage)
+ }
Review comment:
* For my understanding, have you chosen to do the validation here to
ensure that requests with `validateOnly = true` are fully validated?
* Initially, I thought that we could just catch the
`IllegalArgumentException` here
https://github.com/apache/kafka/blob/1b0ec2a3ca79dc8da5908c41afc4ef87d2288d84/core/src/main/scala/kafka/server/AdminManager.scala#L1043.
Doing the conversion here seems reasonable as well. Alternatively, we could
also change `validateIpOrHost` to return a boolean and let the caller decides
which exception to throw.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
