diff --git a/clients/src/main/java/org/apache/kafka/common/quota/ClientQuotaEntity.java b/clients/src/main/java/org/apache/kafka/common/quota/ClientQuotaEntity.java index 0fee3d3bed106..dfad50917a926 100644 --- a/clients/src/main/java/org/apache/kafka/common/quota/ClientQuotaEntity.java +++ b/clients/src/main/java/org/apache/kafka/common/quota/ClientQuotaEntity.java @@ -32,6 +32,7 @@ public class ClientQuotaEntity { */ public static final String USER = "user"; public static final String CLIENT_ID = "client-id"; + public static final String IP = "ip"; /** * Constructs a quota entity for the given types and names. If a name is null, diff --git a/core/src/main/scala/kafka/admin/ConfigCommand.scala b/core/src/main/scala/kafka/admin/ConfigCommand.scala index eb420dd4faa50..7932c8dddb397 100644 --- a/core/src/main/scala/kafka/admin/ConfigCommand.scala +++ b/core/src/main/scala/kafka/admin/ConfigCommand.scala @@ -46,7 +46,7 @@ import scala.jdk.CollectionConverters._ import scala.collection._ /** - * This script can be used to change configs for topics/clients/users/brokers dynamically + * This script can be used to change configs for topics/clients/users/brokers/ips dynamically * An entity described or altered by the command may be one of: * - * --entity-type --entity-default may be specified in place of --entity-type --entity-name - * when describing or altering default configuration for users, clients, or brokers, respectively. - * Alternatively, --user-defaults, --client-defaults, or --broker-defaults may be specified in place of - * --entity-type --entity-default, respectively. + * --entity-type --entity-default may be specified in place of --entity-type --entity-name + * when describing or altering default configuration for users, clients, brokers, or ips, respectively. + * Alternatively, --user-defaults, --client-defaults, --broker-defaults, or --ip-defaults may be specified in place of + * --entity-type --entity-default, respectively. */ object ConfigCommand extends Config { @@ -84,7 +85,7 @@ object ConfigCommand extends Config { try { val opts = new ConfigCommandOptions(args) - CommandLineUtils.printHelpAndExitIfNeeded(opts, "This tool helps to manipulate and describe entity config for a topic, client, user or broker") + CommandLineUtils.printHelpAndExitIfNeeded(opts, "This tool helps to manipulate and describe entity config for a topic, client, user, broker or ip") opts.checkArgs() @@ -401,7 +402,11 @@ object ConfigCommand extends Config { throw new IllegalStateException(s"Altering user SCRAM credentials should never occur for more zero or multiple users: $entityNames") alterUserScramCredentialConfigs(adminClient, entityNames.head, scramConfigsToAddMap, scramConfigsToDelete) } - + case ConfigType.Ip => + val unknownConfigs = (configsToBeAdded.keys ++ configsToBeDeleted).filterNot(key => DynamicConfig.Ip.names.contains(key)) + if (unknownConfigs.nonEmpty) + throw new IllegalArgumentException(s"Only connection quota configs can be added for '${ConfigType.Ip}' using --bootstrap-server. Unexpected config names: ${unknownConfigs.mkString(",")}") + alterQuotaConfigs(adminClient, entityTypes, entityNames, configsToBeAddedMap, configsToBeDeleted) case _ => throw new IllegalArgumentException(s"Unsupported entity type: $entityTypeHead") } @@ -443,12 +448,11 @@ object ConfigCommand extends Config { if (invalidConfigs.nonEmpty) throw new InvalidConfigurationException(s"Invalid config(s): ${invalidConfigs.mkString(",")}") - val alterEntityTypes = entityTypes.map { entType => - entType match { - case ConfigType.User => ClientQuotaEntity.USER - case ConfigType.Client => ClientQuotaEntity.CLIENT_ID - case _ => throw new IllegalArgumentException(s"Unexpected entity type: ${entType}") - } + val alterEntityTypes = entityTypes.map { + case ConfigType.User => ClientQuotaEntity.USER + case ConfigType.Client => ClientQuotaEntity.CLIENT_ID + case ConfigType.Ip => ClientQuotaEntity.IP + case entType => throw new IllegalArgumentException(s"Unexpected entity type: $entType") } val alterEntityNames = entityNames.map(en => if (en.nonEmpty) en else null) @@ -461,7 +465,7 @@ object ConfigCommand extends Config { val alterOps = (configsToBeAddedMap.map { case (key, value) => val doubleValue = try value.toDouble catch { case _: NumberFormatException => - throw new IllegalArgumentException(s"Cannot parse quota configuration value for ${key}: ${value}") + throw new IllegalArgumentException(s"Cannot parse quota configuration value for $key: $value") } new ClientQuotaAlteration.Op(key, doubleValue) } ++ configsToBeDeleted.map(key => new ClientQuotaAlteration.Op(key, null))).asJavaCollection @@ -480,6 +484,8 @@ object ConfigCommand extends Config { describeResourceConfig(adminClient, entityTypes.head, entityNames.headOption, describeAll) case ConfigType.User | ConfigType.Client => describeClientQuotaAndUserScramCredentialConfigs(adminClient, entityTypes, entityNames) + case ConfigType.Ip => + describeQuotaConfigs(adminClient, entityTypes, entityNames) case entityType => throw new IllegalArgumentException(s"Invalid entity type: $entityType") } } @@ -551,7 +557,7 @@ object ConfigCommand extends Config { }).toSeq } - private def describeClientQuotaAndUserScramCredentialConfigs(adminClient: Admin, entityTypes: List[String], entityNames: List[String]) = { + private def describeQuotaConfigs(adminClient: Admin, entityTypes: List[String], entityNames: List[String]) = { val quotaConfigs = getAllClientQuotasConfigs(adminClient, entityTypes, entityNames) quotaConfigs.forKeyValue { (entity, entries) => val entityEntries = entity.entries.asScala @@ -561,15 +567,22 @@ object ConfigCommand extends Config { val typeStr = entityType match { case ClientQuotaEntity.USER => "user-principal" case ClientQuotaEntity.CLIENT_ID => "client-id" + case ClientQuotaEntity.IP => "ip" } - if (name != null) s"${typeStr} '${name}'" - else s"the default ${typeStr}" + if (name != null) s"$typeStr '$name'" + else s"the default $typeStr" } - val entityStr = (entitySubstr(ClientQuotaEntity.USER) ++ entitySubstr(ClientQuotaEntity.CLIENT_ID)).mkString(", ") + val entityStr = (entitySubstr(ClientQuotaEntity.USER) ++ + entitySubstr(ClientQuotaEntity.CLIENT_ID) ++ + entitySubstr(ClientQuotaEntity.IP)).mkString(", ") val entriesStr = entries.asScala.map(e => s"${e._1}=${e._2}").mkString(", ") - println(s"Quota configs for ${entityStr} are ${entriesStr}") + println(s"Quota configs for $entityStr are $entriesStr") } + } + + private def describeClientQuotaAndUserScramCredentialConfigs(adminClient: Admin, entityTypes: List[String], entityNames: List[String]) = { + describeQuotaConfigs(adminClient, entityTypes, entityNames) // we describe user SCRAM credentials only when we are not describing client information // and we are not given either --entity-default or --user-defaults if (!entityTypes.contains(ConfigType.Client) && !entityNames.contains("")) { @@ -597,6 +610,7 @@ object ConfigCommand extends Config { val entityType = entityTypeOpt match { case Some(ConfigType.User) => ClientQuotaEntity.USER case Some(ConfigType.Client) => ClientQuotaEntity.CLIENT_ID + case Some(ConfigType.Ip) => ClientQuotaEntity.IP case Some(_) => throw new IllegalArgumentException(s"Unexpected entity type ${entityTypeOpt.get}") case None => throw new IllegalArgumentException("More entity names specified than entity types") } @@ -735,13 +749,13 @@ object ConfigCommand extends Config { val describeOpt = parser.accepts("describe", "List configs for the given entity.") val allOpt = parser.accepts("all", "List all configs for the given topic, broker, or broker-logger entity (includes static configuration when the entity type is brokers)") - val entityType = parser.accepts("entity-type", "Type of entity (topics/clients/users/brokers/broker-loggers)") + val entityType = parser.accepts("entity-type", "Type of entity (topics/clients/users/brokers/broker-loggers/ips)") .withRequiredArg .ofType(classOf[String]) - val entityName = parser.accepts("entity-name", "Name of entity (topic name/client id/user principal name/broker id)") + val entityName = parser.accepts("entity-name", "Name of entity (topic name/client id/user principal name/broker id/ip)") .withRequiredArg .ofType(classOf[String]) - val entityDefault = parser.accepts("entity-default", "Default entity name for clients/users/brokers (applies to corresponding entity type in command line)") + val entityDefault = parser.accepts("entity-default", "Default entity name for clients/users/brokers/ips (applies to corresponding entity type in command line)") val nl = System.getProperty("line.separator") val addConfig = parser.accepts("add-config", "Key Value pairs of configs to add. Square brackets can be used to group values which contain commas: 'k1=v1,k2=[v1,v2,v2],k3=v3'. The following is a list of valid configurations: " + @@ -749,6 +763,7 @@ object ConfigCommand extends Config { "For entity-type '" + ConfigType.Broker + "': " + DynamicConfig.Broker.names.asScala.toSeq.sorted.map("\t" + _).mkString(nl, nl, nl) + "For entity-type '" + ConfigType.User + "': " + DynamicConfig.User.names.asScala.toSeq.sorted.map("\t" + _).mkString(nl, nl, nl) + "For entity-type '" + ConfigType.Client + "': " + DynamicConfig.Client.names.asScala.toSeq.sorted.map("\t" + _).mkString(nl, nl, nl) + + "For entity-type '" + ConfigType.Ip + "': " + DynamicConfig.Ip.names.asScala.toSeq.sorted.map("\t" + _).mkString(nl, nl, nl) + s"Entity types '${ConfigType.User}' and '${ConfigType.Client}' may be specified together to update config for clients of a specific user.") .withRequiredArg .ofType(classOf[String]) @@ -778,6 +793,10 @@ object ConfigCommand extends Config { val brokerLogger = parser.accepts("broker-logger", "The broker's ID for its logger config.") .withRequiredArg .ofType(classOf[String]) + val ipDefaults = parser.accepts("ip-defaults", "The config defaults for all IPs.") + val ip = parser.accepts("ip", "The IP address.") + .withRequiredArg + .ofType(classOf[String]) val zkTlsConfigFile = parser.accepts("zk-tls-config-file", "Identifies the file where ZooKeeper client TLS connectivity properties are defined. Any properties other than " + KafkaConfig.ZkSslConfigToSystemPropertyMap.keys.toList.sorted.mkString(", ") + " are ignored.") @@ -788,11 +807,13 @@ object ConfigCommand extends Config { (client, ConfigType.Client), (user, ConfigType.User), (broker, ConfigType.Broker), - (brokerLogger, BrokerLoggerConfigType)) + (brokerLogger, BrokerLoggerConfigType), + (ip, ConfigType.Ip)) private val entityDefaultsFlags = List((clientDefaults, ConfigType.Client), (userDefaults, ConfigType.User), - (brokerDefaults, ConfigType.Broker)) + (brokerDefaults, ConfigType.Broker), + (ipDefaults, ConfigType.Ip)) private[admin] def entityTypes: List[String] = { options.valuesOf(entityType).asScala.toList ++ @@ -864,13 +885,23 @@ object ConfigCommand extends Config { } } + if (hasEntityName && entityTypeVals.contains(ConfigType.Ip)) { + Seq(entityName, ip).filter(options.has(_)).map(options.valueOf(_)).foreach { ipEntity => + if (!DynamicConfig.Ip.isValidIpEntity(ipEntity)) + throw new IllegalArgumentException(s"The entity name for ${entityTypeVals.head} must be a valid IP or resolvable host, but it is: $ipEntity") + } + } + if (options.has(describeOpt) && entityTypeVals.contains(BrokerLoggerConfigType) && !hasEntityName) throw new IllegalArgumentException(s"an entity name must be specified with --describe of ${entityTypeVals.mkString(",")}") if (options.has(alterOpt)) { - if (entityTypeVals.contains(ConfigType.User) || entityTypeVals.contains(ConfigType.Client) || entityTypeVals.contains(ConfigType.Broker)) { + if (entityTypeVals.contains(ConfigType.User) || + entityTypeVals.contains(ConfigType.Client) || + entityTypeVals.contains(ConfigType.Broker) || + entityTypeVals.contains(ConfigType.Ip)) { if (!hasEntityName && !hasEntityDefault) - throw new IllegalArgumentException("an entity-name or default entity must be specified with --alter of users, clients or brokers") + throw new IllegalArgumentException("an entity-name or default entity must be specified with --alter of users, clients, brokers or ips") } else if (!hasEntityName) throw new IllegalArgumentException(s"an entity name must be specified with --alter of ${entityTypeVals.mkString(",")}") diff --git a/core/src/main/scala/kafka/server/AdminManager.scala b/core/src/main/scala/kafka/server/AdminManager.scala index f3ff6697b7bf4..f1bd1e276c4f1 100644 --- a/core/src/main/scala/kafka/server/AdminManager.scala +++ b/core/src/main/scala/kafka/server/AdminManager.scala @@ -727,7 +727,7 @@ class AdminManager(val config: KafkaConfig, case _ => DescribeConfigsResponse.ConfigType.UNKNOWN } } - + private def configSynonyms(name: String, synonyms: List[String], isSensitive: Boolean): List[DescribeConfigsResponseData.DescribeConfigsSynonym] = { val dynamicConfig = config.dynamicConfig val allSynonyms = mutable.Buffer[DescribeConfigsResponseData.DescribeConfigsSynonym]() @@ -807,23 +807,25 @@ class AdminManager(val config: KafkaConfig, case name => Sanitizer.desanitize(name) } - private def entityToSanitizedUserClientId(entity: ClientQuotaEntity): (Option[String], Option[String]) = { + private def parseAndSanitizeQuotaEntity(entity: ClientQuotaEntity): (Option[String], Option[String], Option[String]) = { if (entity.entries.isEmpty) throw new InvalidRequestException("Invalid empty client quota entity") var user: Option[String] = None var clientId: Option[String] = None + var ip: Option[String] = None entity.entries.forEach { (entityType, entityName) => val sanitizedEntityName = Some(sanitizeEntityName(entityName)) entityType match { case ClientQuotaEntity.USER => user = sanitizedEntityName case ClientQuotaEntity.CLIENT_ID => clientId = sanitizedEntityName + case ClientQuotaEntity.IP => ip = sanitizedEntityName case _ => throw new InvalidRequestException(s"Unhandled client quota entity type: ${entityType}") } if (entityName != null && entityName.isEmpty) throw new InvalidRequestException(s"Empty ${entityType} not supported") } - (user, clientId) + (user, clientId, ip) } private def userClientIdToEntity(user: Option[String], clientId: Option[String]): ClientQuotaEntity = { @@ -833,16 +835,21 @@ class AdminManager(val config: KafkaConfig, def describeClientQuotas(filter: ClientQuotaFilter): Map[ClientQuotaEntity, Map[String, Double]] = { var userComponent: Option[ClientQuotaFilterComponent] = None var clientIdComponent: Option[ClientQuotaFilterComponent] = None + var ipComponent: Option[ClientQuotaFilterComponent] = None filter.components.forEach { component => component.entityType match { case ClientQuotaEntity.USER => if (userComponent.isDefined) - throw new InvalidRequestException(s"Duplicate user filter component entity type"); + throw new InvalidRequestException(s"Duplicate user filter component entity type") userComponent = Some(component) case ClientQuotaEntity.CLIENT_ID => if (clientIdComponent.isDefined) - throw new InvalidRequestException(s"Duplicate client filter component entity type"); + throw new InvalidRequestException(s"Duplicate client filter component entity type") clientIdComponent = Some(component) + case ClientQuotaEntity.IP => + if (ipComponent.isDefined) + throw new InvalidRequestException(s"Duplicate ip filter component entity type") + ipComponent = Some(component) case "" => throw new InvalidRequestException(s"Unexpected empty filter component entity type") case et => @@ -850,28 +857,55 @@ class AdminManager(val config: KafkaConfig, throw new UnsupportedVersionException(s"Custom entity type '${et}' not supported") } } - handleDescribeClientQuotas(userComponent, clientIdComponent, filter.strict) + if ((userComponent.isDefined || clientIdComponent.isDefined) && ipComponent.isDefined) + throw new InvalidRequestException(s"Invalid entity filter component combination, IP filter component should not be used with " + + s"user or clientId filter component.") + + val userClientQuotas = if (ipComponent.isEmpty) + handleDescribeClientQuotas(userComponent, clientIdComponent, filter.strict) + else + Map.empty + + val ipQuotas = if (userComponent.isEmpty && clientIdComponent.isEmpty) + handleDescribeIpQuotas(ipComponent, filter.strict) + else + Map.empty + + (userClientQuotas ++ ipQuotas).toMap + } + + private def wantExact(component: Option[ClientQuotaFilterComponent]): Boolean = component.exists(_.`match` != null) + + private def toOption(opt: java.util.Optional[String]): Option[String] = { + if (opt == null) + None + else if (opt.isPresent) + Some(opt.get) + else + Some(null) + } + + private def sanitized(name: Option[String]): String = name.map(n => sanitizeEntityName(n)).getOrElse("") + + private 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 + } } def handleDescribeClientQuotas(userComponent: Option[ClientQuotaFilterComponent], clientIdComponent: Option[ClientQuotaFilterComponent], strict: Boolean): Map[ClientQuotaEntity, Map[String, Double]] = { - def toOption(opt: java.util.Optional[String]): Option[String] = - if (opt == null) - None - else if (opt.isPresent) - Some(opt.get) - else - Some(null) - val user = userComponent.flatMap(c => toOption(c.`match`)) val clientId = clientIdComponent.flatMap(c => toOption(c.`match`)) - def sanitized(name: Option[String]): String = name.map(n => sanitizeEntityName(n)).getOrElse("") val sanitizedUser = sanitized(user) val sanitizedClientId = sanitized(clientId) - def wantExact(component: Option[ClientQuotaFilterComponent]): Boolean = component.exists(_.`match` != null) val exactUser = wantExact(userComponent) val exactClientId = wantExact(clientIdComponent) @@ -920,32 +954,54 @@ 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)) => + if (!DynamicConfig.Ip.isValidIpEntity(ip)) + throw new InvalidRequestException(s"$ip is not a valid IP or resolvable host.") + (ip, ConfigType.Ip, DynamicConfig.Ip.configKeys) + case (_, _, Some(_)) => throw new InvalidRequestException(s"Invalid quota entity combination, " + + s"IP entity should not be used with user/client ID entity.") + case _ => throw new InvalidRequestException("Invalid client quota entity") } val props = adminZkClient.fetchEntityConfig(configType, path) @@ -959,12 +1015,15 @@ class AdminManager(val config: KafkaConfig, case key => key.`type` match { case ConfigDef.Type.DOUBLE => props.setProperty(op.key, value.toString) - case ConfigDef.Type.LONG => + case ConfigDef.Type.LONG | ConfigDef.Type.INT => val epsilon = 1e-6 - val longValue = (value + epsilon).toLong - if ((longValue.toDouble - value).abs > epsilon) - throw new InvalidRequestException(s"Configuration ${op.key} must be a Long value") - props.setProperty(op.key, longValue.toString) + val intValue = if (key.`type` == ConfigDef.Type.LONG) + (value + epsilon).toLong + else + (value + epsilon).toInt + if ((intValue.toDouble - value).abs > epsilon) + throw new InvalidRequestException(s"Configuration ${op.key} must be a ${key.`type`} value") + props.setProperty(op.key, intValue.toString) case _ => throw new IllegalStateException(s"Unexpected config type ${key.`type`}") } diff --git a/core/src/main/scala/kafka/server/DynamicConfig.scala b/core/src/main/scala/kafka/server/DynamicConfig.scala index e963e1a639171..f00c610776aa4 100644 --- a/core/src/main/scala/kafka/server/DynamicConfig.scala +++ b/core/src/main/scala/kafka/server/DynamicConfig.scala @@ -17,6 +17,7 @@ package kafka.server +import java.net.{InetAddress, UnknownHostException} import java.util.Properties import kafka.log.LogConfig @@ -141,6 +142,17 @@ object DynamicConfig { def names = ipConfigs.names def validate(props: Properties) = DynamicConfig.validate(ipConfigs, props, customPropsAllowed = false) + + def isValidIpEntity(ip: String): Boolean = { + if (ip != ConfigEntityName.Default) { + try { + InetAddress.getByName(ip) + } catch { + case _: UnknownHostException => return false + } + } + true + } } private def validate(configDef: ConfigDef, props: Properties, customPropsAllowed: Boolean) = { diff --git a/core/src/main/scala/kafka/zk/AdminZkClient.scala b/core/src/main/scala/kafka/zk/AdminZkClient.scala index e558aa0eb0859..1e994b4ecac85 100644 --- a/core/src/main/scala/kafka/zk/AdminZkClient.scala +++ b/core/src/main/scala/kafka/zk/AdminZkClient.scala @@ -28,7 +28,6 @@ import kafka.utils.Implicits._ import org.apache.kafka.common.{TopicPartition, Uuid} import org.apache.kafka.common.errors._ import org.apache.kafka.common.internals.Topic -import org.apache.kafka.common.utils.Utils import org.apache.zookeeper.KeeperException.NodeExistsException import scala.collection.{Map, Seq} @@ -351,7 +350,7 @@ class AdminZkClient(zkClient: KafkaZkClient) extends Logging { case ConfigType.User => changeUserOrUserClientIdConfig(entityName, configs) case ConfigType.Broker => changeBrokerConfig(parseBroker(entityName), configs) case ConfigType.Ip => changeIpConfig(entityName, configs) - case _ => throw new IllegalArgumentException(s"$entityType is not a known entityType. Should be one of ${ConfigType.Topic}, ${ConfigType.Client}, ${ConfigType.Broker}") + case _ => throw new IllegalArgumentException(s"$entityType is not a known entityType. Should be one of ${ConfigType.all}") } } @@ -394,8 +393,8 @@ class AdminZkClient(zkClient: KafkaZkClient) extends Logging { * @param configs properties to validate for the IP */ def validateIpConfig(ip: String, configs: Properties): Unit = { - if (ip != ConfigEntityName.Default && !Utils.validHostPattern(ip)) - throw new AdminOperationException(s"IP $ip is not a valid address.") + if (!DynamicConfig.Ip.isValidIpEntity(ip)) + throw new AdminOperationException(s"$ip is not a valid IP or resolvable host.") DynamicConfig.Ip.validate(configs) } @@ -480,8 +479,8 @@ class AdminZkClient(zkClient: KafkaZkClient) extends Logging { } /** - * Read the entity (topic, broker, client, user or ) config (if any) from zk - * sanitizedEntityName is , , , or /clients/. + * Read the entity (topic, broker, client, user, or ) config (if any) from zk + * sanitizedEntityName is , , , , /clients/ or . * @param rootEntityType * @param sanitizedEntityName * @return diff --git a/core/src/test/scala/integration/kafka/network/DynamicConnectionQuotaTest.scala b/core/src/test/scala/integration/kafka/network/DynamicConnectionQuotaTest.scala index b5cd7f6c7d056..ac8f49ee9b7cf 100644 --- a/core/src/test/scala/integration/kafka/network/DynamicConnectionQuotaTest.scala +++ b/core/src/test/scala/integration/kafka/network/DynamicConnectionQuotaTest.scala @@ -23,12 +23,13 @@ import java.net.{InetAddress, Socket} import java.util.concurrent._ import java.util.{Collections, Properties} -import kafka.server.{BaseRequestTest, ConfigEntityName, DynamicConfig, KafkaConfig} -import kafka.utils.{CoreUtils, TestUtils} +import kafka.server.{BaseRequestTest, DynamicConfig, KafkaConfig} +import kafka.utils.TestUtils import org.apache.kafka.clients.admin.{Admin, AdminClientConfig} import org.apache.kafka.common.message.ProduceRequestData import org.apache.kafka.common.network.ListenerName import org.apache.kafka.common.protocol.Errors +import org.apache.kafka.common.quota.ClientQuotaEntity import org.apache.kafka.common.record.{CompressionType, MemoryRecords, SimpleRecord} import org.apache.kafka.common.requests.{ProduceRequest, ProduceResponse} import org.apache.kafka.common.security.auth.SecurityProtocol @@ -47,6 +48,7 @@ class DynamicConnectionQuotaTest extends BaseRequestTest { val topic = "test" val listener = ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT) val localAddress = InetAddress.getByName("127.0.0.1") + val unknownHost = "255.255.0.1" val plaintextListenerDefaultQuota = 30 var executor: ExecutorService = _ @@ -236,16 +238,19 @@ class DynamicConnectionQuotaTest extends BaseRequestTest { @Test def testDynamicIpConnectionRateQuota(): Unit = { val connRateLimit = 10 + val initialConnectionCount = connectionCount // before setting connection rate to 10, verify we can do at least double that by default (no limit) - verifyConnectionRate(2 * connRateLimit, Int.MaxValue, "PLAINTEXT", ignoreIOExceptions = false) + verifyConnectionRate(2 * connRateLimit, plaintextListenerDefaultQuota, "PLAINTEXT", ignoreIOExceptions = false) + waitForConnectionCount(initialConnectionCount) // set default IP connection rate quota, verify that we don't exceed the limit updateIpConnectionRate(None, connRateLimit) verifyConnectionRate(8, connRateLimit, "PLAINTEXT", ignoreIOExceptions = true) - + waitForConnectionCount(initialConnectionCount) // set a higher IP connection rate quota override, verify that the higher limit is now enforced val newRateLimit = 18 updateIpConnectionRate(Some(localAddress.getHostAddress), newRateLimit) verifyConnectionRate(14, newRateLimit, "PLAINTEXT", ignoreIOExceptions = true) + waitForConnectionCount(initialConnectionCount) } private def reconfigureServers(newProps: Properties, perBrokerConfig: Boolean, aPropToVerify: (String, String)): Unit = { @@ -259,13 +264,22 @@ class DynamicConnectionQuotaTest extends BaseRequestTest { } private def updateIpConnectionRate(ip: Option[String], updatedRate: Int): Unit = { - adminZkClient.changeIpConfig(ip.getOrElse(ConfigEntityName.Default), - CoreUtils.propsWith(DynamicConfig.Ip.IpConnectionRateOverrideProp, updatedRate.toString)) - // use a random throwaway address if ip isn't specified to get the default value - TestUtils.waitUntilTrue(() => servers.head.socketServer.connectionQuotas. - connectionRateForIp(InetAddress.getByName(ip.getOrElse("255.255.3.4"))) == updatedRate, - s"Timed out waiting for connection rate update to propagate" - ) + val initialConnectionCount = connectionCount + val adminClient = createAdminClient() + try { + val entity = new ClientQuotaEntity(Map(ClientQuotaEntity.IP -> ip.orNull).asJava) + val request = Map(entity -> Map(DynamicConfig.Ip.IpConnectionRateOverrideProp -> Some(updatedRate.toDouble))) + TestUtils.alterClientQuotas(adminClient, request).all.get() + // use a random throwaway address if ip isn't specified to get the default value + TestUtils.waitUntilTrue(() => servers.head.socketServer.connectionQuotas. + connectionRateForIp(InetAddress.getByName(ip.getOrElse(unknownHost))) == updatedRate, + s"Timed out waiting for connection rate update to propagate" + ) + } finally { + adminClient.close() + } + TestUtils.waitUntilTrue(() => initialConnectionCount == connectionCount, + s"Admin client connection not closed (initial = $initialConnectionCount, current = $connectionCount)") } private def waitForListener(listenerName: String): Unit = { diff --git a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala index 212e0eb8979c3..1eb18c9815c7f 100644 --- a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala +++ b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala @@ -133,6 +133,16 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging { testArgumentParse("broker-loggers", zkConfig = false) } + @Test + def shouldParseArgumentsForIpEntityType(): Unit = { + testArgumentParse("ips", zkConfig = false) + } + + @Test + def shouldParseArgumentsForIpEntityTypeUsingZookeeper(): Unit = { + testArgumentParse("ips", zkConfig = true) + } + def testArgumentParse(entityType: String, zkConfig: Boolean): Unit = { val shortFlag: String = s"--${entityType.dropRight(1)}" @@ -296,6 +306,7 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging { testExpectedEntityTypeNames(List(ConfigType.Topic), List("A"), "--entity-type", "topics", "--entity-name", "A") testExpectedEntityTypeNames(List(ConfigType.Broker), List("0"), "--entity-name", "0", "--entity-type", "brokers") + testExpectedEntityTypeNames(List(ConfigType.Ip), List("1.2.3.4"), "--entity-name", "1.2.3.4", "--entity-type", "ips") testExpectedEntityTypeNames(List(ConfigType.User, ConfigType.Client), List("A", ""), "--entity-type", "users", "--entity-type", "clients", "--entity-name", "A", "--entity-default") testExpectedEntityTypeNames(List(ConfigType.User, ConfigType.Client), List("", "B"), @@ -303,6 +314,7 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging { testExpectedEntityTypeNames(List(ConfigType.Topic), List("A"), "--topic", "A") testExpectedEntityTypeNames(List(ConfigType.Broker), List("0"), "--broker", "0") + testExpectedEntityTypeNames(List(ConfigType.Ip), List("1.2.3.4"), "--ip", "1.2.3.4") testExpectedEntityTypeNames(List(ConfigType.Client, ConfigType.User), List("B", "A"), "--client", "B", "--user", "A") testExpectedEntityTypeNames(List(ConfigType.Client, ConfigType.User), List("B", ""), "--client", "B", "--user-defaults") testExpectedEntityTypeNames(List(ConfigType.Client, ConfigType.User), List("A"), @@ -311,6 +323,7 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging { testExpectedEntityTypeNames(List(ConfigType.Topic), List.empty, "--entity-type", "topics") testExpectedEntityTypeNames(List(ConfigType.User), List.empty, "--entity-type", "users") testExpectedEntityTypeNames(List(ConfigType.Broker), List.empty, "--entity-type", "brokers") + testExpectedEntityTypeNames(List(ConfigType.Ip), List.empty, "--entity-type", "ips") } @Test @@ -379,6 +392,34 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging { createOpts.checkArgs() } + @Test(expected = classOf[IllegalArgumentException]) + def shouldFailIfInvalidHost(): Unit = { + val createOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092", + "--entity-name", "A,B", "--entity-type", "ips", "--describe")) + createOpts.checkArgs() + } + + @Test(expected = classOf[IllegalArgumentException]) + def shouldFailIfInvalidHostUsingZookeeper(): Unit = { + val createOpts = new ConfigCommandOptions(Array("--zookeeper", zkConnect, + "--entity-name", "A,B", "--entity-type", "ips", "--describe")) + createOpts.checkArgs() + } + + @Test(expected = classOf[IllegalArgumentException]) + def shouldFailIfUnresolvableHost(): Unit = { + val createOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092", + "--entity-name", "admin", "--entity-type", "ips", "--describe")) + createOpts.checkArgs() + } + + @Test(expected = classOf[IllegalArgumentException]) + def shouldFailIfUnresolvableHostUsingZookeeper(): Unit = { + val createOpts = new ConfigCommandOptions(Array("--zookeeper", zkConnect, + "--entity-name", "admin", "--entity-type", "ips", "--describe")) + createOpts.checkArgs() + } + @Test def shouldAddClientConfigUsingZookeeper(): Unit = { val createOpts = new ConfigCommandOptions(Array("--zookeeper", zkConnect, @@ -398,39 +439,117 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging { ConfigCommand.alterConfigWithZk(null, createOpts, new TestAdminZkClient(zkClient)) } - def testShouldAddClientConfig(user: Option[String], clientId: Option[String]): Unit = { - def toValues(entityName: Option[String], entityType: String, command: String): - (Array[String], Option[String], Option[ClientQuotaFilterComponent]) = { - entityName match { - case Some(null) => - (Array("--entity-type", command, "--entity-default"), Some(null), - Some(ClientQuotaFilterComponent.ofDefaultEntity(entityType))) - case Some(name) => - (Array("--entity-type", command, "--entity-name", name), Some(name), - Some(ClientQuotaFilterComponent.ofEntity(entityType, name))) - case None => (Array.empty, None, None) + @Test + def shouldAddIpConfigsUsingZookeeper(): Unit = { + val createOpts = new ConfigCommandOptions(Array("--zookeeper", zkConnect, + "--entity-name", "1.2.3.4", + "--entity-type", "ips", + "--alter", + "--add-config", "a=b,c=d")) + + class TestAdminZkClient(zkClient: KafkaZkClient) extends AdminZkClient(zkClient) { + override def changeIpConfig(ip: String, configChange: Properties): Unit = { + assertEquals("1.2.3.4", ip) + assertEquals("b", configChange.get("a")) + assertEquals("d", configChange.get("c")) } } - val (userArgs, userEntity, userComponent) = toValues(user, ClientQuotaEntity.USER, "users") - val (clientIdArgs, clientIdEntity, clientIdComponent) = toValues(clientId, ClientQuotaEntity.CLIENT_ID, "clients") - val createOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092", - "--alter", - "--add-config", "consumer_byte_rate=20000,producer_byte_rate=10000", - "--delete-config", "request_percentage") ++ userArgs ++ clientIdArgs) + ConfigCommand.alterConfigWithZk(null, createOpts, new TestAdminZkClient(zkClient)) + } + + private def toValues(entityName: Option[String], entityType: String): (Array[String], Map[String, String]) = { + val command = entityType match { + case ClientQuotaEntity.USER => "users" + case ClientQuotaEntity.CLIENT_ID => "clients" + case ClientQuotaEntity.IP => "ips" + } + entityName match { + case Some(null) => + (Array("--entity-type", command, "--entity-default"), Map(entityType -> null)) + case Some(name) => + (Array("--entity-type", command, "--entity-name", name), Map(entityType -> name)) + case None => (Array.empty, Map.empty) + } + } + + private def verifyAlterCommandFails(expectedErrorMessage: String, alterOpts: Seq[String]): Unit = { + val mockAdminClient: Admin = EasyMock.createStrictMock(classOf[Admin]) + val opts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092", + "--alter") ++ alterOpts) + val e = intercept[IllegalArgumentException] { + ConfigCommand.alterConfig(mockAdminClient, opts) + } + assertTrue(s"Unexpected exception: $e", e.getMessage.contains(expectedErrorMessage)) + } + + @Test + def shouldNotAlterNonQuotaIpConfigsUsingBootstrapServer(): Unit = { + // when using --bootstrap-server, it should be illegal to alter anything that is not a connection quota + // for ip entities + val ipEntityOpts = List("--entity-type", "ips", "--entity-name", "127.0.0.1") + val invalidProp = "some_config" + verifyAlterCommandFails(invalidProp, ipEntityOpts ++ List("--add-config", "connection_creation_rate=10000,some_config=10")) + verifyAlterCommandFails(invalidProp, ipEntityOpts ++ List("--add-config", "some_config=10")) + verifyAlterCommandFails(invalidProp, ipEntityOpts ++ List("--delete-config", "connection_creation_rate=10000,some_config=10")) + verifyAlterCommandFails(invalidProp, ipEntityOpts ++ List("--delete-config", "some_config=10")) + } + + private def verifyDescribeQuotas(describeArgs: List[String], expectedFilter: ClientQuotaFilter): Unit = { + val describeOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092", + "--describe") ++ describeArgs) + val describeFuture = new KafkaFutureImpl[util.Map[ClientQuotaEntity, util.Map[String, java.lang.Double]]] + describeFuture.complete(Map.empty[ClientQuotaEntity, util.Map[String, java.lang.Double]].asJava) + val describeResult: DescribeClientQuotasResult = EasyMock.createNiceMock(classOf[DescribeClientQuotasResult]) + EasyMock.expect(describeResult.entities()).andReturn(describeFuture) + + var describedConfigs = false + val node = new Node(1, "localhost", 9092) + val mockAdminClient = new MockAdminClient(util.Collections.singletonList(node), node) { + override def describeClientQuotas(filter: ClientQuotaFilter, options: DescribeClientQuotasOptions): DescribeClientQuotasResult = { + assertTrue(filter.strict) + assertEquals(expectedFilter.components().asScala.toSet, filter.components.asScala.toSet) + describedConfigs = true + describeResult + } + } + EasyMock.replay(describeResult) + ConfigCommand.describeConfig(mockAdminClient, describeOpts) + assertTrue(describedConfigs) + } - // Explicitly populate a HashMap to ensure nulls are recorded properly. - val entityMap = new java.util.HashMap[String, String] - userEntity.foreach(u => entityMap.put(ClientQuotaEntity.USER, u)) - clientIdEntity.foreach(c => entityMap.put(ClientQuotaEntity.CLIENT_ID, c)) - val entity = new ClientQuotaEntity(entityMap) + @Test + def testDescribeIpConfigs(): Unit = { + val entityType = ClientQuotaEntity.IP + val knownHost = "1.2.3.4" + val defaultIpFilter = ClientQuotaFilter.containsOnly(List(ClientQuotaFilterComponent.ofDefaultEntity(entityType)).asJava) + val singleIpFilter = ClientQuotaFilter.containsOnly(List(ClientQuotaFilterComponent.ofEntity(entityType, knownHost)).asJava) + val allIpsFilter = ClientQuotaFilter.containsOnly(List(ClientQuotaFilterComponent.ofEntityType(entityType)).asJava) + verifyDescribeQuotas(List("--entity-default", "--entity-type", "ips"), defaultIpFilter) + verifyDescribeQuotas(List("--ip-defaults"), defaultIpFilter) + verifyDescribeQuotas(List("--entity-type", "ips", "--entity-name", knownHost), singleIpFilter) + verifyDescribeQuotas(List("--ip", knownHost), singleIpFilter) + verifyDescribeQuotas(List("--entity-type", "ips"), allIpsFilter) + } + + def verifyAlterQuotas(alterOpts: Seq[String], expectedAlterEntity: ClientQuotaEntity, + expectedProps: Map[String, java.lang.Double], expectedAlterOps: Set[ClientQuotaAlteration.Op]): Unit = { + val createOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092", + "--alter") ++ alterOpts) var describedConfigs = false val describeFuture = new KafkaFutureImpl[util.Map[ClientQuotaEntity, util.Map[String, java.lang.Double]]] - describeFuture.complete(Map((entity -> Map(("request_percentage" -> Double.box(50.0))).asJava)).asJava) + describeFuture.complete(Map(expectedAlterEntity -> expectedProps.asJava).asJava) val describeResult: DescribeClientQuotasResult = EasyMock.createNiceMock(classOf[DescribeClientQuotasResult]) EasyMock.expect(describeResult.entities()).andReturn(describeFuture) + val expectedFilterComponents = expectedAlterEntity.entries.asScala.map { case (entityType, entityName) => + if (entityName == null) + ClientQuotaFilterComponent.ofDefaultEntity(entityType) + else + ClientQuotaFilterComponent.ofEntity(entityType, entityName) + }.toSet + var alteredConfigs = false val alterFuture = new KafkaFutureImpl[Void] alterFuture.complete(null) @@ -441,9 +560,7 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging { val mockAdminClient = new MockAdminClient(util.Collections.singletonList(node), node) { override def describeClientQuotas(filter: ClientQuotaFilter, options: DescribeClientQuotasOptions): DescribeClientQuotasResult = { assertTrue(filter.strict) - val components = filter.components.asScala.toSeq - userComponent.foreach(c => assertTrue(components.contains(c))) - clientIdComponent.foreach(c => assertTrue(components.contains(c))) + assertEquals(expectedFilterComponents, filter.components().asScala.toSet) describedConfigs = true describeResult } @@ -452,15 +569,9 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging { assertFalse(options.validateOnly) assertEquals(1, entries.size) val alteration = entries.asScala.head - assertEquals(entity, alteration.entity) + assertEquals(expectedAlterEntity, alteration.entity) val ops = alteration.ops.asScala - assertEquals(3, ops.size) - val expectedOps = Set( - new ClientQuotaAlteration.Op("consumer_byte_rate", Double.box(20000)), - new ClientQuotaAlteration.Op("producer_byte_rate", Double.box(10000)), - new ClientQuotaAlteration.Op("request_percentage", null) - ) - assertEquals(expectedOps, ops.toSet) + assertEquals(expectedAlterOps, ops.toSet) alteredConfigs = true alterResult } @@ -469,160 +580,144 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging { ConfigCommand.alterConfig(mockAdminClient, createOpts) assertTrue(describedConfigs) assertTrue(alteredConfigs) - EasyMock.reset(alterResult, describeResult) } + @Test + def testAlterIpConfig(): Unit = { + val (singleIpArgs, singleIpEntry) = toValues(Some("1.2.3.4"), ClientQuotaEntity.IP) + val singleIpEntity = new ClientQuotaEntity(singleIpEntry.asJava) + val (defaultIpArgs, defaultIpEntry) = toValues(Some(null), ClientQuotaEntity.IP) + val defaultIpEntity = new ClientQuotaEntity(defaultIpEntry.asJava) + + val deleteArgs = List("--delete-config", "connection_creation_rate") + val deleteAlterationOps = Set(new ClientQuotaAlteration.Op("connection_creation_rate", null)) + val propsToDelete = Map("connection_creation_rate" -> Double.box(50.0)) + + val addArgs = List("--add-config", "connection_creation_rate=100") + val addAlterationOps = Set(new ClientQuotaAlteration.Op("connection_creation_rate", 100.0)) + + verifyAlterQuotas(singleIpArgs ++ deleteArgs, singleIpEntity, propsToDelete, deleteAlterationOps) + verifyAlterQuotas(singleIpArgs ++ addArgs, singleIpEntity, Map.empty, addAlterationOps) + verifyAlterQuotas(defaultIpArgs ++ deleteArgs, defaultIpEntity, propsToDelete, deleteAlterationOps) + verifyAlterQuotas(defaultIpArgs ++ addArgs, defaultIpEntity, Map.empty, addAlterationOps) + } @Test def shouldAddClientConfig(): Unit = { - testShouldAddClientConfig(Some("test-user-1"), Some("test-client-1")) - testShouldAddClientConfig(Some("test-user-2"), Some(null)) - testShouldAddClientConfig(Some("test-user-3"), None) - testShouldAddClientConfig(Some(null), Some("test-client-2")) - testShouldAddClientConfig(Some(null), Some(null)) - testShouldAddClientConfig(Some(null), None) - testShouldAddClientConfig(None, Some("test-client-3")) - testShouldAddClientConfig(None, Some(null)) + val alterArgs = List("--add-config", "consumer_byte_rate=20000,producer_byte_rate=10000", + "--delete-config", "request_percentage") + val propsToDelete = Map("request_percentage" -> Double.box(50.0)) + + val alterationOps = Set( + new ClientQuotaAlteration.Op("consumer_byte_rate", Double.box(20000)), + new ClientQuotaAlteration.Op("producer_byte_rate", Double.box(10000)), + new ClientQuotaAlteration.Op("request_percentage", null) + ) + + def verifyAlterUserClientQuotas(userOpt: Option[String], clientOpt: Option[String]): Unit = { + val (userArgs, userEntry) = toValues(userOpt, ClientQuotaEntity.USER) + val (clientArgs, clientEntry) = toValues(clientOpt, ClientQuotaEntity.CLIENT_ID) + + val commandArgs = alterArgs ++ userArgs ++ clientArgs + val clientQuotaEntity = new ClientQuotaEntity((userEntry ++ clientEntry).asJava) + verifyAlterQuotas(commandArgs, clientQuotaEntity, propsToDelete, alterationOps) + } + verifyAlterUserClientQuotas(Some("test-user-1"), Some("test-client-1")) + verifyAlterUserClientQuotas(Some("test-user-2"), Some(null)) + verifyAlterUserClientQuotas(Some("test-user-3"), None) + verifyAlterUserClientQuotas(Some(null), Some("test-client-2")) + verifyAlterUserClientQuotas(Some(null), Some(null)) + verifyAlterUserClientQuotas(Some(null), None) + verifyAlterUserClientQuotas(None, Some("test-client-3")) + verifyAlterUserClientQuotas(None, Some(null)) } + private val userEntityOpts = List("--entity-type", "users", "--entity-name", "admin") + private val clientEntityOpts = List("--entity-type", "clients", "--entity-name", "admin") + private val addScramOpts = List("--add-config", "SCRAM-SHA-256=[iterations=8192,password=foo-secret]") + private val deleteScramOpts = List("--delete-config", "SCRAM-SHA-256") + @Test def shouldNotAlterNonQuotaNonScramUserOrClientConfigUsingBootstrapServer(): Unit = { // when using --bootstrap-server, it should be illegal to alter anything that is not a quota and not a SCRAM credential // for both user and client entities - val node = new Node(1, "localhost", 9092) - val mockAdminClient = new MockAdminClient(util.Collections.singletonList(node), node) - - def verifyCommand(entityType: String, alterOpts: String*): Unit = { - val opts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092", - "--entity-type", entityType, "--entity-name", "admin", - "--alter") ++ alterOpts) - val e = intercept[IllegalArgumentException] { - ConfigCommand.alterConfig(mockAdminClient, opts) - } - assertTrue(s"Unexpected exception: $e", e.getMessage.contains("some_config")) - } - - verifyCommand("users", "--add-config", "consumer_byte_rate=20000,producer_byte_rate=10000,some_config=10") - verifyCommand("users", "--add-config", "SCRAM-SHA-256=[iterations=8192,password=foo-secret],some_config=10") - verifyCommand("clients", "--add-config", "some_config=10") - verifyCommand("users", "--delete-config", "consumer_byte_rate,some_config") - verifyCommand("users", "--delete-config", "SCRAM-SHA-256,some_config") - verifyCommand("clients", "--delete-config", "some_config") + val invalidProp = "some_config" + verifyAlterCommandFails(invalidProp, userEntityOpts ++ + List("-add-config", "consumer_byte_rate=20000,producer_byte_rate=10000,some_config=10")) + verifyAlterCommandFails(invalidProp, userEntityOpts ++ + List("--add-config", "consumer_byte_rate=20000,producer_byte_rate=10000,some_config=10")) + verifyAlterCommandFails(invalidProp, clientEntityOpts ++ List("--add-config", "some_config=10")) + verifyAlterCommandFails(invalidProp, userEntityOpts ++ List("--delete-config", "consumer_byte_rate,some_config")) + verifyAlterCommandFails(invalidProp, userEntityOpts ++ List("--delete-config", "SCRAM-SHA-256,some_config")) + verifyAlterCommandFails(invalidProp, clientEntityOpts ++ List("--delete-config", "some_config")) } @Test def shouldNotAlterScramClientConfigUsingBootstrapServer(): Unit = { // when using --bootstrap-server, it should be illegal to alter SCRAM credentials for client entities - val node = new Node(1, "localhost", 9092) - val mockAdminClient = new MockAdminClient(util.Collections.singletonList(node), node) - - def verifyCommand(entityType: String, alterOpts: String*): Unit = { - val opts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092", - "--entity-type", entityType, "--entity-name", "admin", - "--alter") ++ alterOpts) - val e = intercept[IllegalArgumentException] { - ConfigCommand.alterConfig(mockAdminClient, opts) - } - assertTrue(s"Unexpected exception: $e", e.getMessage.contains("SCRAM-SHA-256")) - } - - verifyCommand("clients", "--add-config", "SCRAM-SHA-256=[iterations=8192,password=foo-secret]") - verifyCommand("clients", "--delete-config", "SCRAM-SHA-256") + verifyAlterCommandFails("SCRAM-SHA-256", clientEntityOpts ++ addScramOpts) + verifyAlterCommandFails("SCRAM-SHA-256", clientEntityOpts ++ deleteScramOpts) } @Test def shouldNotCreateUserScramCredentialConfigWithUnderMinimumIterationsUsingBootstrapServer(): Unit = { // when using --bootstrap-server, it should be illegal to create a SCRAM credential for a user // with an iterations value less than the minimum - val node = new Node(1, "localhost", 9092) - val mockAdminClient = new MockAdminClient(util.Collections.singletonList(node), node) - - def verifyCommand(entityType: String, alterOpts: String*): Unit = { - val opts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092", - "--entity-type", entityType, "--entity-name", "admin", - "--alter") ++ alterOpts) - val e = intercept[IllegalArgumentException] { - ConfigCommand.alterConfig(mockAdminClient, opts) - } - assertTrue(s"Unexpected exception: $e", e.getMessage.contains("SCRAM-SHA-256")) - } - - verifyCommand("users", "--add-config", "SCRAM-SHA-256=[iterations=100,password=foo-secret]") + verifyAlterCommandFails("SCRAM-SHA-256", userEntityOpts ++ List("--add-config", "SCRAM-SHA-256=[iterations=100,password=foo-secret]")) } @Test def shouldNotAlterUserScramCredentialAndClientQuotaConfigsSimultaneouslyUsingBootstrapServer(): Unit = { // when using --bootstrap-server, it should be illegal to alter both SCRAM credentials and quotas for user entities - val node = new Node(1, "localhost", 9092) - val mockAdminClient = new MockAdminClient(util.Collections.singletonList(node), node) + val expectedErrorMessage = "SCRAM-SHA-256" + val secondUserEntityOpts = List("--entity-type", "users", "--entity-name", "admin1") + val addQuotaOpts = List("--add-config", "consumer_byte_rate=20000") + val deleteQuotaOpts = List("--delete-config", "consumer_byte_rate") - def verifyCommand(alterOpts: String*): Unit = { - val opts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092", "--alter") ++ alterOpts) - val e = intercept[IllegalArgumentException] { - ConfigCommand.alterConfig(mockAdminClient, opts) - } - assertTrue(s"Unexpected exception: $e", e.getMessage.contains("SCRAM-SHA-256")) - } + verifyAlterCommandFails(expectedErrorMessage, userEntityOpts ++ addScramOpts ++ userEntityOpts ++ deleteQuotaOpts) + verifyAlterCommandFails(expectedErrorMessage, userEntityOpts ++ addScramOpts ++ secondUserEntityOpts ++ deleteQuotaOpts) + verifyAlterCommandFails(expectedErrorMessage, userEntityOpts ++ deleteScramOpts ++ userEntityOpts ++ addQuotaOpts) + verifyAlterCommandFails(expectedErrorMessage, userEntityOpts ++ deleteScramOpts ++ secondUserEntityOpts ++ addQuotaOpts) - verifyCommand("--entity-type", "users", "--entity-name", "admin", "--add-config", "SCRAM-SHA-256=[iterations=8192,password=foo-secret]", - "--entity-type", "users", "--entity-name", "admin", "--delete-config", "consumer_byte_rate") - verifyCommand("--entity-type", "users", "--entity-name", "admin", "--add-config", "SCRAM-SHA-256=[iterations=8192,password=foo-secret]", - "--entity-type", "users", "--entity-name", "admin1", "--delete-config", "consumer_byte_rate") - verifyCommand("--entity-type", "users", "--entity-name", "admin", "--delete-config", "SCRAM-SHA-256", - "--entity-type", "users", "--entity-name", "admin", "--add-config", "consumer_byte_rate=20000") - verifyCommand("--entity-type", "users", "--entity-name", "admin", "--delete-config", "SCRAM-SHA-256", - "--entity-type", "users", "--entity-name", "admin1", "--add-config", "consumer_byte_rate=20000") - - verifyCommand("--entity-type", "clients", "--entity-name", "admin", "--delete-config", "consumer_byte_rate", - "--entity-type", "users", "--entity-name", "admin", "--add-config", "SCRAM-SHA-256=[iterations=8192,password=foo-secret]") - verifyCommand( "--entity-type", "clients", "--entity-name", "admin1", "--delete-config", "consumer_byte_rate", - "--entity-type", "users", "--entity-name", "admin", "--add-config", "SCRAM-SHA-256=[iterations=8192,password=foo-secret]") - verifyCommand("--entity-type", "clients", "--entity-name", "admin", "--add-config", "consumer_byte_rate=20000", - "--entity-type", "users", "--entity-name", "admin", "--delete-config", "SCRAM-SHA-256") - verifyCommand("--entity-type", "users", "--entity-name", "admin1", "--add-config", "consumer_byte_rate=20000", - "--entity-type", "users", "--entity-name", "admin", "--delete-config", "SCRAM-SHA-256") + // change order of quota/SCRAM commands, verify alter still fails + verifyAlterCommandFails(expectedErrorMessage, userEntityOpts ++ deleteQuotaOpts ++ userEntityOpts ++ addScramOpts) + verifyAlterCommandFails(expectedErrorMessage, secondUserEntityOpts ++ deleteQuotaOpts ++ userEntityOpts ++ addScramOpts) + verifyAlterCommandFails(expectedErrorMessage, userEntityOpts ++ addQuotaOpts ++ userEntityOpts ++ deleteScramOpts) + verifyAlterCommandFails(expectedErrorMessage, secondUserEntityOpts ++ addQuotaOpts ++ userEntityOpts ++ deleteScramOpts) } @Test def shouldNotDescribeUserScramCredentialsWithEntityDefaultUsingBootstrapServer(): Unit = { - // User SCRAM credentials should not be described when specifying - // --describe --entity-type users --entity-default (or --user-defaults) with --bootstrap-server - val describeFuture = new KafkaFutureImpl[util.Map[ClientQuotaEntity, util.Map[String, java.lang.Double]]] - describeFuture.complete(Map((new ClientQuotaEntity(Map("" -> "").asJava) -> Map(("request_percentage" -> Double.box(50.0))).asJava)).asJava) - val describeClientQuotasResult: DescribeClientQuotasResult = EasyMock.createNiceMock(classOf[DescribeClientQuotasResult]) - EasyMock.expect(describeClientQuotasResult.entities()).andReturn(describeFuture).times(2) - EasyMock.replay(describeClientQuotasResult) - - val node = new Node(1, "localhost", 9092) - val mockAdminClient = new MockAdminClient(util.Collections.singletonList(node), node) { - override def describeClientQuotas(filter: ClientQuotaFilter, options: DescribeClientQuotasOptions): DescribeClientQuotasResult = { - describeClientQuotasResult - } - override def describeUserScramCredentials(users: util.List[String], options: DescribeUserScramCredentialsOptions): DescribeUserScramCredentialsResult = { - throw new IllegalStateException("Incorrectly described SCRAM credentials when specifying --entity-default with --bootstrap-server") - } - } - - def verifyCommand(expectedMessage: String, alterOrDescribeOpt: String, requestOpts: String*): Unit = { - val opts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092", - alterOrDescribeOpt) ++ requestOpts) - if (alterOrDescribeOpt.equals("--describe")) - ConfigCommand.describeConfig(mockAdminClient, opts) // fails if describeUserScramCredentials() is invoked - else { - val e = intercept[IllegalArgumentException] { - ConfigCommand.alterConfig(mockAdminClient, opts) + def verifyUserScramCredentialsNotDescribed(requestOpts: List[String]): Unit = { + // User SCRAM credentials should not be described when specifying + // --describe --entity-type users --entity-default (or --user-defaults) with --bootstrap-server + val describeFuture = new KafkaFutureImpl[util.Map[ClientQuotaEntity, util.Map[String, java.lang.Double]]] + describeFuture.complete(Map((new ClientQuotaEntity(Map("" -> "").asJava) -> Map(("request_percentage" -> Double.box(50.0))).asJava)).asJava) + val describeClientQuotasResult: DescribeClientQuotasResult = EasyMock.createNiceMock(classOf[DescribeClientQuotasResult]) + EasyMock.expect(describeClientQuotasResult.entities()).andReturn(describeFuture) + EasyMock.replay(describeClientQuotasResult) + val node = new Node(1, "localhost", 9092) + val mockAdminClient = new MockAdminClient(util.Collections.singletonList(node), node) { + override def describeClientQuotas(filter: ClientQuotaFilter, options: DescribeClientQuotasOptions): DescribeClientQuotasResult = { + describeClientQuotasResult + } + override def describeUserScramCredentials(users: util.List[String], options: DescribeUserScramCredentialsOptions): DescribeUserScramCredentialsResult = { + throw new IllegalStateException("Incorrectly described SCRAM credentials when specifying --entity-default with --bootstrap-server") } - assertTrue(s"Unexpected exception: $e", e.getMessage.contains(expectedMessage)) } + val opts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092", "--describe") ++ requestOpts) + ConfigCommand.describeConfig(mockAdminClient, opts) // fails if describeUserScramCredentials() is invoked } val expectedMsg = "The use of --entity-default or --user-defaults is not allowed with User SCRAM Credentials using --bootstrap-server." - verifyCommand(expectedMsg, "--alter", "--entity-type", "users", "--entity-default", "--add-config", "SCRAM-SHA-256=[iterations=8192,password=foo-secret]") - verifyCommand(expectedMsg, "--alter", "--entity-type", "users", "--entity-default", "--delete-config", "SCRAM-SHA-256") - verifyCommand(expectedMsg, "--describe", "--entity-type", "users", "--entity-default") - verifyCommand(expectedMsg, "--alter", "--user-defaults", "--add-config", "SCRAM-SHA-256=[iterations=8192,password=foo-secret]") - verifyCommand(expectedMsg, "--alter", "--user-defaults", "--delete-config", "SCRAM-SHA-256") - verifyCommand(expectedMsg, "--describe", "--user-defaults") + val defaultUserOpt = List("--user-defaults") + val verboseDefaultUserOpts = List("--entity-type", "users", "--entity-default") + verifyAlterCommandFails(expectedMsg, verboseDefaultUserOpts ++ addScramOpts) + verifyAlterCommandFails(expectedMsg, verboseDefaultUserOpts ++ deleteScramOpts) + verifyUserScramCredentialsNotDescribed(verboseDefaultUserOpts) + verifyAlterCommandFails(expectedMsg, defaultUserOpt ++ addScramOpts) + verifyAlterCommandFails(expectedMsg, defaultUserOpt ++ deleteScramOpts) + verifyUserScramCredentialsNotDescribed(defaultUserOpt) } @Test diff --git a/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala b/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala index f012e49dd1c87..b72f3b9121b17 100644 --- a/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala +++ b/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala @@ -17,6 +17,8 @@ package kafka.server +import java.net.InetAddress + import org.apache.kafka.clients.admin.{ScramCredentialInfo, ScramMechanism, UserScramCredentialUpsertion} import org.apache.kafka.common.errors.{InvalidRequestException, UnsupportedVersionException} import org.apache.kafka.common.internals.KafkaFutureImpl @@ -33,6 +35,7 @@ class ClientQuotasRequestTest extends BaseRequestTest { private val ConsumerByteRateProp = DynamicConfig.Client.ConsumerByteRateOverrideProp private val ProducerByteRateProp = DynamicConfig.Client.ProducerByteRateOverrideProp private val RequestPercentageProp = DynamicConfig.Client.RequestPercentageOverrideProp + private val IpConnectionRateProp = DynamicConfig.Ip.IpConnectionRateOverrideProp override val brokerCount = 1 @@ -187,6 +190,59 @@ class ClientQuotasRequestTest extends BaseRequestTest { )) } + @Test + def testAlterIpQuotasRequest(): Unit = { + val knownHost = "1.2.3.4" + val unknownHost = "2.3.4.5" + val entity = toIpEntity(Some(knownHost)) + val defaultEntity = toIpEntity(Some(null)) + val entityFilter = ClientQuotaFilterComponent.ofEntity(ClientQuotaEntity.IP, knownHost) + val defaultEntityFilter = ClientQuotaFilterComponent.ofDefaultEntity(ClientQuotaEntity.IP) + val allIpEntityFilter = ClientQuotaFilterComponent.ofEntityType(ClientQuotaEntity.IP) + + def verifyIpQuotas(entityFilter: ClientQuotaFilterComponent, expectedMatches: Map[ClientQuotaEntity, Double]): Unit = { + val result = describeClientQuotas(ClientQuotaFilter.containsOnly(List(entityFilter).asJava)) + assertEquals(expectedMatches.keySet, result.asScala.keySet) + result.asScala.foreach { case (entity, props) => + assertEquals(Set(IpConnectionRateProp), props.asScala.keySet) + assertEquals(expectedMatches(entity), props.get(IpConnectionRateProp)) + val entityName = entity.entries.get(ClientQuotaEntity.IP) + // ClientQuotaEntity with null name maps to default entity + val entityIp = if (entityName == null) + InetAddress.getByName(unknownHost) + else + InetAddress.getByName(entityName) + assertEquals(expectedMatches(entity), servers.head.socketServer.connectionQuotas.connectionRateForIp(entityIp), 0.01) + } + } + + // Expect an empty configuration. + verifyIpQuotas(allIpEntityFilter, Map.empty) + + // Add a configuration entry. + alterEntityQuotas(entity, Map(IpConnectionRateProp -> Some(100.0)), validateOnly = false) + verifyIpQuotas(entityFilter, Map(entity -> 100.0)) + + // update existing entry + alterEntityQuotas(entity, Map(IpConnectionRateProp -> Some(150.0)), validateOnly = false) + verifyIpQuotas(entityFilter, Map(entity -> 150.0)) + + // update default value + alterEntityQuotas(defaultEntity, Map(IpConnectionRateProp -> Some(200.0)), validateOnly = false) + verifyIpQuotas(defaultEntityFilter, Map(defaultEntity -> 200.0)) + + // describe all IP quotas + verifyIpQuotas(allIpEntityFilter, Map(entity -> 150.0, defaultEntity -> 200.0)) + + // remove entry + alterEntityQuotas(entity, Map(IpConnectionRateProp -> None), validateOnly = false) + verifyIpQuotas(entityFilter, Map.empty) + + // remove default value + alterEntityQuotas(defaultEntity, Map(IpConnectionRateProp -> None), validateOnly = false) + verifyIpQuotas(allIpEntityFilter, Map.empty) + } + @Test(expected = classOf[InvalidRequestException]) def testAlterClientQuotasBadUser(): Unit = { val entity = new ClientQuotaEntity(Map((ClientQuotaEntity.USER -> "")).asJava) @@ -223,8 +279,47 @@ class ClientQuotasRequestTest extends BaseRequestTest { alterEntityQuotas(entity, Map((ProducerByteRateProp -> Some(10000.5))), validateOnly = true) } + private def expectInvalidRequestWithMessage(runnable: => Unit, expectedMessage: String): Unit = { + val exception = assertThrows(classOf[InvalidRequestException], () => runnable) + assertTrue(s"Expected message $exception to contain $expectedMessage", exception.getMessage.contains(expectedMessage)) + } + + @Test + def testAlterClientQuotasInvalidEntityCombination(): Unit = { + val userAndIpEntity = new ClientQuotaEntity(Map(ClientQuotaEntity.USER -> "user", ClientQuotaEntity.IP -> "1.2.3.4").asJava) + val clientAndIpEntity = new ClientQuotaEntity(Map(ClientQuotaEntity.CLIENT_ID -> "client", ClientQuotaEntity.IP -> "1.2.3.4").asJava) + val expectedExceptionMessage = "Invalid quota entity combination" + expectInvalidRequestWithMessage(alterEntityQuotas(userAndIpEntity, Map(RequestPercentageProp -> Some(12.34)), + validateOnly = true), expectedExceptionMessage) + expectInvalidRequestWithMessage(alterEntityQuotas(clientAndIpEntity, Map(RequestPercentageProp -> Some(12.34)), + validateOnly = true), expectedExceptionMessage) + } + + @Test + def testAlterClientQuotasBadIp(): Unit = { + val invalidHostPatternEntity = new ClientQuotaEntity(Map(ClientQuotaEntity.IP -> "abc-123").asJava) + val unresolvableHostEntity = new ClientQuotaEntity(Map(ClientQuotaEntity.IP -> "ip").asJava) + val expectedExceptionMessage = "not a valid IP" + expectInvalidRequestWithMessage(alterEntityQuotas(invalidHostPatternEntity, Map(IpConnectionRateProp -> Some(50.0)), + validateOnly = true), expectedExceptionMessage) + expectInvalidRequestWithMessage(alterEntityQuotas(unresolvableHostEntity, Map(IpConnectionRateProp -> Some(50.0)), + validateOnly = true), expectedExceptionMessage) + } + + @Test + def testDescribeClientQuotasInvalidFilterCombination(): Unit = { + val ipFilterComponent = ClientQuotaFilterComponent.ofEntityType(ClientQuotaEntity.IP) + val userFilterComponent = ClientQuotaFilterComponent.ofEntityType(ClientQuotaEntity.USER) + val clientIdFilterComponent = ClientQuotaFilterComponent.ofEntityType(ClientQuotaEntity.CLIENT_ID) + val expectedExceptionMessage = "Invalid entity filter component combination" + expectInvalidRequestWithMessage(describeClientQuotas(ClientQuotaFilter.contains(List(ipFilterComponent, userFilterComponent).asJava)), + expectedExceptionMessage) + expectInvalidRequestWithMessage(describeClientQuotas(ClientQuotaFilter.contains(List(ipFilterComponent, clientIdFilterComponent).asJava)), + expectedExceptionMessage) + } + // Entities to be matched against. - private val matchEntities = List( + private val matchUserClientEntities = List( (Some("user-1"), Some("client-id-1"), 50.50), (Some("user-2"), Some("client-id-1"), 51.51), (Some("user-3"), Some("client-id-2"), 52.52), @@ -236,16 +331,22 @@ class ClientQuotasRequestTest extends BaseRequestTest { (Some("user-3"), None, 58.58), (Some(null), None, 59.59), (None, Some("client-id-2"), 60.60) - ).map { case (u, c, v) => (toEntity(u, c), v) } + ).map { case (u, c, v) => (toClientEntity(u, c), v) } - private def setupDescribeClientQuotasMatchTest() = { - val result = alterClientQuotas(matchEntities.map { case (e, v) => - (e -> Map((RequestPercentageProp, Some(v)))) - }.toMap, validateOnly = false) - matchEntities.foreach(e => result.get(e._1).get.get(10, TimeUnit.SECONDS)) + private val matchIpEntities = List( + (Some("1.2.3.4"), 10.0), + (Some("2.3.4.5"), 20.0) + ).map { case (ip, quota) => (toIpEntity(ip), quota)} - // Allow time for watch callbacks to be triggered. - Thread.sleep(500) + private def setupDescribeClientQuotasMatchTest() = { + val userClientQuotas = matchUserClientEntities.map { case (e, v) => + e -> Map((RequestPercentageProp, Some(v))) + }.toMap + val ipQuotas = matchIpEntities.map { case (e, v) => + e -> Map((IpConnectionRateProp, Some(v))) + }.toMap + val result = alterClientQuotas(userClientQuotas ++ ipQuotas, validateOnly = false) + (matchUserClientEntities ++ matchIpEntities).foreach(e => result(e._1).get(10, TimeUnit.SECONDS)) } @Test @@ -263,7 +364,7 @@ class ClientQuotasRequestTest extends BaseRequestTest { } // Test exact matches. - matchEntities.foreach { case (e, v) => + matchUserClientEntities.foreach { case (e, v) => val result = matchEntity(e) assertEquals(1, result.size) assertTrue(result.get(e) != null) @@ -299,19 +400,30 @@ class ClientQuotasRequestTest extends BaseRequestTest { def testMatchEntities(filter: ClientQuotaFilter, expectedMatchSize: Int, partition: ClientQuotaEntity => Boolean): Unit = { val result = describeClientQuotas(filter) - val (expectedMatches, expectedNonMatches) = matchEntities.partition(e => partition(e._1)) + val (expectedMatches, _) = (matchUserClientEntities ++ matchIpEntities).partition(e => partition(e._1)) assertEquals(expectedMatchSize, expectedMatches.size) // for test verification assertEquals(expectedMatchSize, result.size) val expectedMatchesMap = expectedMatches.toMap - matchEntities.foreach { case (entity, expectedValue) => + matchUserClientEntities.foreach { case (entity, expectedValue) => if (expectedMatchesMap.contains(entity)) { val config = result.get(entity) - assertTrue(config != null) + assertNotNull(config) val value = config.get(RequestPercentageProp) - assertTrue(value != null) + assertNotNull(value) + assertEquals(expectedValue, value, 1e-6) + } else { + assertNull(result.get(entity)) + } + } + matchIpEntities.foreach { case (entity, expectedValue) => + if (expectedMatchesMap.contains(entity)) { + val config = result.get(entity) + assertNotNull(config) + val value = config.get(IpConnectionRateProp) + assertNotNull(value) assertEquals(expectedValue, value, 1e-6) } else { - assertTrue(result.get(entity) == null) + assertNull(result.get(entity)) } } } @@ -376,8 +488,14 @@ class ClientQuotasRequestTest extends BaseRequestTest { entity => entity.entries.containsKey(ClientQuotaEntity.CLIENT_ID) ) + // Match against all entities with IP type in an open-ended match. + testMatchEntities( + ClientQuotaFilter.contains(List(ClientQuotaFilterComponent.ofEntityType(ClientQuotaEntity.IP)).asJava), 2, + entity => entity.entries.containsKey(ClientQuotaEntity.IP) + ) + // Match open-ended empty filter list. This should match all entities. - testMatchEntities(ClientQuotaFilter.contains(List.empty.asJava), 11, entity => true) + testMatchEntities(ClientQuotaFilter.contains(List.empty.asJava), 13, entity => true) // Match close-ended empty filter list. This should match no entities. testMatchEntities(ClientQuotaFilter.containsOnly(List.empty.asJava), 0, entity => false) @@ -386,11 +504,7 @@ class ClientQuotasRequestTest extends BaseRequestTest { @Test def testClientQuotasUnsupportedEntityTypes(): Unit = { val entity = new ClientQuotaEntity(Map(("other" -> "name")).asJava) - try { - verifyDescribeEntityQuotas(entity, Map()) - } catch { - case e: ExecutionException => assertTrue(e.getCause.isInstanceOf[UnsupportedVersionException]) - } + assertThrows(classOf[UnsupportedVersionException], () => verifyDescribeEntityQuotas(entity, Map.empty)) } @Test @@ -422,7 +536,11 @@ class ClientQuotasRequestTest extends BaseRequestTest { } private def verifyDescribeEntityQuotas(entity: ClientQuotaEntity, quotas: Map[String, Double]) = { - val components = entity.entries.asScala.map(e => ClientQuotaFilterComponent.ofEntity(e._1, e._2)) + val components = entity.entries.asScala.map { case (entityType, entityName) => + Option(entityName).map{ name => ClientQuotaFilterComponent.ofEntity(entityType, name)} + .getOrElse(ClientQuotaFilterComponent.ofDefaultEntity(entityType) + ) + } val describe = describeClientQuotas(ClientQuotaFilter.containsOnly(components.toList.asJava)) if (quotas.isEmpty) { assertEquals(0, describe.size) @@ -439,13 +557,17 @@ class ClientQuotasRequestTest extends BaseRequestTest { } } - private def toEntity(user: Option[String], clientId: Option[String]) = + private def toClientEntity(user: Option[String], clientId: Option[String]) = new ClientQuotaEntity((user.map((ClientQuotaEntity.USER -> _)) ++ clientId.map((ClientQuotaEntity.CLIENT_ID -> _))).toMap.asJava) + private def toIpEntity(ip: Option[String]) = new ClientQuotaEntity(ip.map(ClientQuotaEntity.IP -> _).toMap.asJava) + private def describeClientQuotas(filter: ClientQuotaFilter) = { val result = new KafkaFutureImpl[java.util.Map[ClientQuotaEntity, java.util.Map[String, java.lang.Double]]] sendDescribeClientQuotasRequest(filter).complete(result) - result.get + try result.get catch { + case e: ExecutionException => throw e.getCause + } } private def sendDescribeClientQuotasRequest(filter: ClientQuotaFilter): DescribeClientQuotasResponse = { diff --git a/core/src/test/scala/unit/kafka/server/DynamicConfigTest.scala b/core/src/test/scala/unit/kafka/server/DynamicConfigTest.scala index 3f8ffdf6bb24a..ce65df9b72566 100644 --- a/core/src/test/scala/unit/kafka/server/DynamicConfigTest.scala +++ b/core/src/test/scala/unit/kafka/server/DynamicConfigTest.scala @@ -57,4 +57,9 @@ class DynamicConfigTest extends ZooKeeperTestHarness { def shouldFailIpConfigsWithInvalidIpv4Entity(): Unit = { adminZkClient.changeIpConfig("1,1.1.1", propsWith(DynamicConfig.Ip.IpConnectionRateOverrideProp, "2")); } + + @Test(expected = classOf[AdminOperationException]) + def shouldFailIpConfigsWithBadHost(): Unit = { + adminZkClient.changeIpConfig("ip", propsWith(DynamicConfig.Ip.IpConnectionRateOverrideProp, "2")); + } } diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala index 9ff32227e081f..094b0fab68547 100755 --- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala +++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala @@ -49,6 +49,7 @@ import org.apache.kafka.common.errors.{KafkaStorageException, UnknownTopicOrPart import org.apache.kafka.common.header.Header import org.apache.kafka.common.internals.Topic import org.apache.kafka.common.network.{ListenerName, Mode} +import org.apache.kafka.common.quota.{ClientQuotaAlteration, ClientQuotaEntity} import org.apache.kafka.common.record._ import org.apache.kafka.common.resource.ResourcePattern import org.apache.kafka.common.security.auth.SecurityProtocol @@ -1529,6 +1530,16 @@ object TestUtils extends Logging { adminClient.incrementalAlterConfigs(configs) } + def alterClientQuotas(adminClient: Admin, request: Map[ClientQuotaEntity, Map[String, Option[Double]]]): AlterClientQuotasResult = { + val entries = request.map { case (entity, alter) => + val ops = alter.map { case (key, value) => + new ClientQuotaAlteration.Op(key, value.map(Double.box).getOrElse(null)) + }.asJavaCollection + new ClientQuotaAlteration(entity, ops) + }.asJavaCollection + adminClient.alterClientQuotas(entries) + } + def assertLeader(client: Admin, topicPartition: TopicPartition, expectedLeader: Int): Unit = { waitForLeaderToBecome(client, topicPartition, Some(expectedLeader)) }