You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2025/02/06 09:48:24 UTC
(kafka) branch trunk updated: KAFKA-18530 Remove ZooKeeperInternals (#18641)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new a3d9d881e1c KAFKA-18530 Remove ZooKeeperInternals (#18641)
a3d9d881e1c is described below
commit a3d9d881e1c0c7fe47962cb42b7a19111b614fdd
Author: Ken Huang <s7...@gmail.com>
AuthorDate: Thu Feb 6 17:48:17 2025 +0800
KAFKA-18530 Remove ZooKeeperInternals (#18641)
Since zk has been removed in 4.0, config handlers no longer need to handle the "<default>" value. This PR streamlines the config update process by eliminating the unnecessary string checks for "<default>"
Reviewers: Christo Lolov <lo...@amazon.com>, Ismael Juma <is...@juma.me.uk>, Chia-Ping Tsai <ch...@gmail.com>
---
.../src/main/scala/kafka/admin/ConfigCommand.scala | 12 +-
.../scala/kafka/server/ClientQuotaManager.scala | 44 ++--
.../main/scala/kafka/server/ConfigHandler.scala | 4 +-
.../scala/kafka/server/DynamicBrokerConfig.scala | 12 +-
.../main/scala/kafka/server/DynamicConfig.scala | 14 +-
.../metadata/ClientQuotaMetadataManager.scala | 51 ++--
.../server/metadata/DynamicConfigPublisher.scala | 4 +-
.../server/DynamicBrokerReconfigurationTest.scala | 7 +-
.../metadata/ClientQuotaMetadataManagerTest.scala | 42 ++-
.../unit/kafka/server/ClientQuotaManagerTest.scala | 292 +++++++++++++--------
.../kafka/server/ClientQuotasRequestTest.scala | 16 +-
.../server/ClientRequestQuotaManagerTest.scala | 8 +-
.../ControllerMutationQuotaManagerTest.scala | 21 +-
.../kafka/server/config/ZooKeeperInternals.java | 27 --
14 files changed, 332 insertions(+), 222 deletions(-)
diff --git a/core/src/main/scala/kafka/admin/ConfigCommand.scala b/core/src/main/scala/kafka/admin/ConfigCommand.scala
index 1cf2a800de9..edf0ff3bb6f 100644
--- a/core/src/main/scala/kafka/admin/ConfigCommand.scala
+++ b/core/src/main/scala/kafka/admin/ConfigCommand.scala
@@ -36,6 +36,7 @@ import org.apache.kafka.server.config.{ConfigType, QuotaConfig}
import org.apache.kafka.server.util.{CommandDefaultOptions, CommandLineUtils}
import org.apache.kafka.storage.internals.log.LogConfig
+import java.net.{InetAddress, UnknownHostException}
import scala.jdk.CollectionConverters._
import scala.collection._
@@ -649,7 +650,7 @@ object ConfigCommand extends Logging {
if (hasEntityName && entityTypeVals.contains(ConfigType.IP)) {
Seq(entityName, ip).filter(options.has(_)).map(options.valueOf(_)).foreach { ipEntity =>
- if (!DynamicConfig.Ip.isValidIpEntity(ipEntity))
+ if (!isValidIpEntity(ipEntity))
throw new IllegalArgumentException(s"The entity name for ${entityTypeVals.head} must be a valid IP or resolvable host, but it is: $ipEntity")
}
}
@@ -688,4 +689,13 @@ object ConfigCommand extends Logging {
}
}
}
+
+ def isValidIpEntity(ip: String): Boolean = {
+ try {
+ InetAddress.getByName(ip)
+ } catch {
+ case _: UnknownHostException => return false
+ }
+ true
+ }
}
diff --git a/core/src/main/scala/kafka/server/ClientQuotaManager.scala b/core/src/main/scala/kafka/server/ClientQuotaManager.scala
index f621d0dbcf4..d8346e6ab85 100644
--- a/core/src/main/scala/kafka/server/ClientQuotaManager.scala
+++ b/core/src/main/scala/kafka/server/ClientQuotaManager.scala
@@ -29,7 +29,7 @@ import org.apache.kafka.common.metrics.Metrics
import org.apache.kafka.common.metrics.stats.{Avg, CumulativeSum, Rate}
import org.apache.kafka.common.security.auth.KafkaPrincipal
import org.apache.kafka.common.utils.{Sanitizer, Time}
-import org.apache.kafka.server.config.{ClientQuotaManagerConfig, ZooKeeperInternals}
+import org.apache.kafka.server.config.ClientQuotaManagerConfig
import org.apache.kafka.server.quota.{ClientQuotaCallback, ClientQuotaEntity, ClientQuotaType, QuotaType, QuotaUtils, SensorAccess, ThrottleCallback, ThrottledChannel}
import org.apache.kafka.server.util.ShutdownableThread
import org.apache.kafka.network.Session
@@ -55,7 +55,7 @@ object QuotaTypes {
object ClientQuotaManager {
// Purge sensors after 1 hour of inactivity
val InactiveSensorExpirationTimeSeconds = 3600
-
+ private val DefaultName = "<default>"
val DefaultClientIdQuotaEntity: KafkaQuotaEntity = KafkaQuotaEntity(None, Some(DefaultClientIdEntity))
val DefaultUserQuotaEntity: KafkaQuotaEntity = KafkaQuotaEntity(Some(DefaultUserEntity), None)
val DefaultUserClientIdQuotaEntity: KafkaQuotaEntity = KafkaQuotaEntity(Some(DefaultUserEntity), Some(DefaultClientIdEntity))
@@ -76,13 +76,13 @@ object ClientQuotaManager {
case object DefaultUserEntity extends BaseUserEntity {
override def entityType: ClientQuotaEntity.ConfigEntityType = ClientQuotaEntity.ConfigEntityType.DEFAULT_USER
- override def name: String = ZooKeeperInternals.DEFAULT_STRING
+ override def name: String = DefaultName
override def toString: String = "default user"
}
case object DefaultClientIdEntity extends ClientQuotaEntity.ConfigEntity {
override def entityType: ClientQuotaEntity.ConfigEntityType = ClientQuotaEntity.ConfigEntityType.DEFAULT_CLIENT_ID
- override def name: String = ZooKeeperInternals.DEFAULT_STRING
+ override def name: String = DefaultName
override def toString: String = "default client-id"
}
@@ -93,7 +93,7 @@ object ClientQuotaManager {
def sanitizedUser: String = userEntity.map {
case entity: UserEntity => entity.sanitizedUser
- case DefaultUserEntity => ZooKeeperInternals.DEFAULT_STRING
+ case DefaultUserEntity => DefaultName
}.getOrElse("")
def clientId: String = clientIdEntity.map(_.name).getOrElse("")
@@ -260,7 +260,7 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
*/
def unrecordQuotaSensor(request: RequestChannel.Request, value: Double, timeMs: Long): Unit = {
val clientSensors = getOrCreateQuotaSensors(request.session, request.header.clientId)
- clientSensors.quotaSensor.record(value * (-1), timeMs, false)
+ clientSensors.quotaSensor.record(value * -1, timeMs, false)
}
/**
@@ -403,12 +403,15 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
* Overrides quotas for <user>, <client-id> or <user, client-id> or the dynamic defaults
* for any of these levels.
*
- * @param sanitizedUser user to override if quota applies to <user> or <user, client-id>
- * @param clientId client to override if quota applies to <client-id> or <user, client-id>
- * @param sanitizedClientId sanitized client ID to override if quota applies to <client-id> or <user, client-id>
- * @param quota custom quota to apply or None if quota override is being removed
+ * @param userEntity user to override if quota applies to <user> or <user, client-id>
+ * @param clientEntity sanitized client entity to override if quota applies to <client-id> or <user, client-id>
+ * @param quota custom quota to apply or None if quota override is being removed
*/
- def updateQuota(sanitizedUser: Option[String], clientId: Option[String], sanitizedClientId: Option[String], quota: Option[Quota]): Unit = {
+ def updateQuota(
+ userEntity: Option[BaseUserEntity],
+ clientEntity: Option[ClientQuotaEntity.ConfigEntity],
+ quota: Option[Quota]
+ ): Unit = {
/*
* Acquire the write lock to apply changes in the quota objects.
* This method changes the quota in the overriddenQuota map and applies the update on the actual KafkaMetric object (if it exists).
@@ -418,29 +421,21 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
*/
lock.writeLock().lock()
try {
- val userEntity = sanitizedUser.map {
- case ZooKeeperInternals.DEFAULT_STRING => DefaultUserEntity
- case user => UserEntity(user)
- }
- val clientIdEntity = sanitizedClientId.map {
- case ZooKeeperInternals.DEFAULT_STRING => DefaultClientIdEntity
- case _ => ClientIdEntity(clientId.getOrElse(throw new IllegalStateException("Client-id not provided")))
- }
- val quotaEntity = KafkaQuotaEntity(userEntity, clientIdEntity)
+ val quotaEntity = KafkaQuotaEntity(userEntity, clientEntity)
if (userEntity.nonEmpty) {
if (quotaEntity.clientIdEntity.nonEmpty)
quotaTypesEnabled |= QuotaTypes.UserClientIdQuotaEnabled
else
quotaTypesEnabled |= QuotaTypes.UserQuotaEnabled
- } else if (clientIdEntity.nonEmpty)
+ } else if (clientEntity.nonEmpty)
quotaTypesEnabled |= QuotaTypes.ClientIdQuotaEnabled
quota match {
case Some(newQuota) => quotaCallback.updateQuota(clientQuotaType, quotaEntity, newQuota.bound)
case None => quotaCallback.removeQuota(clientQuotaType, quotaEntity)
}
- val updatedEntity = if (userEntity.contains(DefaultUserEntity) || clientIdEntity.contains(DefaultClientIdEntity))
+ val updatedEntity = if (userEntity.contains(DefaultUserEntity) || clientEntity.contains(DefaultClientIdEntity))
None // more than one entity may need updating, so `updateQuotaMetricConfigs` will go through all metrics
else
Some(quotaEntity)
@@ -452,9 +447,8 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
}
/**
- * Updates metrics configs. This is invoked when quota configs are updated in ZooKeeper
- * or when partitions leaders change and custom callbacks that implement partition-based quotas
- * have updated quotas.
+ * Updates metrics configs. This is invoked when quota configs are updated when partitions leaders change
+ * and custom callbacks that implement partition-based quotas have updated quotas.
*
* @param updatedQuotaEntity If set to one entity and quotas have only been enabled at one
* level, then an optimized update is performed with a single metric update. If None is provided,
diff --git a/core/src/main/scala/kafka/server/ConfigHandler.scala b/core/src/main/scala/kafka/server/ConfigHandler.scala
index aba4f9f7767..0892afcfc6f 100644
--- a/core/src/main/scala/kafka/server/ConfigHandler.scala
+++ b/core/src/main/scala/kafka/server/ConfigHandler.scala
@@ -21,7 +21,7 @@ import java.util.{Collections, Properties}
import kafka.log.UnifiedLog
import kafka.server.QuotaFactory.QuotaManagers
import kafka.utils.Logging
-import org.apache.kafka.server.config.{QuotaConfig, ZooKeeperInternals}
+import org.apache.kafka.server.config.QuotaConfig
import org.apache.kafka.common.metrics.Quota._
import org.apache.kafka.coordinator.group.GroupCoordinator
import org.apache.kafka.server.ClientMetricsManager
@@ -146,7 +146,7 @@ class TopicConfigHandler(private val replicaManager: ReplicaManager,
class BrokerConfigHandler(private val brokerConfig: KafkaConfig,
private val quotaManagers: QuotaManagers) extends ConfigHandler with Logging {
def processConfigChanges(brokerId: String, properties: Properties): Unit = {
- if (brokerId == ZooKeeperInternals.DEFAULT_STRING)
+ if (brokerId.isEmpty)
brokerConfig.dynamicConfig.updateDefaultConfig(properties)
else if (brokerConfig.brokerId == brokerId.trim.toInt) {
brokerConfig.dynamicConfig.updateBrokerConfig(brokerConfig.brokerId, properties)
diff --git a/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala b/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
index 5887bd98c1b..aed5c57cbeb 100755
--- a/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
+++ b/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
@@ -46,12 +46,12 @@ import scala.collection._
import scala.jdk.CollectionConverters._
/**
- * Dynamic broker configurations are stored in ZooKeeper and may be defined at two levels:
+ * Dynamic broker configurations may be defined at two levels:
* <ul>
- * <li>Per-broker configs persisted at <tt>/configs/brokers/{brokerId}</tt>: These can be described/altered
- * using AdminClient using the resource name brokerId.</li>
- * <li>Cluster-wide defaults persisted at <tt>/configs/brokers/<default></tt>: These can be described/altered
- * using AdminClient using an empty resource name.</li>
+ * <li>Per-broker configurations are persisted at the controller and can be described
+ * or altered using AdminClient with the resource name brokerId.</li>
+ * <li>Cluster-wide default configurations are persisted at the cluster level and can be
+ * described or altered using AdminClient with an empty resource name.</li>
* </ul>
* The order of precedence for broker configs is:
* <ol>
@@ -367,7 +367,7 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends Logging
error(s"$errorMessage: $invalidPropNames")
}
}
- removeInvalidProps(nonDynamicConfigs(props), "Non-dynamic configs configured in ZooKeeper will be ignored")
+ removeInvalidProps(nonDynamicConfigs(props), "Non-dynamic configs will be ignored")
removeInvalidProps(securityConfigsWithoutListenerPrefix(props),
"Security configs can be dynamically updated only using listener prefix, base configs will be ignored")
if (!perBrokerConfig)
diff --git a/core/src/main/scala/kafka/server/DynamicConfig.scala b/core/src/main/scala/kafka/server/DynamicConfig.scala
index 988c567f1cf..7a401ec1eb4 100644
--- a/core/src/main/scala/kafka/server/DynamicConfig.scala
+++ b/core/src/main/scala/kafka/server/DynamicConfig.scala
@@ -19,11 +19,10 @@ package kafka.server
import kafka.server.DynamicBrokerConfig.AllDynamicConfigs
-import java.net.{InetAddress, UnknownHostException}
import java.util.Properties
import org.apache.kafka.common.config.ConfigDef
import org.apache.kafka.coordinator.group.GroupConfig
-import org.apache.kafka.server.config.{QuotaConfig, ZooKeeperInternals}
+import org.apache.kafka.server.config.QuotaConfig
import java.util
import scala.jdk.CollectionConverters._
@@ -83,17 +82,6 @@ object DynamicConfig {
def names: util.Set[String] = ipConfigs.names
def validate(props: Properties): util.Map[String, AnyRef] = DynamicConfig.validate(ipConfigs, props, customPropsAllowed = false)
-
- def isValidIpEntity(ip: String): Boolean = {
- if (ip != ZooKeeperInternals.DEFAULT_STRING) {
- try {
- InetAddress.getByName(ip)
- } catch {
- case _: UnknownHostException => return false
- }
- }
- true
- }
}
object ClientMetrics {
diff --git a/core/src/main/scala/kafka/server/metadata/ClientQuotaMetadataManager.scala b/core/src/main/scala/kafka/server/metadata/ClientQuotaMetadataManager.scala
index a6d3133f186..8fae9941b41 100644
--- a/core/src/main/scala/kafka/server/metadata/ClientQuotaMetadataManager.scala
+++ b/core/src/main/scala/kafka/server/metadata/ClientQuotaMetadataManager.scala
@@ -18,15 +18,19 @@
package kafka.server.metadata
import kafka.network.ConnectionQuotas
+import kafka.server.ClientQuotaManager
+import kafka.server.ClientQuotaManager.BaseUserEntity
import kafka.server.QuotaFactory.QuotaManagers
+import kafka.server.metadata.ClientQuotaMetadataManager.transferToClientQuotaEntity
import kafka.utils.Logging
import org.apache.kafka.common.metrics.Quota
import org.apache.kafka.common.quota.ClientQuotaEntity
import org.apache.kafka.common.utils.Sanitizer
+import org.apache.kafka.server.quota.ClientQuotaEntity.{ConfigEntity => ClientQuotaConfigEntity}
import java.net.{InetAddress, UnknownHostException}
import org.apache.kafka.image.{ClientQuotaDelta, ClientQuotasDelta}
-import org.apache.kafka.server.config.{QuotaConfig, ZooKeeperInternals}
+import org.apache.kafka.server.config.QuotaConfig
import scala.jdk.OptionConverters.RichOptionalDouble
@@ -145,27 +149,42 @@ class ClientQuotaMetadataManager(private[metadata] val quotaManagers: QuotaManag
}
// Convert entity into Options with sanitized values for QuotaManagers
- val (sanitizedUser, sanitizedClientId) = quotaEntity match {
- case UserEntity(user) => (Some(Sanitizer.sanitize(user)), None)
- case DefaultUserEntity => (Some(ZooKeeperInternals.DEFAULT_STRING), None)
- case ClientIdEntity(clientId) => (None, Some(Sanitizer.sanitize(clientId)))
- case DefaultClientIdEntity => (None, Some(ZooKeeperInternals.DEFAULT_STRING))
- case ExplicitUserExplicitClientIdEntity(user, clientId) => (Some(Sanitizer.sanitize(user)), Some(Sanitizer.sanitize(clientId)))
- case ExplicitUserDefaultClientIdEntity(user) => (Some(Sanitizer.sanitize(user)), Some(ZooKeeperInternals.DEFAULT_STRING))
- case DefaultUserExplicitClientIdEntity(clientId) => (Some(ZooKeeperInternals.DEFAULT_STRING), Some(Sanitizer.sanitize(clientId)))
- case DefaultUserDefaultClientIdEntity => (Some(ZooKeeperInternals.DEFAULT_STRING), Some(ZooKeeperInternals.DEFAULT_STRING))
- case IpEntity(_) | DefaultIpEntity => throw new IllegalStateException("Should not see IP quota entities here")
- }
+ val (userEntity, clientEntity) = transferToClientQuotaEntity(quotaEntity)
val quotaValue = newValue.map(new Quota(_, true))
try {
manager.updateQuota(
- sanitizedUser = sanitizedUser,
- clientId = sanitizedClientId.map(Sanitizer.desanitize),
- sanitizedClientId = sanitizedClientId,
- quota = quotaValue)
+ userEntity = userEntity,
+ clientEntity = clientEntity,
+ quota = quotaValue
+ )
} catch {
case t: Throwable => error(s"Failed to update user-client quota $quotaEntity", t)
}
}
}
+
+object ClientQuotaMetadataManager {
+
+ def transferToClientQuotaEntity(quotaEntity: QuotaEntity): (Option[BaseUserEntity], Option[ClientQuotaConfigEntity]) = {
+ quotaEntity match {
+ case UserEntity(user) =>
+ (Some(ClientQuotaManager.UserEntity(Sanitizer.sanitize(user))), None)
+ case DefaultUserEntity =>
+ (Some(ClientQuotaManager.DefaultUserEntity), None)
+ case ClientIdEntity(clientId) =>
+ (None, Some(ClientQuotaManager.ClientIdEntity(clientId)))
+ case DefaultClientIdEntity =>
+ (None, Some(ClientQuotaManager.DefaultClientIdEntity))
+ case ExplicitUserExplicitClientIdEntity(user, clientId) =>
+ (Some(ClientQuotaManager.UserEntity(Sanitizer.sanitize(user))), Some(ClientQuotaManager.ClientIdEntity(clientId)))
+ case ExplicitUserDefaultClientIdEntity(user) =>
+ (Some(ClientQuotaManager.UserEntity(Sanitizer.sanitize(user))), Some(ClientQuotaManager.DefaultClientIdEntity))
+ case DefaultUserExplicitClientIdEntity(clientId) =>
+ (Some(ClientQuotaManager.DefaultUserEntity), Some(ClientQuotaManager.ClientIdEntity(clientId)))
+ case DefaultUserDefaultClientIdEntity =>
+ (Some(ClientQuotaManager.DefaultUserEntity), Some(ClientQuotaManager.DefaultClientIdEntity))
+ case IpEntity(_) | DefaultIpEntity => throw new IllegalStateException("Should not see IP quota entities here")
+ }
+ }
+}
\ No newline at end of file
diff --git a/core/src/main/scala/kafka/server/metadata/DynamicConfigPublisher.scala b/core/src/main/scala/kafka/server/metadata/DynamicConfigPublisher.scala
index 763aa387962..6904921fb0c 100644
--- a/core/src/main/scala/kafka/server/metadata/DynamicConfigPublisher.scala
+++ b/core/src/main/scala/kafka/server/metadata/DynamicConfigPublisher.scala
@@ -24,7 +24,7 @@ import kafka.utils.Logging
import org.apache.kafka.common.config.ConfigResource.Type.{BROKER, CLIENT_METRICS, GROUP, TOPIC}
import org.apache.kafka.image.loader.LoaderManifest
import org.apache.kafka.image.{MetadataDelta, MetadataImage}
-import org.apache.kafka.server.config.{ConfigType, ZooKeeperInternals}
+import org.apache.kafka.server.config.ConfigType
import org.apache.kafka.server.fault.FaultHandler
@@ -78,7 +78,7 @@ class DynamicConfigPublisher(
// These are stored in KRaft with an empty name field.
info("Updating cluster configuration : " +
toLoggableProps(resource, props).mkString(","))
- nodeConfigHandler.processConfigChanges(ZooKeeperInternals.DEFAULT_STRING, props)
+ nodeConfigHandler.processConfigChanges(resource.name(), props)
} catch {
case t: Throwable => faultHandler.handleFault("Error updating " +
s"cluster with new configuration: ${toLoggableProps(resource, props).mkString(",")} " +
diff --git a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala
index 49a0ebc21f4..f4de50d7cd8 100644
--- a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala
+++ b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala
@@ -934,8 +934,11 @@ class DynamicBrokerReconfigurationTest extends QuorumTestHarness with SaslSetup
// non-default value to trigger a new metric
val clientId = "test-client-1"
servers.foreach { server =>
- server.quotaManagers.produce.updateQuota(None, Some(clientId), Some(clientId),
- Some(Quota.upperBound(10000000)))
+ server.quotaManagers.produce.updateQuota(
+ None,
+ Some(ClientQuotaManager.ClientIdEntity(clientId)),
+ Some(Quota.upperBound(10000000))
+ )
}
val (producerThread, consumerThread) = startProduceConsume(retries = 0, groupProtocol, clientId)
TestUtils.waitUntilTrue(() => consumerThread.received >= 5, "Messages not sent")
diff --git a/core/src/test/scala/kafka/server/metadata/ClientQuotaMetadataManagerTest.scala b/core/src/test/scala/kafka/server/metadata/ClientQuotaMetadataManagerTest.scala
index 03d8d30cea0..9d36cae25c2 100644
--- a/core/src/test/scala/kafka/server/metadata/ClientQuotaMetadataManagerTest.scala
+++ b/core/src/test/scala/kafka/server/metadata/ClientQuotaMetadataManagerTest.scala
@@ -16,8 +16,9 @@
*/
package kafka.server.metadata
+import kafka.server.ClientQuotaManager
import org.apache.kafka.image.ClientQuotaDelta
-import org.junit.jupiter.api.Assertions.{assertDoesNotThrow, assertThrows}
+import org.junit.jupiter.api.Assertions.{assertDoesNotThrow, assertEquals, assertThrows}
import org.junit.jupiter.api.Test
import org.junit.jupiter.api.function.Executable
@@ -33,4 +34,43 @@ class ClientQuotaMetadataManagerTest {
assertDoesNotThrow { new Executable { def execute(): Unit = manager.handleIpQuota(IpEntity("192.168.1.1"), new ClientQuotaDelta(null)) } }
assertDoesNotThrow { new Executable { def execute(): Unit = manager.handleIpQuota(IpEntity("2001:db8::1"), new ClientQuotaDelta(null)) } }
}
+
+ @Test
+ def testTransferToClientQuotaEntity(): Unit = {
+
+ assertThrows(classOf[IllegalStateException],() => ClientQuotaMetadataManager.transferToClientQuotaEntity(IpEntity("a")))
+ assertThrows(classOf[IllegalStateException],() => ClientQuotaMetadataManager.transferToClientQuotaEntity(DefaultIpEntity))
+ assertEquals(
+ (Some(ClientQuotaManager.UserEntity("user")), None),
+ ClientQuotaMetadataManager.transferToClientQuotaEntity(UserEntity("user"))
+ )
+ assertEquals(
+ (Some(ClientQuotaManager.DefaultUserEntity), None),
+ ClientQuotaMetadataManager.transferToClientQuotaEntity(DefaultUserEntity)
+ )
+ assertEquals(
+ (None, Some(ClientQuotaManager.ClientIdEntity("client"))),
+ ClientQuotaMetadataManager.transferToClientQuotaEntity(ClientIdEntity("client"))
+ )
+ assertEquals(
+ (None, Some(ClientQuotaManager.DefaultClientIdEntity)),
+ ClientQuotaMetadataManager.transferToClientQuotaEntity(DefaultClientIdEntity)
+ )
+ assertEquals(
+ (Some(ClientQuotaManager.UserEntity("user")), Some(ClientQuotaManager.ClientIdEntity("client"))),
+ ClientQuotaMetadataManager.transferToClientQuotaEntity(ExplicitUserExplicitClientIdEntity("user", "client"))
+ )
+ assertEquals(
+ (Some(ClientQuotaManager.UserEntity("user")), Some(ClientQuotaManager.DefaultClientIdEntity)),
+ ClientQuotaMetadataManager.transferToClientQuotaEntity(ExplicitUserDefaultClientIdEntity("user"))
+ )
+ assertEquals(
+ (Some(ClientQuotaManager.DefaultUserEntity), Some(ClientQuotaManager.ClientIdEntity("client"))),
+ ClientQuotaMetadataManager.transferToClientQuotaEntity(DefaultUserExplicitClientIdEntity("client"))
+ )
+ assertEquals(
+ (Some(ClientQuotaManager.DefaultUserEntity), Some(ClientQuotaManager.DefaultClientIdEntity)),
+ ClientQuotaMetadataManager.transferToClientQuotaEntity(DefaultUserDefaultClientIdEntity)
+ )
+ }
}
\ No newline at end of file
diff --git a/core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala
index 4a4107226dc..6c268d3c3fb 100644
--- a/core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala
@@ -16,11 +16,12 @@
*/
package kafka.server
+import kafka.server.ClientQuotaManager.BaseUserEntity
+
import java.net.InetAddress
import org.apache.kafka.common.metrics.Quota
import org.apache.kafka.common.security.auth.KafkaPrincipal
-import org.apache.kafka.common.utils.Sanitizer
-import org.apache.kafka.server.config.{ClientQuotaManagerConfig, ZooKeeperInternals}
+import org.apache.kafka.server.config.ClientQuotaManagerConfig
import org.apache.kafka.network.Session
import org.apache.kafka.server.quota.QuotaType
import org.junit.jupiter.api.Assertions._
@@ -34,8 +35,16 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest {
try {
// Case 1: Update the quota. Assert that the new quota value is returned
- clientQuotaManager.updateQuota(client1.configUser, client1.configClientId, client1.sanitizedConfigClientId, Some(new Quota(2000, true)))
- clientQuotaManager.updateQuota(client2.configUser, client2.configClientId, client2.sanitizedConfigClientId, Some(new Quota(4000, true)))
+ clientQuotaManager.updateQuota(
+ client1.configUser,
+ client1.configClientEntity,
+ Some(new Quota(2000, true))
+ )
+ clientQuotaManager.updateQuota(
+ client2.configUser,
+ client2.configClientEntity,
+ Some(new Quota(4000, true))
+ )
assertEquals(Long.MaxValue.toDouble, clientQuotaManager.quota(randomClient.user, randomClient.clientId).bound, 0.0,
"Default producer quota should be " + Long.MaxValue.toDouble)
@@ -50,22 +59,38 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest {
// Case 2: Change quota again. The quota should be updated within KafkaMetrics as well since the sensor was created.
// p1 should not longer be throttled after the quota change
- clientQuotaManager.updateQuota(client1.configUser, client1.configClientId, client1.sanitizedConfigClientId, Some(new Quota(3000, true)))
+ clientQuotaManager.updateQuota(
+ client1.configUser,
+ client1.configClientEntity,
+ Some(new Quota(3000, true))
+ )
assertEquals(3000, clientQuotaManager.quota(client1.user, client1.clientId).bound, 0.0, "Should return the newly overridden value (3000)")
throttleTimeMs = maybeRecord(clientQuotaManager, client1.user, client1.clientId, 0)
assertEquals(0, throttleTimeMs, s"throttleTimeMs should be 0. was $throttleTimeMs")
// Case 3: Change quota back to default. Should be throttled again
- clientQuotaManager.updateQuota(client1.configUser, client1.configClientId, client1.sanitizedConfigClientId, Some(new Quota(500, true)))
+ clientQuotaManager.updateQuota(
+ client1.configUser,
+ client1.configClientEntity,
+ Some(new Quota(500, true))
+ )
assertEquals(500, clientQuotaManager.quota(client1.user, client1.clientId).bound, 0.0, "Should return the default value (500)")
throttleTimeMs = maybeRecord(clientQuotaManager, client1.user, client1.clientId, 0)
assertTrue(throttleTimeMs > 0, s"throttleTimeMs should be > 0. was $throttleTimeMs")
// Case 4: Set high default quota, remove p1 quota. p1 should no longer be throttled
- clientQuotaManager.updateQuota(client1.configUser, client1.configClientId, client1.sanitizedConfigClientId, None)
- clientQuotaManager.updateQuota(defaultConfigClient.configUser, defaultConfigClient.configClientId, defaultConfigClient.sanitizedConfigClientId, Some(new Quota(4000, true)))
+ clientQuotaManager.updateQuota(
+ client1.configUser,
+ client1.configClientEntity,
+ None
+ )
+ clientQuotaManager.updateQuota(
+ defaultConfigClient.configUser,
+ defaultConfigClient.configClientEntity,
+ Some(new Quota(4000, true))
+ )
assertEquals(4000, clientQuotaManager.quota(client1.user, client1.clientId).bound, 0.0, "Should return the newly overridden value (4000)")
throttleTimeMs = maybeRecord(clientQuotaManager, client1.user, client1.clientId, 1000 * config.numQuotaSamples)
@@ -76,68 +101,15 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest {
}
}
- /**
- * Tests parsing for <client-id> quotas.
- * Quota overrides persisted in ZooKeeper in /config/clients/<client-id>, default persisted in /config/clients/<default>
- */
- @Test
- def testClientIdQuotaParsing(): Unit = {
- val client1 = UserClient("ANONYMOUS", "p1", None, Some("p1"))
- val client2 = UserClient("ANONYMOUS", "p2", None, Some("p2"))
- val randomClient = UserClient("ANONYMOUS", "random-client-id", None, None)
- val defaultConfigClient = UserClient("", "", None, Some(ZooKeeperInternals.DEFAULT_STRING))
- testQuotaParsing(config, client1, client2, randomClient, defaultConfigClient)
- }
-
- /**
- * Tests parsing for <user> quotas.
- * Quota overrides persisted in ZooKeeper in /config/users/<user>, default persisted in /config/users/<default>
- */
- @Test
- def testUserQuotaParsing(): Unit = {
- val client1 = UserClient("User1", "p1", Some("User1"), None)
- val client2 = UserClient("User2", "p2", Some("User2"), None)
- val randomClient = UserClient("RandomUser", "random-client-id", None, None)
- val defaultConfigClient = UserClient("", "", Some(ZooKeeperInternals.DEFAULT_STRING), None)
- val config = new ClientQuotaManagerConfig()
- testQuotaParsing(config, client1, client2, randomClient, defaultConfigClient)
- }
-
- /**
- * Tests parsing for <user, client-id> quotas.
- * Quotas persisted in ZooKeeper in /config/users/<user>/clients/<client-id>, default in /config/users/<default>/clients/<default>
- */
- @Test
- def testUserClientIdQuotaParsing(): Unit = {
- val client1 = UserClient("User1", "p1", Some("User1"), Some("p1"))
- val client2 = UserClient("User2", "p2", Some("User2"), Some("p2"))
- val randomClient = UserClient("RandomUser", "random-client-id", None, None)
- val defaultConfigClient = UserClient("", "", Some(ZooKeeperInternals.DEFAULT_STRING), Some(ZooKeeperInternals.DEFAULT_STRING))
- val config = new ClientQuotaManagerConfig()
- testQuotaParsing(config, client1, client2, randomClient, defaultConfigClient)
- }
-
/**
* Tests parsing for <user> quotas when client-id default quota properties are set.
*/
@Test
def testUserQuotaParsingWithDefaultClientIdQuota(): Unit = {
- val client1 = UserClient("User1", "p1", Some("User1"), None)
- val client2 = UserClient("User2", "p2", Some("User2"), None)
- val randomClient = UserClient("RandomUser", "random-client-id", None, None)
- val defaultConfigClient = UserClient("", "", Some(ZooKeeperInternals.DEFAULT_STRING), None)
- testQuotaParsing(config, client1, client2, randomClient, defaultConfigClient)
- }
-
- /**
- * Tests parsing for <user, client-id> quotas when client-id default quota properties are set.
- */
- @Test
- def testUserClientQuotaParsingIdWithDefaultClientIdQuota(): Unit = {
- val client1 = UserClient("User1", "p1", Some("User1"), Some("p1"))
- val client2 = UserClient("User2", "p2", Some("User2"), Some("p2"))
+ val client1 = UserClient("User1", "p1", Some(ClientQuotaManager.UserEntity("User1")), None)
+ val client2 = UserClient("User2", "p2", Some(ClientQuotaManager.UserEntity("User2")), None)
val randomClient = UserClient("RandomUser", "random-client-id", None, None)
- val defaultConfigClient = UserClient("", "", Some(ZooKeeperInternals.DEFAULT_STRING), Some(ZooKeeperInternals.DEFAULT_STRING))
+ val defaultConfigClient = UserClient("", "", Some(ClientQuotaManager.DefaultUserEntity), None)
testQuotaParsing(config, client1, client2, randomClient, defaultConfigClient)
}
@@ -168,7 +140,11 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest {
assertEquals(Double.MaxValue, clientQuotaManager.getMaxValueInQuotaWindow(userSession, "client1"), 0.01)
// Set default <user> quota config
- clientQuotaManager.updateQuota(Some(ZooKeeperInternals.DEFAULT_STRING), None, None, Some(new Quota(10, true)))
+ clientQuotaManager.updateQuota(
+ Some(ClientQuotaManager.DefaultUserEntity),
+ None,
+ Some(new Quota(10, true))
+ )
assertEquals(10 * numFullQuotaWindows, clientQuotaManager.getMaxValueInQuotaWindow(userSession, "client1"), 0.01)
} finally {
clientQuotaManager.shutdown()
@@ -186,11 +162,19 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest {
checkQuota(clientQuotaManager, "userA", "client1", Long.MaxValue, 1000, expectThrottle = false)
// Set default <user> quota config
- clientQuotaManager.updateQuota(Some(ZooKeeperInternals.DEFAULT_STRING), None, None, Some(new Quota(10, true)))
+ clientQuotaManager.updateQuota(
+ Some(ClientQuotaManager.DefaultUserEntity),
+ None,
+ Some(new Quota(10, true))
+ )
checkQuota(clientQuotaManager, "userA", "client1", 10, 1000, expectThrottle = true)
// Remove default <user> quota config, back to no quotas
- clientQuotaManager.updateQuota(Some(ZooKeeperInternals.DEFAULT_STRING), None, None, None)
+ clientQuotaManager.updateQuota(
+ Some(ClientQuotaManager.DefaultUserEntity),
+ None,
+ None
+ )
checkQuota(clientQuotaManager, "userA", "client1", Long.MaxValue, 1000, expectThrottle = false)
} finally {
clientQuotaManager.shutdown()
@@ -205,11 +189,19 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest {
try {
// Set <user> quota config
- clientQuotaManager.updateQuota(Some("userA"), None, None, Some(new Quota(10, true)))
+ clientQuotaManager.updateQuota(
+ Some(ClientQuotaManager.UserEntity("userA")),
+ None,
+ Some(new Quota(10, true))
+ )
checkQuota(clientQuotaManager, "userA", "client1", 10, 1000, expectThrottle = true)
// Remove <user> quota config, back to no quotas
- clientQuotaManager.updateQuota(Some("userA"), None, None, None)
+ clientQuotaManager.updateQuota(
+ Some(ClientQuotaManager.UserEntity("userA")),
+ None,
+ None
+ )
checkQuota(clientQuotaManager, "userA", "client1", Long.MaxValue, 1000, expectThrottle = false)
} finally {
clientQuotaManager.shutdown()
@@ -224,11 +216,19 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest {
try {
// Set <user, client-id> quota config
- clientQuotaManager.updateQuota(Some("userA"), Some("client1"), Some("client1"), Some(new Quota(10, true)))
+ clientQuotaManager.updateQuota(
+ Some(ClientQuotaManager.UserEntity("userA")),
+ Some(ClientQuotaManager.ClientIdEntity("client1")),
+ Some(new Quota(10, true))
+ )
checkQuota(clientQuotaManager, "userA", "client1", 10, 1000, expectThrottle = true)
// Remove <user, client-id> quota config, back to no quotas
- clientQuotaManager.updateQuota(Some("userA"), Some("client1"), Some("client1"), None)
+ clientQuotaManager.updateQuota(
+ Some(ClientQuotaManager.UserEntity("userA")),
+ Some(ClientQuotaManager.ClientIdEntity("client1")),
+ None
+ )
checkQuota(clientQuotaManager, "userA", "client1", Long.MaxValue, 1000, expectThrottle = false)
} finally {
clientQuotaManager.shutdown()
@@ -241,16 +241,56 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest {
metrics, QuotaType.PRODUCE, time, "")
try {
- clientQuotaManager.updateQuota(Some(ZooKeeperInternals.DEFAULT_STRING), None, None, Some(new Quota(1000, true)))
- clientQuotaManager.updateQuota(None, Some(ZooKeeperInternals.DEFAULT_STRING), Some(ZooKeeperInternals.DEFAULT_STRING), Some(new Quota(2000, true)))
- clientQuotaManager.updateQuota(Some(ZooKeeperInternals.DEFAULT_STRING), Some(ZooKeeperInternals.DEFAULT_STRING), Some(ZooKeeperInternals.DEFAULT_STRING), Some(new Quota(3000, true)))
- clientQuotaManager.updateQuota(Some("userA"), None, None, Some(new Quota(4000, true)))
- clientQuotaManager.updateQuota(Some("userA"), Some("client1"), Some("client1"), Some(new Quota(5000, true)))
- clientQuotaManager.updateQuota(Some("userB"), None, None, Some(new Quota(6000, true)))
- clientQuotaManager.updateQuota(Some("userB"), Some("client1"), Some("client1"), Some(new Quota(7000, true)))
- clientQuotaManager.updateQuota(Some("userB"), Some(ZooKeeperInternals.DEFAULT_STRING), Some(ZooKeeperInternals.DEFAULT_STRING), Some(new Quota(8000, true)))
- clientQuotaManager.updateQuota(Some("userC"), None, None, Some(new Quota(10000, true)))
- clientQuotaManager.updateQuota(None, Some("client1"), Some("client1"), Some(new Quota(9000, true)))
+ clientQuotaManager.updateQuota(
+ Some(ClientQuotaManager.DefaultUserEntity),
+ None,
+ Some(new Quota(1000, true))
+ )
+ clientQuotaManager.updateQuota(
+ None,
+ Some(ClientQuotaManager.DefaultClientIdEntity),
+ Some(new Quota(2000, true))
+ )
+ clientQuotaManager.updateQuota(
+ Some(ClientQuotaManager.DefaultUserEntity),
+ Some(ClientQuotaManager.DefaultClientIdEntity),
+ Some(new Quota(3000, true))
+ )
+ clientQuotaManager.updateQuota(
+ Some(ClientQuotaManager.UserEntity("userA")),
+ None,
+ Some(new Quota(4000, true))
+ )
+ clientQuotaManager.updateQuota(
+ Some(ClientQuotaManager.UserEntity("userA")),
+ Some(ClientQuotaManager.ClientIdEntity("client1")),
+ Some(new Quota(5000, true))
+ )
+ clientQuotaManager.updateQuota(
+ Some(ClientQuotaManager.UserEntity("userB")),
+ None,
+ Some(new Quota(6000, true))
+ )
+ clientQuotaManager.updateQuota(
+ Some(ClientQuotaManager.UserEntity("userB")),
+ Some(ClientQuotaManager.ClientIdEntity("client1")),
+ Some(new Quota(7000, true))
+ )
+ clientQuotaManager.updateQuota(
+ Some(ClientQuotaManager.UserEntity("userB")),
+ Some(ClientQuotaManager.DefaultClientIdEntity),
+ Some(new Quota(8000, true))
+ )
+ clientQuotaManager.updateQuota(
+ Some(ClientQuotaManager.UserEntity("userC")),
+ None,
+ Some(new Quota(10000, true))
+ )
+ clientQuotaManager.updateQuota(
+ None,
+ Some(ClientQuotaManager.ClientIdEntity("client1")),
+ Some(new Quota(9000, true))
+ )
checkQuota(clientQuotaManager, "userA", "client1", 5000, 4500, expectThrottle = false) // <user, client> quota takes precedence over <user>
checkQuota(clientQuotaManager, "userA", "client2", 4000, 4500, expectThrottle = true) // <user> quota takes precedence over <client> and defaults
@@ -266,32 +306,64 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest {
checkQuota(clientQuotaManager, "userE", "client1", 3000, 2500, expectThrottle = false)
// Remove default <user, client> quota config, revert to <user> default
- clientQuotaManager.updateQuota(Some(ZooKeeperInternals.DEFAULT_STRING), Some(ZooKeeperInternals.DEFAULT_STRING), Some(ZooKeeperInternals.DEFAULT_STRING), None)
+ clientQuotaManager.updateQuota(
+ Some(ClientQuotaManager.DefaultUserEntity),
+ Some(ClientQuotaManager.DefaultClientIdEntity),
+ None
+ )
checkQuota(clientQuotaManager, "userD", "client1", 1000, 0, expectThrottle = false) // Metrics tags changed, restart counter
checkQuota(clientQuotaManager, "userE", "client4", 1000, 1500, expectThrottle = true)
checkQuota(clientQuotaManager, "userF", "client4", 1000, 800, expectThrottle = false) // Default <user> quota shared across clients of user
checkQuota(clientQuotaManager, "userF", "client5", 1000, 800, expectThrottle = true)
// Remove default <user> quota config, revert to <client-id> default
- clientQuotaManager.updateQuota(Some(ZooKeeperInternals.DEFAULT_STRING), None, None, None)
+ clientQuotaManager.updateQuota(
+ Some(ClientQuotaManager.DefaultUserEntity),
+ None,
+ None
+ )
checkQuota(clientQuotaManager, "userF", "client4", 2000, 0, expectThrottle = false) // Default <client-id> quota shared across client-id of all users
checkQuota(clientQuotaManager, "userF", "client5", 2000, 0, expectThrottle = false)
checkQuota(clientQuotaManager, "userF", "client5", 2000, 2500, expectThrottle = true)
checkQuota(clientQuotaManager, "userG", "client5", 2000, 0, expectThrottle = true)
// Update quotas
- clientQuotaManager.updateQuota(Some("userA"), None, None, Some(new Quota(8000, true)))
- clientQuotaManager.updateQuota(Some("userA"), Some("client1"), Some("client1"), Some(new Quota(10000, true)))
+ clientQuotaManager.updateQuota(
+ Some(ClientQuotaManager.UserEntity("userA")),
+ None,
+ Some(new Quota(8000, true))
+ )
+ clientQuotaManager.updateQuota(
+ Some(ClientQuotaManager.UserEntity("userA")),
+ Some(ClientQuotaManager.ClientIdEntity("client1")),
+ Some(new Quota(10000, true))
+ )
checkQuota(clientQuotaManager, "userA", "client2", 8000, 0, expectThrottle = false)
checkQuota(clientQuotaManager, "userA", "client2", 8000, 4500, expectThrottle = true) // Throttled due to sum of new and earlier values
checkQuota(clientQuotaManager, "userA", "client1", 10000, 0, expectThrottle = false)
checkQuota(clientQuotaManager, "userA", "client1", 10000, 6000, expectThrottle = true)
- clientQuotaManager.updateQuota(Some("userA"), Some("client1"), Some("client1"), None)
+ clientQuotaManager.updateQuota(
+ Some(ClientQuotaManager.UserEntity("userA")),
+ Some(ClientQuotaManager.ClientIdEntity("client1")),
+ None
+ )
checkQuota(clientQuotaManager, "userA", "client6", 8000, 0, expectThrottle = true) // Throttled due to shared user quota
- clientQuotaManager.updateQuota(Some("userA"), Some("client6"), Some("client6"), Some(new Quota(11000, true)))
+ clientQuotaManager.updateQuota(
+ Some(ClientQuotaManager.UserEntity("userA")),
+ Some(ClientQuotaManager.ClientIdEntity("client6")),
+ Some(new Quota(11000, true))
+ )
checkQuota(clientQuotaManager, "userA", "client6", 11000, 8500, expectThrottle = false)
- clientQuotaManager.updateQuota(Some("userA"), Some(ZooKeeperInternals.DEFAULT_STRING), Some(ZooKeeperInternals.DEFAULT_STRING), Some(new Quota(12000, true)))
- clientQuotaManager.updateQuota(Some("userA"), Some("client6"), Some("client6"), None)
+ clientQuotaManager.updateQuota(
+ Some(ClientQuotaManager.UserEntity("userA")),
+ Some(ClientQuotaManager.DefaultClientIdEntity),
+ Some(new Quota(12000, true))
+ )
+ clientQuotaManager.updateQuota(
+ Some(ClientQuotaManager.UserEntity("userA")),
+ Some(ClientQuotaManager.ClientIdEntity("client6")),
+ None
+ )
checkQuota(clientQuotaManager, "userA", "client6", 12000, 4000, expectThrottle = true) // Throttled due to sum of new and earlier values
} finally {
@@ -304,10 +376,13 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest {
val clientQuotaManager = new ClientQuotaManager(config, metrics, QuotaType.PRODUCE, time, "")
val queueSizeMetric = metrics.metrics().get(metrics.metricName("queue-size", "Produce", ""))
try {
- clientQuotaManager.updateQuota(None, Some(ZooKeeperInternals.DEFAULT_STRING), Some(ZooKeeperInternals.DEFAULT_STRING),
- Some(new Quota(500, true)))
+ clientQuotaManager.updateQuota(
+ None,
+ Some(ClientQuotaManager.DefaultClientIdEntity),
+ Some(new Quota(500, true))
+ )
- // We have 10 second windows. Make sure that there is no quota violation
+ // We have 10 seconds windows. Make sure that there is no quota violation
// if we produce under the quota
for (_ <- 0 until 10) {
assertEquals(0, maybeRecord(clientQuotaManager, "ANONYMOUS", "unknown", 400))
@@ -352,8 +427,11 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest {
def testExpireThrottleTimeSensor(): Unit = {
val clientQuotaManager = new ClientQuotaManager(config, metrics, QuotaType.PRODUCE, time, "")
try {
- clientQuotaManager.updateQuota(None, Some(ZooKeeperInternals.DEFAULT_STRING), Some(ZooKeeperInternals.DEFAULT_STRING),
- Some(new Quota(500, true)))
+ clientQuotaManager.updateQuota(
+ None,
+ Some(ClientQuotaManager.DefaultClientIdEntity),
+ Some(new Quota(500, true))
+ )
maybeRecord(clientQuotaManager, "ANONYMOUS", "client1", 100)
// remove the throttle time sensor
@@ -374,8 +452,11 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest {
def testExpireQuotaSensors(): Unit = {
val clientQuotaManager = new ClientQuotaManager(config, metrics, QuotaType.PRODUCE, time, "")
try {
- clientQuotaManager.updateQuota(None, Some(ZooKeeperInternals.DEFAULT_STRING), Some(ZooKeeperInternals.DEFAULT_STRING),
- Some(new Quota(500, true)))
+ clientQuotaManager.updateQuota(
+ None,
+ Some(ClientQuotaManager.DefaultClientIdEntity),
+ Some(new Quota(500, true))
+ )
maybeRecord(clientQuotaManager, "ANONYMOUS", "client1", 100)
// remove all the sensors
@@ -401,8 +482,11 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest {
val clientQuotaManager = new ClientQuotaManager(config, metrics, QuotaType.PRODUCE, time, "")
val clientId = "client@#$%"
try {
- clientQuotaManager.updateQuota(None, Some(ZooKeeperInternals.DEFAULT_STRING), Some(ZooKeeperInternals.DEFAULT_STRING),
- Some(new Quota(500, true)))
+ clientQuotaManager.updateQuota(
+ None,
+ Some(ClientQuotaManager.DefaultClientIdEntity),
+ Some(new Quota(500, true))
+ )
maybeRecord(clientQuotaManager, "ANONYMOUS", clientId, 100)
@@ -417,10 +501,10 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest {
}
}
- private case class UserClient(user: String, clientId: String, configUser: Option[String] = None, configClientId: Option[String] = None) {
- // The class under test expects only sanitized client configs. We pass both the default value (which should not be
- // sanitized to ensure it remains unique) and non-default values, so we need to take care in generating the sanitized
- // client ID
- def sanitizedConfigClientId = configClientId.map(x => if (x == ZooKeeperInternals.DEFAULT_STRING) ZooKeeperInternals.DEFAULT_STRING else Sanitizer.sanitize(x))
- }
+ private case class UserClient(
+ user: String,
+ clientId: String,
+ configUser: Option[BaseUserEntity] = None,
+ configClientEntity: Option[ClientQuotaManager.ClientIdEntity] = None
+ )
}
diff --git a/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala b/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala
index cec9289e691..8c30f749427 100644
--- a/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala
@@ -28,7 +28,7 @@ import org.apache.kafka.common.internals.KafkaFutureImpl
import org.apache.kafka.common.quota.{ClientQuotaAlteration, ClientQuotaEntity, ClientQuotaFilter, ClientQuotaFilterComponent}
import org.apache.kafka.common.requests.{AlterClientQuotasRequest, AlterClientQuotasResponse, DescribeClientQuotasRequest, DescribeClientQuotasResponse}
import org.apache.kafka.common.test.ClusterInstance
-import org.apache.kafka.server.config.{QuotaConfig, ZooKeeperInternals}
+import org.apache.kafka.server.config.QuotaConfig
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.Disabled
@@ -517,20 +517,6 @@ class ClientQuotasRequestTest(cluster: ClusterInstance) {
))
}
- @ClusterTest
- def testClientQuotasWithDefaultName(): Unit = {
- // An entity using the name associated with the default entity name. The entity's name should be sanitized so
- // that it does not conflict with the default entity name.
- val entity = new ClientQuotaEntity(Map(ClientQuotaEntity.CLIENT_ID -> ZooKeeperInternals.DEFAULT_STRING).asJava)
- alterEntityQuotas(entity, Map(QuotaConfig.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG -> Some(20000.0)), validateOnly = false)
- verifyDescribeEntityQuotas(entity, Map(QuotaConfig.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG -> 20000.0))
-
- // This should not match.
- val result = describeClientQuotas(
- ClientQuotaFilter.containsOnly(List(ClientQuotaFilterComponent.ofDefaultEntity(ClientQuotaEntity.CLIENT_ID)).asJava))
- assert(result.isEmpty)
- }
-
private def verifyDescribeEntityQuotas(entity: ClientQuotaEntity, quotas: Map[String, Double]): Unit = {
TestUtils.tryUntilNoAssertionError(waitTime = 5000L) {
val components = entity.entries.asScala.map { case (entityType, entityName) =>
diff --git a/core/src/test/scala/unit/kafka/server/ClientRequestQuotaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ClientRequestQuotaManagerTest.scala
index 2b5471653d9..368280d2354 100644
--- a/core/src/test/scala/unit/kafka/server/ClientRequestQuotaManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ClientRequestQuotaManagerTest.scala
@@ -30,11 +30,15 @@ class ClientRequestQuotaManagerTest extends BaseClientQuotaManagerTest {
@Test
def testRequestPercentageQuotaViolation(): Unit = {
val clientRequestQuotaManager = new ClientRequestQuotaManager(config, metrics, time, "", Optional.empty())
- clientRequestQuotaManager.updateQuota(Some("ANONYMOUS"), Some("test-client"), Some("test-client"), Some(Quota.upperBound(1)))
+ clientRequestQuotaManager.updateQuota(
+ Some(ClientQuotaManager.UserEntity("ANONYMOUS")),
+ Some(ClientQuotaManager.ClientIdEntity("test-client")),
+ Some(Quota.upperBound(1))
+ )
val queueSizeMetric = metrics.metrics().get(metrics.metricName("queue-size", QuotaType.REQUEST.toString, ""))
def millisToPercent(millis: Double) = millis * 1000 * 1000 * ClientRequestQuotaManager.NANOS_TO_PERCENTAGE_PER_SECOND
try {
- // We have 10 second windows. Make sure that there is no quota violation
+ // We have 10 seconds windows. Make sure that there is no quota violation
// if we are under the quota
for (_ <- 0 until 10) {
assertEquals(0, maybeRecord(clientRequestQuotaManager, "ANONYMOUS", "test-client", millisToPercent(4)))
diff --git a/core/src/test/scala/unit/kafka/server/ControllerMutationQuotaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ControllerMutationQuotaManagerTest.scala
index 8fa4a290cc7..2a7cfe35a85 100644
--- a/core/src/test/scala/unit/kafka/server/ControllerMutationQuotaManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ControllerMutationQuotaManagerTest.scala
@@ -145,8 +145,11 @@ class ControllerMutationQuotaManagerTest extends BaseClientQuotaManagerTest {
@Test
def testControllerMutationQuotaViolation(): Unit = {
withQuotaManager { quotaManager =>
- quotaManager.updateQuota(Some(User), Some(ClientId), Some(ClientId),
- Some(Quota.upperBound(10)))
+ quotaManager.updateQuota(
+ Some(User).map(s => ClientQuotaManager.UserEntity(s)),
+ Some(ClientQuotaManager.ClientIdEntity(ClientId)),
+ Some(Quota.upperBound(10))
+ )
val queueSizeMetric = metrics.metrics().get(
metrics.metricName("queue-size", QuotaType.CONTROLLER_MUTATION.toString, ""))
@@ -204,8 +207,11 @@ class ControllerMutationQuotaManagerTest extends BaseClientQuotaManagerTest {
@Test
def testNewStrictQuotaForReturnsStrictQuotaWhenQuotaIsEnabled(): Unit = {
withQuotaManager { quotaManager =>
- quotaManager.updateQuota(Some(User), Some(ClientId), Some(ClientId),
- Some(Quota.upperBound(10)))
+ quotaManager.updateQuota(
+ Some(User).map(s => ClientQuotaManager.UserEntity(s)),
+ Some(ClientQuotaManager.ClientIdEntity(ClientId)),
+ Some(Quota.upperBound(10))
+ )
val quota = quotaManager.newStrictQuotaFor(buildSession(User), ClientId)
assertTrue(quota.isInstanceOf[StrictControllerMutationQuota])
@@ -223,8 +229,11 @@ class ControllerMutationQuotaManagerTest extends BaseClientQuotaManagerTest {
@Test
def testNewPermissiveQuotaForReturnsStrictQuotaWhenQuotaIsEnabled(): Unit = {
withQuotaManager { quotaManager =>
- quotaManager.updateQuota(Some(User), Some(ClientId), Some(ClientId),
- Some(Quota.upperBound(10)))
+ quotaManager.updateQuota(
+ Some(User).map(s => ClientQuotaManager.UserEntity(s)),
+ Some(ClientQuotaManager.ClientIdEntity(ClientId)),
+ Some(Quota.upperBound(10))
+ )
val quota = quotaManager.newPermissiveQuotaFor(buildSession(User), ClientId)
assertTrue(quota.isInstanceOf[PermissiveControllerMutationQuota])
}
diff --git a/server-common/src/main/java/org/apache/kafka/server/config/ZooKeeperInternals.java b/server-common/src/main/java/org/apache/kafka/server/config/ZooKeeperInternals.java
deleted file mode 100644
index 72ac563ae52..00000000000
--- a/server-common/src/main/java/org/apache/kafka/server/config/ZooKeeperInternals.java
+++ /dev/null
@@ -1,27 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.kafka.server.config;
-
-public class ZooKeeperInternals {
- /**
- * This string is used in ZooKeeper in several places to indicate a default entity type.
- * For example, default user quotas are stored under /config/users/<default>
- * Note that AdminClient does <b>not</b> use this to indicate a default, nor do records in KRaft mode.
- * This constant will go away in Apache Kafka 4.0 with the end of ZK mode.
- */
- public static final String DEFAULT_STRING = "<default>";
-}