You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ju...@apache.org on 2016/09/17 17:06:10 UTC
[1/2] kafka git commit: KAFKA-3492;
Secure quotas for authenticated users
Repository: kafka
Updated Branches:
refs/heads/trunk ecc1fb10f -> 69356fbc6
http://git-wip-us.apache.org/repos/asf/kafka/blob/69356fbc/core/src/test/scala/integration/kafka/api/ClientQuotasTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/integration/kafka/api/ClientQuotasTest.scala b/core/src/test/scala/integration/kafka/api/ClientQuotasTest.scala
deleted file mode 100644
index 366507d..0000000
--- a/core/src/test/scala/integration/kafka/api/ClientQuotasTest.scala
+++ /dev/null
@@ -1,206 +0,0 @@
-/**
- * Licensed 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 kafka.api
-
-import java.util.Properties
-
-import kafka.admin.AdminUtils
-import kafka.consumer.SimpleConsumer
-import kafka.integration.KafkaServerTestHarness
-import kafka.server._
-import kafka.utils.TestUtils
-import org.apache.kafka.clients.consumer.{ConsumerConfig, KafkaConsumer}
-import org.apache.kafka.clients.producer._
-import org.apache.kafka.clients.producer.internals.ErrorLoggingCallback
-import org.apache.kafka.common.MetricName
-import org.apache.kafka.common.metrics.{Quota, KafkaMetric}
-import org.apache.kafka.common.protocol.ApiKeys
-import org.junit.Assert.assertEquals
-import org.junit.Assert.assertTrue
-import org.junit.{After, Before, Test}
-import scala.collection.JavaConversions._
-import scala.collection.JavaConverters._
-
-import scala.collection.Map
-import scala.collection.mutable
-
-class ClientQuotasTest extends KafkaServerTestHarness {
- private val producerBufferSize = 300000
- private val producerId1 = "QuotasTestProducer-1"
- private val producerId2 = "QuotasTestProducer-2"
- private val consumerId1 = "QuotasTestConsumer-1"
- private val consumerId2 = "QuotasTestConsumer-2"
-
- val numServers = 2
- val overridingProps = new Properties()
-
- // Low enough quota that a producer sending a small payload in a tight loop should get throttled
- overridingProps.put(KafkaConfig.ProducerQuotaBytesPerSecondDefaultProp, "8000")
- overridingProps.put(KafkaConfig.ConsumerQuotaBytesPerSecondDefaultProp, "2500")
-
- override def generateConfigs() = {
- FixedPortTestUtils.createBrokerConfigs(numServers,
- zkConnect,
- enableControlledShutdown = false)
- .map(KafkaConfig.fromProps(_, overridingProps))
- }
-
- var producers = mutable.Buffer[KafkaProducer[Array[Byte], Array[Byte]]]()
- var consumers = mutable.Buffer[KafkaConsumer[Array[Byte], Array[Byte]]]()
- var replicaConsumers = mutable.Buffer[SimpleConsumer]()
-
- var leaderNode: KafkaServer = null
- var followerNode: KafkaServer = null
- private val topic1 = "topic-1"
-
- @Before
- override def setUp() {
- super.setUp()
- val producerProps = new Properties()
- producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList)
- producerProps.put(ProducerConfig.ACKS_CONFIG, "0")
- producerProps.put(ProducerConfig.BUFFER_MEMORY_CONFIG, producerBufferSize.toString)
- producerProps.put(ProducerConfig.CLIENT_ID_CONFIG, producerId1)
- producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
- classOf[org.apache.kafka.common.serialization.ByteArraySerializer])
- producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
- classOf[org.apache.kafka.common.serialization.ByteArraySerializer])
- producers += new KafkaProducer[Array[Byte], Array[Byte]](producerProps)
-
- producerProps.put(ProducerConfig.CLIENT_ID_CONFIG, producerId2)
- producers += new KafkaProducer[Array[Byte], Array[Byte]](producerProps)
-
- val numPartitions = 1
- val leaders = TestUtils.createTopic(zkUtils, topic1, numPartitions, numServers, servers)
- leaderNode = if (leaders(0).get == servers.head.config.brokerId) servers.head else servers(1)
- followerNode = if (leaders(0).get != servers.head.config.brokerId) servers.head else servers(1)
- assertTrue("Leader of all partitions of the topic should exist", leaders.values.forall(leader => leader.isDefined))
-
- // Create consumers
- val consumerProps = new Properties
- consumerProps.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "QuotasTest")
- consumerProps.setProperty(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 4096.toString)
- consumerProps.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
- consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList)
- consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
- classOf[org.apache.kafka.common.serialization.ByteArrayDeserializer])
- consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
- classOf[org.apache.kafka.common.serialization.ByteArrayDeserializer])
-
- consumerProps.put(ConsumerConfig.CLIENT_ID_CONFIG, consumerId1)
- consumers += new KafkaConsumer(consumerProps)
- // Create replica consumers with the same clientId as the high level consumer. These requests should never be throttled
- replicaConsumers += new SimpleConsumer("localhost", leaderNode.boundPort(), 1000000, 64*1024, consumerId1)
-
- consumerProps.put(ConsumerConfig.CLIENT_ID_CONFIG, consumerId2)
- consumers += new KafkaConsumer(consumerProps)
- replicaConsumers += new SimpleConsumer("localhost", leaderNode.boundPort(), 1000000, 64*1024, consumerId2)
- }
-
- @After
- override def tearDown() {
- producers.foreach( _.close )
- consumers.foreach( _.close )
- replicaConsumers.foreach( _.close )
- super.tearDown()
- }
-
- @Test
- def testThrottledProducerConsumer() {
- val allMetrics: mutable.Map[MetricName, KafkaMetric] = leaderNode.metrics.metrics().asScala
-
- val numRecords = 1000
- produce(producers.head, numRecords)
-
- val producerMetricName = leaderNode.metrics.metricName("throttle-time",
- ApiKeys.PRODUCE.name,
- "Tracking throttle-time per client",
- "client-id", producerId1)
- assertTrue("Should have been throttled", allMetrics(producerMetricName).value() > 0)
-
- // Consumer should read in a bursty manner and get throttled immediately
- consume(consumers.head, numRecords)
- // The replica consumer should not be throttled also. Create a fetch request which will exceed the quota immediately
- val request = new FetchRequestBuilder().addFetch(topic1, 0, 0, 1024*1024).replicaId(followerNode.config.brokerId).build()
- replicaConsumers.head.fetch(request)
- val consumerMetricName = leaderNode.metrics.metricName("throttle-time",
- ApiKeys.FETCH.name,
- "Tracking throttle-time per client",
- "client-id", consumerId1)
- assertTrue("Should have been throttled", allMetrics(consumerMetricName).value() > 0)
- }
-
- @Test
- def testProducerConsumerOverrideUnthrottled() {
- // Give effectively unlimited quota for producerId2 and consumerId2
- val props = new Properties()
- props.put(ClientConfigOverride.ProducerOverride, Long.MaxValue.toString)
- props.put(ClientConfigOverride.ConsumerOverride, Long.MaxValue.toString)
-
- AdminUtils.changeClientIdConfig(zkUtils, producerId2, props)
- AdminUtils.changeClientIdConfig(zkUtils, consumerId2, props)
-
- TestUtils.retry(10000) {
- val overrideProducerQuota = leaderNode.apis.quotas.produce.quota(producerId2)
- val overrideConsumerQuota = leaderNode.apis.quotas.fetch.quota(consumerId2)
-
- assertEquals(s"ClientId $producerId2 must have unlimited producer quota", Quota.upperBound(Long.MaxValue), overrideProducerQuota)
- assertEquals(s"ClientId $consumerId2 must have unlimited consumer quota", Quota.upperBound(Long.MaxValue), overrideConsumerQuota)
- }
-
-
- val allMetrics: mutable.Map[MetricName, KafkaMetric] = leaderNode.metrics.metrics().asScala
- val numRecords = 1000
- produce(producers(1), numRecords)
- val producerMetricName = leaderNode.metrics.metricName("throttle-time",
- ApiKeys.PRODUCE.name,
- "Tracking throttle-time per client",
- "client-id", producerId2)
- assertEquals("Should not have been throttled", 0.0, allMetrics(producerMetricName).value(), 0.0)
-
- // The "client" consumer does not get throttled.
- consume(consumers(1), numRecords)
- // The replica consumer should not be throttled also. Create a fetch request which will exceed the quota immediately
- val request = new FetchRequestBuilder().addFetch(topic1, 0, 0, 1024*1024).replicaId(followerNode.config.brokerId).build()
- replicaConsumers(1).fetch(request)
- val consumerMetricName = leaderNode.metrics.metricName("throttle-time",
- ApiKeys.FETCH.name,
- "Tracking throttle-time per client",
- "client-id", consumerId2)
- assertEquals("Should not have been throttled", 0.0, allMetrics(consumerMetricName).value(), 0.0)
- }
-
- def produce(p: KafkaProducer[Array[Byte], Array[Byte]], count: Int): Int = {
- var numBytesProduced = 0
- for (i <- 0 to count) {
- val payload = i.toString.getBytes
- numBytesProduced += payload.length
- p.send(new ProducerRecord[Array[Byte], Array[Byte]](topic1, null, null, payload),
- new ErrorLoggingCallback(topic1, null, null, true)).get()
- Thread.sleep(1)
- }
- numBytesProduced
- }
-
- def consume(consumer: KafkaConsumer[Array[Byte], Array[Byte]], numRecords: Int) {
- consumer.subscribe(List(topic1))
- var numConsumed = 0
- while (numConsumed < numRecords) {
- for (cr <- consumer.poll(100)) {
- numConsumed += 1
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/kafka/blob/69356fbc/core/src/test/scala/integration/kafka/api/UserClientIdQuotaTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/integration/kafka/api/UserClientIdQuotaTest.scala b/core/src/test/scala/integration/kafka/api/UserClientIdQuotaTest.scala
new file mode 100644
index 0000000..081fa0b
--- /dev/null
+++ b/core/src/test/scala/integration/kafka/api/UserClientIdQuotaTest.scala
@@ -0,0 +1,66 @@
+/**
+ * Licensed 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 kafka.api
+
+import java.io.File
+import java.util.Properties
+
+import kafka.admin.AdminUtils
+
+import kafka.server.{KafkaConfig, ConfigEntityName, QuotaConfigOverride, QuotaId}
+
+import org.apache.kafka.common.protocol.SecurityProtocol
+import org.junit.Before
+
+class UserClientIdQuotaTest extends BaseQuotaTest {
+
+ override protected def securityProtocol = SecurityProtocol.SSL
+ override protected lazy val trustStoreFile = Some(File.createTempFile("truststore", ".jks"))
+
+ override val userPrincipal = "O=A client,CN=localhost"
+ override def producerQuotaId = QuotaId(Some(QuotaId.sanitize(userPrincipal)), Some(producerClientId))
+ override def consumerQuotaId = QuotaId(Some(QuotaId.sanitize(userPrincipal)), Some(consumerClientId))
+
+ @Before
+ override def setUp() {
+ this.serverConfig.setProperty(KafkaConfig.SslClientAuthProp, "required")
+ this.serverConfig.setProperty(KafkaConfig.ProducerQuotaBytesPerSecondDefaultProp, Long.MaxValue.toString)
+ this.serverConfig.setProperty(KafkaConfig.ConsumerQuotaBytesPerSecondDefaultProp, Long.MaxValue.toString)
+ super.setUp()
+ val defaultProps = quotaProperties(defaultProducerQuota, defaultConsumerQuota)
+ AdminUtils.changeUserOrUserClientIdConfig(zkUtils, ConfigEntityName.Default + "/clients/" + ConfigEntityName.Default, defaultProps)
+ waitForQuotaUpdate(defaultProducerQuota, defaultConsumerQuota)
+ }
+
+ override def overrideQuotas(producerQuota: Long, consumerQuota: Long) {
+ val producerProps = new Properties()
+ producerProps.setProperty(QuotaConfigOverride.ProducerOverride, producerQuota.toString)
+ updateQuotaOverride(userPrincipal, producerClientId, producerProps)
+
+ val consumerProps = new Properties()
+ consumerProps.setProperty(QuotaConfigOverride.ConsumerOverride, consumerQuota.toString)
+ updateQuotaOverride(userPrincipal, consumerClientId, consumerProps)
+ }
+
+ override def removeQuotaOverrides() {
+ val emptyProps = new Properties
+ AdminUtils.changeUserOrUserClientIdConfig(zkUtils, QuotaId.sanitize(userPrincipal) + "/clients/" + producerClientId, emptyProps)
+ AdminUtils.changeUserOrUserClientIdConfig(zkUtils, QuotaId.sanitize(userPrincipal) + "/clients/" + consumerClientId, emptyProps)
+ }
+
+ private def updateQuotaOverride(userPrincipal: String, clientId: String, properties: Properties) {
+ AdminUtils.changeUserOrUserClientIdConfig(zkUtils, QuotaId.sanitize(userPrincipal) + "/clients/" + clientId, properties)
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/69356fbc/core/src/test/scala/integration/kafka/api/UserQuotaTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/integration/kafka/api/UserQuotaTest.scala b/core/src/test/scala/integration/kafka/api/UserQuotaTest.scala
new file mode 100644
index 0000000..3d5d702
--- /dev/null
+++ b/core/src/test/scala/integration/kafka/api/UserQuotaTest.scala
@@ -0,0 +1,61 @@
+/**
+ * Licensed 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 kafka.api
+
+import java.io.File
+import java.util.Properties
+
+import kafka.admin.AdminUtils
+import kafka.server.{KafkaConfig, ConfigEntityName, QuotaId}
+
+import org.apache.kafka.common.protocol.SecurityProtocol
+import org.junit.Before
+
+class UserQuotaTest extends BaseQuotaTest with SaslTestHarness {
+
+ override protected def securityProtocol = SecurityProtocol.SASL_SSL
+ override protected lazy val trustStoreFile = Some(File.createTempFile("truststore", ".jks"))
+ override protected val zkSaslEnabled = false
+ override protected val saslProperties = Some(kafkaSaslProperties(kafkaClientSaslMechanism, Some(kafkaServerSaslMechanisms)))
+
+ override val userPrincipal = "client"
+ override val producerQuotaId = QuotaId(Some(userPrincipal), None)
+ override val consumerQuotaId = QuotaId(Some(userPrincipal), None)
+
+ @Before
+ override def setUp() {
+ this.serverConfig.setProperty(KafkaConfig.ProducerQuotaBytesPerSecondDefaultProp, Long.MaxValue.toString)
+ this.serverConfig.setProperty(KafkaConfig.ConsumerQuotaBytesPerSecondDefaultProp, Long.MaxValue.toString)
+ super.setUp()
+ val defaultProps = quotaProperties(defaultProducerQuota, defaultConsumerQuota)
+ AdminUtils.changeUserOrUserClientIdConfig(zkUtils, ConfigEntityName.Default, defaultProps)
+ waitForQuotaUpdate(defaultProducerQuota, defaultConsumerQuota)
+ }
+
+ override def overrideQuotas(producerQuota: Long, consumerQuota: Long) {
+ val props = quotaProperties(producerQuota, consumerQuota)
+ updateQuotaOverride(props)
+ }
+
+ override def removeQuotaOverrides() {
+ val emptyProps = new Properties
+ updateQuotaOverride(emptyProps)
+ updateQuotaOverride(emptyProps)
+ }
+
+ private def updateQuotaOverride(properties: Properties) {
+ AdminUtils.changeUserOrUserClientIdConfig(zkUtils, QuotaId.sanitize(userPrincipal), properties)
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/69356fbc/core/src/test/scala/unit/kafka/admin/AdminTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/admin/AdminTest.scala b/core/src/test/scala/unit/kafka/admin/AdminTest.scala
index 5dca17f..892e26b 100755
--- a/core/src/test/scala/unit/kafka/admin/AdminTest.scala
+++ b/core/src/test/scala/unit/kafka/admin/AdminTest.scala
@@ -513,8 +513,8 @@ class AdminTest extends ZooKeeperTestHarness with Logging with RackAwareTest {
// Test that the existing clientId overrides are read
val server = TestUtils.createServer(KafkaConfig.fromProps(TestUtils.createBrokerConfig(0, zkConnect)))
try {
- assertEquals(new Quota(1000, true), server.apis.quotas.produce.quota(clientId))
- assertEquals(new Quota(2000, true), server.apis.quotas.fetch.quota(clientId))
+ assertEquals(new Quota(1000, true), server.apis.quotas.produce.quota("ANONYMOUS", clientId))
+ assertEquals(new Quota(2000, true), server.apis.quotas.fetch.quota("ANONYMOUS", clientId))
} finally {
server.shutdown()
CoreUtils.delete(server.config.logDirs)
http://git-wip-us.apache.org/repos/asf/kafka/blob/69356fbc/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
index be798ff..ddfbb51 100644
--- a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
+++ b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
@@ -19,10 +19,14 @@ package kafka.admin
import java.util.Properties
import kafka.admin.ConfigCommand.ConfigCommandOptions
+import kafka.common.InvalidConfigException
+import kafka.server.{ConfigEntityName, QuotaId}
+import kafka.utils.{Logging, ZkUtils}
+import kafka.zk.ZooKeeperTestHarness
+
+import org.easymock.EasyMock
import org.junit.Assert._
import org.junit.Test
-import kafka.utils.{ZkUtils, Logging}
-import kafka.zk.ZooKeeperTestHarness
class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
@Test
@@ -217,4 +221,182 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
}
ConfigCommand.alterConfig(null, createOpts, configChange)
}
+
+ @Test
+ def testQuotaConfigEntity() {
+
+ def createOpts(entityType: String, entityName: Option[String], otherArgs: Array[String]) : ConfigCommandOptions = {
+ val optArray = Array("--zookeeper", zkConnect,
+ "--entity-type", entityType)
+ val nameArray = entityName match {
+ case Some(name) => Array("--entity-name", name)
+ case None => Array[String]()
+ }
+ new ConfigCommandOptions(optArray ++ nameArray ++ otherArgs)
+ }
+
+ def checkEntity(entityType: String, entityName: Option[String], expectedEntityName: String, otherArgs: Array[String]) {
+ val opts = createOpts(entityType, entityName, otherArgs)
+ opts.checkArgs()
+ val entity = ConfigCommand.parseEntity(opts)
+ assertEquals(entityType, entity.root.entityType)
+ assertEquals(expectedEntityName, entity.fullSanitizedName)
+ }
+
+ def checkInvalidEntity(entityType: String, entityName: Option[String], otherArgs: Array[String]) {
+ val opts = createOpts(entityType, entityName, otherArgs)
+ try {
+ opts.checkArgs()
+ ConfigCommand.parseEntity(opts)
+ fail("Did not fail with invalid argument list")
+ } catch {
+ case e: IllegalArgumentException => // expected exception
+ }
+ }
+
+ val describeOpts = Array("--describe")
+ val alterOpts = Array("--alter", "--add-config", "a=b,c=d")
+
+ // <client-id> quota
+ val clientId = "client-1"
+ for (opts <- Seq(describeOpts, alterOpts)) {
+ checkEntity("clients", Some(clientId), clientId, opts)
+ checkEntity("clients", Some(""), ConfigEntityName.Default, opts)
+ }
+ checkEntity("clients", None, "", describeOpts)
+ checkInvalidEntity("clients", None, alterOpts)
+
+ // <user> quota
+ val principal = "CN=ConfigCommandTest,O=Apache,L=<default>"
+ val sanitizedPrincipal = QuotaId.sanitize(principal)
+ assertEquals(-1, sanitizedPrincipal.indexOf('='))
+ assertEquals(principal, QuotaId.desanitize(sanitizedPrincipal))
+ for (opts <- Seq(describeOpts, alterOpts)) {
+ checkEntity("users", Some(principal), sanitizedPrincipal, opts)
+ checkEntity("users", Some(""), ConfigEntityName.Default, opts)
+ }
+ checkEntity("users", None, "", describeOpts)
+ checkInvalidEntity("users", None, alterOpts)
+
+ // <user, client-id> quota
+ val userClient = sanitizedPrincipal + "/clients/" + clientId
+ def clientIdOpts(name: String) = Array("--entity-type", "clients", "--entity-name", name)
+ for (opts <- Seq(describeOpts, alterOpts)) {
+ checkEntity("users", Some(principal), userClient, opts ++ clientIdOpts(clientId))
+ checkEntity("users", Some(principal), sanitizedPrincipal + "/clients/" + ConfigEntityName.Default, opts ++ clientIdOpts(""))
+ checkEntity("users", Some(""), ConfigEntityName.Default + "/clients/" + clientId, describeOpts ++ clientIdOpts(clientId))
+ checkEntity("users", Some(""), ConfigEntityName.Default + "/clients/" + ConfigEntityName.Default, opts ++ clientIdOpts(""))
+ }
+ checkEntity("users", Some(principal), sanitizedPrincipal + "/clients", describeOpts ++ Array("--entity-type", "clients"))
+ // Both user and client-id must be provided for alter
+ checkInvalidEntity("users", Some(principal), alterOpts ++ Array("--entity-type", "clients"))
+ checkInvalidEntity("users", None, alterOpts ++ clientIdOpts(clientId))
+ checkInvalidEntity("users", None, alterOpts ++ Array("--entity-type", "clients"))
+ }
+
+ @Test
+ def testUserClientQuotaOpts() {
+ def checkEntity(expectedEntityType: String, expectedEntityName: String, args: String*) {
+ val opts = new ConfigCommandOptions(Array("--zookeeper", zkConnect) ++ args)
+ opts.checkArgs()
+ val entity = ConfigCommand.parseEntity(opts)
+ assertEquals(expectedEntityType, entity.root.entityType)
+ assertEquals(expectedEntityName, entity.fullSanitizedName)
+ }
+
+ // <default> is a valid user principal (can be handled with URL-encoding),
+ // but an invalid client-id (cannot be handled since client-ids are not encoded)
+ checkEntity("users", QuotaId.sanitize("<default>"),
+ "--entity-type", "users", "--entity-name", "<default>",
+ "--alter", "--add-config", "a=b,c=d")
+ try {
+ checkEntity("clients", QuotaId.sanitize("<default>"),
+ "--entity-type", "clients", "--entity-name", "<default>",
+ "--alter", "--add-config", "a=b,c=d")
+ fail("Did not fail with invalid client-id")
+ } catch {
+ case e: InvalidConfigException => // expected
+ }
+
+ checkEntity("users", QuotaId.sanitize("CN=user1") + "/clients/client1",
+ "--entity-type", "users", "--entity-name", "CN=user1", "--entity-type", "clients", "--entity-name", "client1",
+ "--alter", "--add-config", "a=b,c=d")
+ checkEntity("users", QuotaId.sanitize("CN=user1") + "/clients/client1",
+ "--entity-name", "CN=user1", "--entity-type", "users", "--entity-name", "client1", "--entity-type", "clients",
+ "--alter", "--add-config", "a=b,c=d")
+ checkEntity("users", QuotaId.sanitize("CN=user1") + "/clients/client1",
+ "--entity-type", "clients", "--entity-name", "client1", "--entity-type", "users", "--entity-name", "CN=user1",
+ "--alter", "--add-config", "a=b,c=d")
+ checkEntity("users", QuotaId.sanitize("CN=user1") + "/clients/client1",
+ "--entity-name", "client1", "--entity-type", "clients", "--entity-name", "CN=user1", "--entity-type", "users",
+ "--alter", "--add-config", "a=b,c=d")
+ checkEntity("users", QuotaId.sanitize("CN=user1") + "/clients",
+ "--entity-type", "clients", "--entity-name", "CN=user1", "--entity-type", "users",
+ "--describe")
+ checkEntity("users", "/clients",
+ "--entity-type", "clients", "--entity-type", "users",
+ "--describe")
+ }
+
+ @Test
+ def testQuotaDescribeEntities() {
+ val zkUtils = EasyMock.createNiceMock(classOf[ZkUtils])
+
+ def checkEntities(opts: Array[String], expectedFetches: Map[String, Seq[String]], expectedEntityNames: Seq[String]) {
+ val entity = ConfigCommand.parseEntity(new ConfigCommandOptions(opts :+ "--describe"))
+ expectedFetches.foreach {
+ case (name, values) => EasyMock.expect(zkUtils.getAllEntitiesWithConfig(name)).andReturn(values)
+ }
+ EasyMock.replay(zkUtils)
+ val entities = entity.getAllEntities(zkUtils)
+ assertEquals(expectedEntityNames, entities.map(e => e.fullSanitizedName))
+ EasyMock.reset(zkUtils)
+ }
+
+ val clientId = "a-client"
+ val principal = "CN=ConfigCommandTest.testQuotaDescribeEntities , O=Apache, L=<default>"
+ val sanitizedPrincipal = QuotaId.sanitize(principal)
+ val userClient = sanitizedPrincipal + "/clients/" + clientId
+
+ var opts = Array("--entity-type", "clients", "--entity-name", clientId)
+ checkEntities(opts, Map.empty, Seq(clientId))
+
+ opts = Array("--entity-type", "clients", "--entity-default")
+ checkEntities(opts, Map.empty, Seq("<default>"))
+
+ opts = Array("--entity-type", "clients")
+ checkEntities(opts, Map("clients" -> Seq(clientId)), Seq(clientId))
+
+ opts = Array("--entity-type", "users", "--entity-name", principal)
+ checkEntities(opts, Map.empty, Seq(sanitizedPrincipal))
+
+ opts = Array("--entity-type", "users", "--entity-default")
+ checkEntities(opts, Map.empty, Seq("<default>"))
+
+ opts = Array("--entity-type", "users")
+ checkEntities(opts, Map("users" -> Seq("<default>", sanitizedPrincipal)), Seq("<default>", sanitizedPrincipal))
+
+ opts = Array("--entity-type", "users", "--entity-name", principal, "--entity-type", "clients", "--entity-name", clientId)
+ checkEntities(opts, Map.empty, Seq(userClient))
+
+ opts = Array("--entity-type", "users", "--entity-name", principal, "--entity-type", "clients", "--entity-default")
+ checkEntities(opts, Map.empty, Seq(sanitizedPrincipal + "/clients/<default>"))
+
+ opts = Array("--entity-type", "users", "--entity-name", principal, "--entity-type", "clients")
+ checkEntities(opts,
+ Map("users/" + sanitizedPrincipal + "/clients" -> Seq("client-4")),
+ Seq(sanitizedPrincipal + "/clients/client-4"))
+
+ opts = Array("--entity-type", "users", "--entity-default", "--entity-type", "clients")
+ checkEntities(opts,
+ Map("users/<default>/clients" -> Seq("client-5")),
+ Seq("<default>/clients/client-5"))
+
+ opts = Array("--entity-type", "users", "--entity-type", "clients")
+ val userMap = Map("users/" + sanitizedPrincipal + "/clients" -> Seq("client-2"))
+ val defaultUserMap = Map("users/<default>/clients" -> Seq("client-3"))
+ checkEntities(opts,
+ Map("users" -> Seq("<default>", sanitizedPrincipal)) ++ defaultUserMap ++ userMap,
+ Seq("<default>/clients/client-3", sanitizedPrincipal + "/clients/client-2"))
+ }
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/69356fbc/core/src/test/scala/unit/kafka/admin/TestAdminUtils.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/admin/TestAdminUtils.scala b/core/src/test/scala/unit/kafka/admin/TestAdminUtils.scala
index c7e101f..55fe587 100644
--- a/core/src/test/scala/unit/kafka/admin/TestAdminUtils.scala
+++ b/core/src/test/scala/unit/kafka/admin/TestAdminUtils.scala
@@ -23,5 +23,6 @@ class TestAdminUtils extends AdminUtilities {
override def changeBrokerConfig(zkUtils: ZkUtils, brokerIds: Seq[Int], configs: Properties): Unit = {}
override def fetchEntityConfig(zkUtils: ZkUtils, entityType: String, entityName: String): Properties = {new Properties}
override def changeClientIdConfig(zkUtils: ZkUtils, clientId: String, configs: Properties): Unit = {}
+ override def changeUserOrUserClientIdConfig(zkUtils: ZkUtils, sanitizedEntityName: String, configs: Properties): Unit = {}
override def changeTopicConfig(zkUtils: ZkUtils, topic: String, configs: Properties): Unit = {}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/69356fbc/core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala
index b14464f..291e822 100644
--- a/core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala
@@ -22,6 +22,7 @@ import org.apache.kafka.common.metrics.{MetricConfig, Metrics, Quota}
import org.apache.kafka.common.utils.MockTime
import org.junit.Assert.{assertEquals, assertTrue}
import org.junit.{Before, Test}
+import kafka.admin.ConfigCommand
class ClientQuotaManagerTest {
private val time = new MockTime
@@ -38,42 +39,187 @@ class ClientQuotaManagerTest {
numCallbacks = 0
}
- @Test
- def testQuotaParsing() {
+ private def testQuotaParsing(config: ClientQuotaManagerConfig, client1: UserClient, client2: UserClient, randomClient: UserClient, defaultConfigClient: UserClient) {
val clientMetrics = new ClientQuotaManager(config, newMetrics, QuotaType.Produce, time)
- // Case 1: Update the quota. Assert that the new quota value is returned
- clientMetrics.updateQuota("p1", new Quota(2000, true))
- clientMetrics.updateQuota("p2", new Quota(4000, true))
-
try {
- assertEquals("Default producer quota should be 500", new Quota(500, true), clientMetrics.quota("random-client-id"))
- assertEquals("Should return the overridden value (2000)", new Quota(2000, true), clientMetrics.quota("p1"))
- assertEquals("Should return the overridden value (4000)", new Quota(4000, true), clientMetrics.quota("p2"))
+ // Case 1: Update the quota. Assert that the new quota value is returned
+ clientMetrics.updateQuota(client1.configUser, client1.configClientId, Some(new Quota(2000, true)))
+ clientMetrics.updateQuota(client2.configUser, client2.configClientId, Some(new Quota(4000, true)))
+
+ assertEquals("Default producer quota should be " + config.quotaBytesPerSecondDefault, new Quota(config.quotaBytesPerSecondDefault, true), clientMetrics.quota(randomClient.user, randomClient.clientId))
+ assertEquals("Should return the overridden value (2000)", new Quota(2000, true), clientMetrics.quota(client1.user, client1.clientId))
+ assertEquals("Should return the overridden value (4000)", new Quota(4000, true), clientMetrics.quota(client2.user, client2.clientId))
// p1 should be throttled using the overridden quota
- var throttleTimeMs = clientMetrics.recordAndMaybeThrottle("p1", 2500 * config.numQuotaSamples, this.callback)
+ var throttleTimeMs = clientMetrics.recordAndMaybeThrottle(client1.user, client1.clientId, 2500 * config.numQuotaSamples, this.callback)
assertTrue(s"throttleTimeMs should be > 0. was $throttleTimeMs", throttleTimeMs > 0)
// 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
- clientMetrics.updateQuota("p1", new Quota(3000, true))
- assertEquals("Should return the newly overridden value (3000)", new Quota(3000, true), clientMetrics.quota("p1"))
+ clientMetrics.updateQuota(client1.configUser, client1.configClientId, Some(new Quota(3000, true)))
+ assertEquals("Should return the newly overridden value (3000)", new Quota(3000, true), clientMetrics.quota(client1.user, client1.clientId))
- throttleTimeMs = clientMetrics.recordAndMaybeThrottle("p1", 0, this.callback)
+ throttleTimeMs = clientMetrics.recordAndMaybeThrottle(client1.user, client1.clientId, 0, this.callback)
assertEquals(s"throttleTimeMs should be 0. was $throttleTimeMs", 0, throttleTimeMs)
// Case 3: Change quota back to default. Should be throttled again
- clientMetrics.updateQuota("p1", new Quota(500, true))
- assertEquals("Should return the default value (500)", new Quota(500, true), clientMetrics.quota("p1"))
+ clientMetrics.updateQuota(client1.configUser, client1.configClientId, Some(new Quota(500, true)))
+ assertEquals("Should return the default value (500)", new Quota(500, true), clientMetrics.quota(client1.user, client1.clientId))
- throttleTimeMs = clientMetrics.recordAndMaybeThrottle("p1", 0, this.callback)
+ throttleTimeMs = clientMetrics.recordAndMaybeThrottle(client1.user, client1.clientId, 0, this.callback)
assertTrue(s"throttleTimeMs should be > 0. was $throttleTimeMs", throttleTimeMs > 0)
+
+ // Case 4: Set high default quota, remove p1 quota. p1 should no longer be throttled
+ clientMetrics.updateQuota(client1.configUser, client1.configClientId, None)
+ clientMetrics.updateQuota(defaultConfigClient.configUser, defaultConfigClient.configClientId, Some(new Quota(4000, true)))
+ assertEquals("Should return the newly overridden value (4000)", new Quota(4000, true), clientMetrics.quota(client1.user, client1.clientId))
+
+ throttleTimeMs = clientMetrics.recordAndMaybeThrottle(client1.user, client1.clientId, 1000 * config.numQuotaSamples, this.callback)
+ assertEquals(s"throttleTimeMs should be 0. was $throttleTimeMs", 0, throttleTimeMs)
+
} finally {
clientMetrics.shutdown()
}
}
+ /**
+ * 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() {
+ 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(ConfigEntityName.Default))
+ 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() {
+ 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(ConfigEntityName.Default), None)
+ val config = ClientQuotaManagerConfig(quotaBytesPerSecondDefault = Long.MaxValue)
+ 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() {
+ 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(ConfigEntityName.Default), Some(ConfigEntityName.Default))
+ val config = ClientQuotaManagerConfig(quotaBytesPerSecondDefault = Long.MaxValue)
+ testQuotaParsing(config, client1, client2, randomClient, defaultConfigClient)
+ }
+
+ /**
+ * Tests parsing for <user> quotas when client-id default quota properties are set.
+ */
+ @Test
+ def testUserQuotaParsingWithDefaultClientIdQuota() {
+ 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(ConfigEntityName.Default), 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() {
+ 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(ConfigEntityName.Default), Some(ConfigEntityName.Default))
+ testQuotaParsing(config, client1, client2, randomClient, defaultConfigClient)
+ }
+
+ @Test
+ def testQuotaConfigPrecedence() {
+ val quotaManager = new ClientQuotaManager(ClientQuotaManagerConfig(quotaBytesPerSecondDefault=Long.MaxValue), newMetrics, QuotaType.Produce, time)
+
+ def checkQuota(user: String, clientId: String, expectedBound: Int, value: Int, expectThrottle: Boolean) {
+ assertEquals(new Quota(expectedBound, true), quotaManager.quota(user, clientId))
+ val throttleTimeMs = quotaManager.recordAndMaybeThrottle(user, clientId, value * config.numQuotaSamples, this.callback)
+ if (expectThrottle)
+ assertTrue(s"throttleTimeMs should be > 0. was $throttleTimeMs", throttleTimeMs > 0)
+ else
+ assertEquals(s"throttleTimeMs should be 0. was $throttleTimeMs", 0, throttleTimeMs)
+ }
+
+ try {
+ quotaManager.updateQuota(Some(ConfigEntityName.Default), None, Some(new Quota(1000, true)))
+ quotaManager.updateQuota(None, Some(ConfigEntityName.Default), Some(new Quota(2000, true)))
+ quotaManager.updateQuota(Some(ConfigEntityName.Default), Some(ConfigEntityName.Default), Some(new Quota(3000, true)))
+ quotaManager.updateQuota(Some("userA"), None, Some(new Quota(4000, true)))
+ quotaManager.updateQuota(Some("userA"), Some("client1"), Some(new Quota(5000, true)))
+ quotaManager.updateQuota(Some("userB"), None, Some(new Quota(6000, true)))
+ quotaManager.updateQuota(Some("userB"), Some("client1"), Some(new Quota(7000, true)))
+ quotaManager.updateQuota(Some("userB"), Some(ConfigEntityName.Default), Some(new Quota(8000, true)))
+ quotaManager.updateQuota(Some("userC"), None, Some(new Quota(10000, true)))
+ quotaManager.updateQuota(None, Some("client1"), Some(new Quota(9000, true)))
+
+ checkQuota("userA", "client1", 5000, 4500, false) // <user, client> quota takes precedence over <user>
+ checkQuota("userA", "client2", 4000, 4500, true) // <user> quota takes precedence over <client> and defaults
+ checkQuota("userA", "client3", 4000, 0, true) // <user> quota is shared across clients of user
+ checkQuota("userA", "client1", 5000, 0, false) // <user, client> is exclusive use, unaffected by other clients
+
+ checkQuota("userB", "client1", 7000, 8000, true)
+ checkQuota("userB", "client2", 8000, 7000, false) // Default per-client quota for exclusive use of <user, client>
+ checkQuota("userB", "client3", 8000, 7000, false)
+
+ checkQuota("userD", "client1", 3000, 3500, true) // Default <user, client> quota
+ checkQuota("userD", "client2", 3000, 2500, false)
+ checkQuota("userE", "client1", 3000, 2500, false)
+
+ // Remove default <user, client> quota config, revert to <user> default
+ quotaManager.updateQuota(Some(ConfigEntityName.Default), Some(ConfigEntityName.Default), None)
+ checkQuota("userD", "client1", 1000, 0, false) // Metrics tags changed, restart counter
+ checkQuota("userE", "client4", 1000, 1500, true)
+ checkQuota("userF", "client4", 1000, 800, false) // Default <user> quota shared across clients of user
+ checkQuota("userF", "client5", 1000, 800, true)
+
+ // Remove default <user> quota config, revert to <client-id> default
+ quotaManager.updateQuota(Some(ConfigEntityName.Default), None, None)
+ checkQuota("userF", "client4", 2000, 0, false) // Default <client-id> quota shared across client-id of all users
+ checkQuota("userF", "client5", 2000, 0, false)
+ checkQuota("userF", "client5", 2000, 2500, true)
+ checkQuota("userG", "client5", 2000, 0, true)
+
+ // Update quotas
+ quotaManager.updateQuota(Some("userA"), None, Some(new Quota(8000, true)))
+ quotaManager.updateQuota(Some("userA"), Some("client1"), Some(new Quota(10000, true)))
+ checkQuota("userA", "client2", 8000, 0, false)
+ checkQuota("userA", "client2", 8000, 4500, true) // Throttled due to sum of new and earlier values
+ checkQuota("userA", "client1", 10000, 0, false)
+ checkQuota("userA", "client1", 10000, 6000, true)
+ quotaManager.updateQuota(Some("userA"), Some("client1"), None)
+ checkQuota("userA", "client6", 8000, 0, true) // Throttled due to shared user quota
+ quotaManager.updateQuota(Some("userA"), Some("client6"), Some(new Quota(11000, true)))
+ checkQuota("userA", "client6", 11000, 8500, false)
+ quotaManager.updateQuota(Some("userA"), Some(ConfigEntityName.Default), Some(new Quota(12000, true)))
+ quotaManager.updateQuota(Some("userA"), Some("client6"), None)
+ checkQuota("userA", "client6", 12000, 4000, true) // Throttled due to sum of new and earlier values
+
+ } finally {
+ quotaManager.shutdown()
+ }
+ }
+
@Test
def testQuotaViolation() {
val metrics = newMetrics
@@ -84,7 +230,7 @@ class ClientQuotaManagerTest {
* if we produce under the quota
*/
for (i <- 0 until 10) {
- clientMetrics.recordAndMaybeThrottle("unknown", 400, callback)
+ clientMetrics.recordAndMaybeThrottle("ANONYMOUS", "unknown", 400, callback)
time.sleep(1000)
}
assertEquals(10, numCallbacks)
@@ -95,7 +241,7 @@ class ClientQuotaManagerTest {
// (600 - quota)/quota*window-size = (600-500)/500*10.5 seconds = 2100
// 10.5 seconds because the last window is half complete
time.sleep(500)
- val sleepTime = clientMetrics.recordAndMaybeThrottle("unknown", 2300, callback)
+ val sleepTime = clientMetrics.recordAndMaybeThrottle("ANONYMOUS", "unknown", 2300, callback)
assertEquals("Should be throttled", 2100, sleepTime)
assertEquals(1, queueSizeMetric.value().toInt)
@@ -111,12 +257,12 @@ class ClientQuotaManagerTest {
// Could continue to see delays until the bursty sample disappears
for (i <- 0 until 10) {
- clientMetrics.recordAndMaybeThrottle("unknown", 400, callback)
+ clientMetrics.recordAndMaybeThrottle("ANONYMOUS", "unknown", 400, callback)
time.sleep(1000)
}
assertEquals("Should be unthrottled since bursty sample has rolled over",
- 0, clientMetrics.recordAndMaybeThrottle("unknown", 0, callback))
+ 0, clientMetrics.recordAndMaybeThrottle("ANONYMOUS", "unknown", 0, callback))
} finally {
clientMetrics.shutdown()
}
@@ -127,14 +273,14 @@ class ClientQuotaManagerTest {
val metrics = newMetrics
val clientMetrics = new ClientQuotaManager(config, metrics, QuotaType.Produce, time)
try {
- clientMetrics.recordAndMaybeThrottle("client1", 100, callback)
+ clientMetrics.recordAndMaybeThrottle("ANONYMOUS", "client1", 100, callback)
// remove the throttle time sensor
- metrics.removeSensor("ProduceThrottleTime-client1")
+ metrics.removeSensor("ProduceThrottleTime-:client1")
// should not throw an exception even if the throttle time sensor does not exist.
- val throttleTime = clientMetrics.recordAndMaybeThrottle("client1", 10000, callback)
+ val throttleTime = clientMetrics.recordAndMaybeThrottle("ANONYMOUS", "client1", 10000, callback)
assertTrue("Should be throttled", throttleTime > 0)
// the sensor should get recreated
- val throttleTimeSensor = metrics.getSensor("ProduceThrottleTime-client1")
+ val throttleTimeSensor = metrics.getSensor("ProduceThrottleTime-:client1")
assertTrue("Throttle time sensor should exist", throttleTimeSensor != null)
} finally {
clientMetrics.shutdown()
@@ -146,26 +292,37 @@ class ClientQuotaManagerTest {
val metrics = newMetrics
val clientMetrics = new ClientQuotaManager(config, metrics, QuotaType.Produce, time)
try {
- clientMetrics.recordAndMaybeThrottle("client1", 100, callback)
+ clientMetrics.recordAndMaybeThrottle("ANONYMOUS", "client1", 100, callback)
// remove all the sensors
- metrics.removeSensor("ProduceThrottleTime-client1")
- metrics.removeSensor("Produce-client1")
+ metrics.removeSensor("ProduceThrottleTime-:client1")
+ metrics.removeSensor("Produce-ANONYMOUS:client1")
// should not throw an exception
- val throttleTime = clientMetrics.recordAndMaybeThrottle("client1", 10000, callback)
+ val throttleTime = clientMetrics.recordAndMaybeThrottle("ANONYMOUS", "client1", 10000, callback)
assertTrue("Should be throttled", throttleTime > 0)
// all the sensors should get recreated
- val throttleTimeSensor = metrics.getSensor("ProduceThrottleTime-client1")
+ val throttleTimeSensor = metrics.getSensor("ProduceThrottleTime-:client1")
assertTrue("Throttle time sensor should exist", throttleTimeSensor != null)
- val byteRateSensor = metrics.getSensor("Produce-client1")
+ val byteRateSensor = metrics.getSensor("Produce-:client1")
assertTrue("Byte rate sensor should exist", byteRateSensor != null)
} finally {
clientMetrics.shutdown()
}
}
+ @Test
+ def testQuotaUserSanitize() {
+ val principal = "CN=Some characters !@#$%&*()_-+=';:,/~"
+ val sanitizedPrincipal = QuotaId.sanitize(principal)
+ // Apart from % used in percent-encoding all characters of sanitized principal must be characters allowed in client-id
+ ConfigCommand.validateChars("sanitized-principal", sanitizedPrincipal.replace('%', '_'))
+ assertEquals(principal, QuotaId.desanitize(sanitizedPrincipal))
+ }
+
def newMetrics: Metrics = {
new Metrics(new MetricConfig(), Collections.emptyList(), time)
}
+
+ private case class UserClient(val user: String, val clientId: String, val configUser: Option[String] = None, val configClientId: Option[String] = None)
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/69356fbc/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala b/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala
index bf11332..14faa80 100644
--- a/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala
+++ b/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala
@@ -56,45 +56,78 @@ class DynamicConfigChangeTest extends KafkaServerTestHarness {
}
}
- @Test
- def testClientQuotaConfigChange() {
- assertTrue("Should contain a ConfigHandler for topics",
- this.servers.head.dynamicConfigHandlers.contains(ConfigType.Client))
- val clientId = "testClient"
+ private def testQuotaConfigChange(user: String, clientId: String, rootEntityType: String, configEntityName: String) {
+ assertTrue("Should contain a ConfigHandler for " + rootEntityType ,
+ this.servers.head.dynamicConfigHandlers.contains(rootEntityType))
val props = new Properties()
- props.put(ClientConfigOverride.ProducerOverride, "1000")
- props.put(ClientConfigOverride.ConsumerOverride, "2000")
- AdminUtils.changeClientIdConfig(zkUtils, clientId, props)
+ props.put(QuotaConfigOverride.ProducerOverride, "1000")
+ props.put(QuotaConfigOverride.ConsumerOverride, "2000")
val quotaManagers = servers.head.apis.quotas
+ rootEntityType match {
+ case ConfigType.Client => AdminUtils.changeClientIdConfig(zkUtils, configEntityName, props)
+ case _ => AdminUtils.changeUserOrUserClientIdConfig(zkUtils, configEntityName, props)
+ }
TestUtils.retry(10000) {
- val overrideProducerQuota = quotaManagers.produce.quota(clientId)
- val overrideConsumerQuota = quotaManagers.fetch.quota(clientId)
+ val overrideProducerQuota = quotaManagers.produce.quota(user, clientId)
+ val overrideConsumerQuota = quotaManagers.fetch.quota(user, clientId)
- assertEquals(s"ClientId $clientId must have overridden producer quota of 1000",
+ assertEquals(s"User $user clientId $clientId must have overridden producer quota of 1000",
Quota.upperBound(1000), overrideProducerQuota)
- assertEquals(s"ClientId $clientId must have overridden consumer quota of 2000",
+ assertEquals(s"User $user clientId $clientId must have overridden consumer quota of 2000",
Quota.upperBound(2000), overrideConsumerQuota)
}
- val defaultProducerQuota = servers.head.apis.config.producerQuotaBytesPerSecondDefault.doubleValue
- val defaultConsumerQuota = servers.head.apis.config.consumerQuotaBytesPerSecondDefault.doubleValue
- assertNotEquals("defaultProducerQuota should be different from 1000", 1000, defaultProducerQuota)
- assertNotEquals("defaultConsumerQuota should be different from 2000", 2000, defaultConsumerQuota)
- AdminUtils.changeClientIdConfig(zkUtils, clientId, new Properties())
+ val defaultProducerQuota = Long.MaxValue.asInstanceOf[Double]
+ val defaultConsumerQuota = Long.MaxValue.asInstanceOf[Double]
+ val emptyProps = new Properties()
+ rootEntityType match {
+ case ConfigType.Client => AdminUtils.changeClientIdConfig(zkUtils, configEntityName, emptyProps)
+ case _ => AdminUtils.changeUserOrUserClientIdConfig(zkUtils, configEntityName, emptyProps)
+ }
TestUtils.retry(10000) {
- val producerQuota = quotaManagers.produce.quota(clientId)
- val consumerQuota = quotaManagers.fetch.quota(clientId)
+ val producerQuota = quotaManagers.produce.quota(user, clientId)
+ val consumerQuota = quotaManagers.fetch.quota(user, clientId)
- assertEquals(s"ClientId $clientId must have reset producer quota to " + defaultProducerQuota,
+ assertEquals(s"User $user clientId $clientId must have reset producer quota to " + defaultProducerQuota,
Quota.upperBound(defaultProducerQuota), producerQuota)
- assertEquals(s"ClientId $clientId must have reset consumer quota to " + defaultConsumerQuota,
+ assertEquals(s"User $user clientId $clientId must have reset consumer quota to " + defaultConsumerQuota,
Quota.upperBound(defaultConsumerQuota), consumerQuota)
}
}
@Test
+ def testClientIdQuotaConfigChange() {
+ testQuotaConfigChange("ANONYMOUS", "testClient", ConfigType.Client, "testClient")
+ }
+
+ @Test
+ def testUserQuotaConfigChange() {
+ testQuotaConfigChange("ANONYMOUS", "testClient", ConfigType.User, "ANONYMOUS")
+ }
+
+ @Test
+ def testUserClientIdQuotaChange() {
+ testQuotaConfigChange("ANONYMOUS", "testClient", ConfigType.User, "ANONYMOUS/clients/testClient")
+ }
+
+ @Test
+ def testDefaultClientIdQuotaConfigChange() {
+ testQuotaConfigChange("ANONYMOUS", "testClient", ConfigType.Client, "<default>")
+ }
+
+ @Test
+ def testDefaultUserQuotaConfigChange() {
+ testQuotaConfigChange("ANONYMOUS", "testClient", ConfigType.User, "<default>")
+ }
+
+ @Test
+ def testDefaultUserClientIdQuotaConfigChange() {
+ testQuotaConfigChange("ANONYMOUS", "testClient", ConfigType.User, "<default>/clients/<default>")
+ }
+
+ @Test
def testConfigChangeOnNonExistingTopic() {
val topic = TestUtils.tempTopic
try {
@@ -205,4 +238,4 @@ class DynamicConfigChangeTest extends KafkaServerTestHarness {
//Then
assertEquals(Seq(), result)
}
-}
\ No newline at end of file
+}
[2/2] kafka git commit: KAFKA-3492;
Secure quotas for authenticated users
Posted by ju...@apache.org.
KAFKA-3492; Secure quotas for authenticated users
Implementation and tests for secure quotas at <user> and <user, client-id> levels as described in KIP-55. Also adds dynamic default quotas for <client-id>, <user> and <user-client-id>. For each client connection, the most specific quota matching the connection is used, with user quota taking precedence over client-id quota.
Author: Rajini Sivaram <ra...@googlemail.com>
Reviewers: Jun Rao <ju...@gmail.com>
Closes #1753 from rajinisivaram/KAFKA-3492
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/69356fbc
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/69356fbc
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/69356fbc
Branch: refs/heads/trunk
Commit: 69356fbc6e76ab4291ff4957f0d6ea04e7245909
Parents: ecc1fb1
Author: Rajini Sivaram <ra...@googlemail.com>
Authored: Sat Sep 17 10:06:05 2016 -0700
Committer: Jun Rao <ju...@gmail.com>
Committed: Sat Sep 17 10:06:05 2016 -0700
----------------------------------------------------------------------
.../kafka/common/metrics/KafkaMetric.java | 2 +-
.../org/apache/kafka/common/metrics/Quota.java | 5 +
.../src/main/scala/kafka/admin/AdminUtils.scala | 72 +++-
.../main/scala/kafka/admin/ConfigCommand.scala | 192 +++++++++--
.../scala/kafka/network/RequestChannel.scala | 5 +-
.../scala/kafka/server/ClientQuotaManager.scala | 332 ++++++++++++++++---
.../main/scala/kafka/server/ConfigHandler.scala | 66 ++--
.../kafka/server/DynamicConfigManager.scala | 94 ++++--
.../src/main/scala/kafka/server/KafkaApis.scala | 3 +-
.../main/scala/kafka/server/KafkaConfig.scala | 6 +-
.../main/scala/kafka/server/KafkaServer.scala | 7 +-
core/src/main/scala/kafka/utils/ZkUtils.scala | 3 +
.../integration/kafka/api/BaseQuotaTest.scala | 195 +++++++++++
.../kafka/api/ClientIdQuotaTest.scala | 55 +++
.../kafka/api/ClientQuotasTest.scala | 206 ------------
.../kafka/api/UserClientIdQuotaTest.scala | 66 ++++
.../integration/kafka/api/UserQuotaTest.scala | 61 ++++
.../test/scala/unit/kafka/admin/AdminTest.scala | 4 +-
.../unit/kafka/admin/ConfigCommandTest.scala | 186 ++++++++++-
.../scala/unit/kafka/admin/TestAdminUtils.scala | 1 +
.../kafka/server/ClientQuotaManagerTest.scala | 217 ++++++++++--
.../kafka/server/DynamicConfigChangeTest.scala | 77 +++--
22 files changed, 1449 insertions(+), 406 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/69356fbc/clients/src/main/java/org/apache/kafka/common/metrics/KafkaMetric.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/KafkaMetric.java b/clients/src/main/java/org/apache/kafka/common/metrics/KafkaMetric.java
index e4d3ae8..86014e5 100644
--- a/clients/src/main/java/org/apache/kafka/common/metrics/KafkaMetric.java
+++ b/clients/src/main/java/org/apache/kafka/common/metrics/KafkaMetric.java
@@ -37,7 +37,7 @@ public final class KafkaMetric implements Metric {
this.time = time;
}
- MetricConfig config() {
+ public MetricConfig config() {
return this.config;
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/69356fbc/clients/src/main/java/org/apache/kafka/common/metrics/Quota.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/Quota.java b/clients/src/main/java/org/apache/kafka/common/metrics/Quota.java
index 8431e50..663b963 100644
--- a/clients/src/main/java/org/apache/kafka/common/metrics/Quota.java
+++ b/clients/src/main/java/org/apache/kafka/common/metrics/Quota.java
@@ -67,4 +67,9 @@ public final class Quota {
Quota that = (Quota) obj;
return (that.bound == this.bound) && (that.upper == this.upper);
}
+
+ @Override
+ public String toString() {
+ return (upper ? "upper=" : "lower=") + bound;
+ }
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/69356fbc/core/src/main/scala/kafka/admin/AdminUtils.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/admin/AdminUtils.scala b/core/src/main/scala/kafka/admin/AdminUtils.scala
index 400cc47..b3f8e5c 100644
--- a/core/src/main/scala/kafka/admin/AdminUtils.scala
+++ b/core/src/main/scala/kafka/admin/AdminUtils.scala
@@ -41,6 +41,7 @@ import org.I0Itec.zkclient.exception.ZkNodeExistsException
trait AdminUtilities {
def changeTopicConfig(zkUtils: ZkUtils, topic: String, configs: Properties)
def changeClientIdConfig(zkUtils: ZkUtils, clientId: String, configs: Properties)
+ def changeUserOrUserClientIdConfig(zkUtils: ZkUtils, sanitizedEntityName: String, configs: Properties)
def changeBrokerConfig(zkUtils: ZkUtils, brokerIds: Seq[Int], configs: Properties)
def fetchEntityConfig(zkUtils: ZkUtils,entityType: String, entityName: String): Properties
}
@@ -449,7 +450,7 @@ object AdminUtils extends Logging with AdminUtilities {
if (!update) {
// write out the config if there is any, this isn't transactional with the partition assignments
LogConfig.validate(config)
- writeEntityConfig(zkUtils, ConfigType.Topic, topic, config)
+ writeEntityConfig(zkUtils, getEntityConfigPath(ConfigType.Topic, topic), config)
}
// create the partition assignment
@@ -476,7 +477,9 @@ object AdminUtils extends Logging with AdminUtilities {
}
/**
- * Update the config for a client and create a change notification so the change will propagate to other brokers
+ * Update the config for a client and create a change notification so the change will propagate to other brokers.
+ * If clientId is <default>, default clientId config is updated. ClientId configs are used only if <user, clientId>
+ * and <user> configs are not specified.
*
* @param zkUtils Zookeeper utilities used to write the config to ZK
* @param clientId: The clientId for which configs are being changed
@@ -489,6 +492,21 @@ object AdminUtils extends Logging with AdminUtilities {
}
/**
+ * Update the config for a <user> or <user, clientId> and create a change notification so the change will propagate to other brokers.
+ * User and/or clientId components of the path may be <default>, indicating that the configuration is the default
+ * value to be applied if a more specific override is not configured.
+ *
+ * @param zkUtils Zookeeper utilities used to write the config to ZK
+ * @param sanitizedEntityName: <sanitizedUserPrincipal> or <sanitizedUserPrincipal>/clients/<clientId>
+ * @param configs: The final set of configs that will be applied to the topic. If any new configs need to be added or
+ * existing configs need to be deleted, it should be done prior to invoking this API
+ *
+ */
+ def changeUserOrUserClientIdConfig(zkUtils: ZkUtils, sanitizedEntityName: String, configs: Properties) {
+ changeEntityConfig(zkUtils, ConfigType.User, sanitizedEntityName, configs)
+ }
+
+ /**
* Update the config for an existing topic and create a change notification so the change will propagate to other brokers
*
* @param zkUtils Zookeeper utilities used to write the config to ZK
@@ -520,37 +538,41 @@ object AdminUtils extends Logging with AdminUtilities {
}
}
- private def changeEntityConfig(zkUtils: ZkUtils, entityType: String, entityName: String, configs: Properties) {
+ private def changeEntityConfig(zkUtils: ZkUtils, rootEntityType: String, fullSanitizedEntityName: String, configs: Properties) {
+ val sanitizedEntityPath = rootEntityType + '/' + fullSanitizedEntityName
+ val entityConfigPath = getEntityConfigPath(rootEntityType, fullSanitizedEntityName)
// write the new config--may not exist if there were previously no overrides
- writeEntityConfig(zkUtils, entityType, entityName, configs)
+ writeEntityConfig(zkUtils, entityConfigPath, configs)
// create the change notification
val seqNode = ZkUtils.EntityConfigChangesPath + "/" + EntityConfigChangeZnodePrefix
- val content = Json.encode(getConfigChangeZnodeData(entityType, entityName))
+ val content = Json.encode(getConfigChangeZnodeData(sanitizedEntityPath))
zkUtils.zkClient.createPersistentSequential(seqNode, content)
}
- def getConfigChangeZnodeData(entityType: String, entityName: String) : Map[String, Any] = {
- Map("version" -> 1, "entity_type" -> entityType, "entity_name" -> entityName)
+ def getConfigChangeZnodeData(sanitizedEntityPath: String) : Map[String, Any] = {
+ Map("version" -> 2, "entity_path" -> sanitizedEntityPath)
}
/**
- * Write out the topic config to zk, if there is any
+ * Write out the entity config to zk, if there is any
*/
- private def writeEntityConfig(zkUtils: ZkUtils, entityType: String, entityName: String, config: Properties) {
+ private def writeEntityConfig(zkUtils: ZkUtils, entityPath: String, config: Properties) {
val configMap: mutable.Map[String, String] = {
import JavaConversions._
config
}
val map = Map("version" -> 1, "config" -> configMap)
- zkUtils.updatePersistentPath(getEntityConfigPath(entityType, entityName), Json.encode(map))
+ zkUtils.updatePersistentPath(entityPath, Json.encode(map))
}
/**
- * Read the entity (topic or client) config (if any) from zk
+ * Read the entity (topic, broker, client, user or <user, client>) config (if any) from zk
+ * sanitizedEntityName is <topic>, <broker>, <client-id>, <user> or <user>/clients/<client-id>.
*/
- def fetchEntityConfig(zkUtils: ZkUtils, entityType: String, entity: String): Properties = {
- val str: String = zkUtils.zkClient.readData(getEntityConfigPath(entityType, entity), true)
+ def fetchEntityConfig(zkUtils: ZkUtils, rootEntityType: String, sanitizedEntityName: String): Properties = {
+ val entityConfigPath = getEntityConfigPath(rootEntityType, sanitizedEntityName)
+ val str: String = zkUtils.zkClient.readData(entityConfigPath, true)
val props = new Properties()
if (str != null) {
Json.parseFull(str) match {
@@ -564,13 +586,12 @@ object AdminUtils extends Logging with AdminUtilities {
configTup match {
case (k: String, v: String) =>
props.setProperty(k, v)
- case _ => throw new IllegalArgumentException("Invalid " + entityType + " config: " + str)
+ case _ => throw new IllegalArgumentException(s"Invalid ${entityConfigPath} config: ${str}")
}
- case _ => throw new IllegalArgumentException("Invalid " + entityType + " config: " + str)
+ case _ => throw new IllegalArgumentException(s"Invalid ${entityConfigPath} config: ${str}")
}
- case o => throw new IllegalArgumentException("Unexpected value in config:(%s), entity_type: (%s), entity: (%s)"
- .format(str, entityType, entity))
+ case o => throw new IllegalArgumentException(s"Unexpected value in config:(${str}), entity_config_path: ${entityConfigPath}")
}
}
props
@@ -582,6 +603,23 @@ object AdminUtils extends Logging with AdminUtilities {
def fetchAllEntityConfigs(zkUtils: ZkUtils, entityType: String): Map[String, Properties] =
zkUtils.getAllEntitiesWithConfig(entityType).map(entity => (entity, fetchEntityConfig(zkUtils, entityType, entity))).toMap
+ def fetchAllChildEntityConfigs(zkUtils: ZkUtils, rootEntityType: String, childEntityType: String): Map[String, Properties] = {
+ def entityPaths(zkUtils: ZkUtils, rootPath: Option[String]): Seq[String] = {
+ val root = rootPath match {
+ case Some(path) => rootEntityType + '/' + rootPath
+ case None => rootEntityType
+ }
+ val entityNames = zkUtils.getAllEntitiesWithConfig(root)
+ rootPath match {
+ case Some(path) => entityNames.map(entityName => path + '/' + entityName)
+ case None => entityNames
+ }
+ }
+ entityPaths(zkUtils, None)
+ .flatMap(entity => entityPaths(zkUtils, Some(entity + '/' + childEntityType)))
+ .map(entityPath => (entityPath, fetchEntityConfig(zkUtils, rootEntityType, entityPath))).toMap
+ }
+
def fetchTopicMetadataFromZk(topic: String, zkUtils: ZkUtils): MetadataResponse.TopicMetadata =
fetchTopicMetadataFromZk(topic, zkUtils, new mutable.HashMap[Int, Broker])
http://git-wip-us.apache.org/repos/asf/kafka/blob/69356fbc/core/src/main/scala/kafka/admin/ConfigCommand.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/admin/ConfigCommand.scala b/core/src/main/scala/kafka/admin/ConfigCommand.scala
index ebf9e61..58bdb7a 100644
--- a/core/src/main/scala/kafka/admin/ConfigCommand.scala
+++ b/core/src/main/scala/kafka/admin/ConfigCommand.scala
@@ -21,8 +21,9 @@ import java.util.Properties
import joptsimple._
import kafka.admin.TopicCommand._
+import kafka.common.Config
import kafka.log.{Defaults, LogConfig}
-import kafka.server.{KafkaConfig, ClientConfigOverride, ConfigType}
+import kafka.server.{KafkaConfig, QuotaConfigOverride, ConfigType, ConfigEntityName, QuotaId}
import kafka.utils.{CommandLineUtils, ZkUtils}
import org.apache.kafka.common.security.JaasUtils
import org.apache.kafka.common.utils.Utils
@@ -33,15 +34,26 @@ import scala.collection._
/**
* This script can be used to change configs for topics/clients/brokers dynamically
+ * This script can be used to change configs for topics/clients/users/brokers dynamically
+ * An entity described or altered by the command may be one of:
+ * <ul>
+ * <li> topic: --entity-type topics --entity-name <topic>
+ * <li> client: --entity-type clients --entity-name <client-id>
+ * <li> user: --entity-type users --entity-name <user-principal>
+ * <li> <user, client>: --entity-type users --entity-name <user-principal> --entity-type clients --entity-name <client-id>
+ * <li> broker: --entity-type brokers --entity-name <broker>
+ * </ul>
+ * --entity-default may be used instead of --entity-name when describing or altering default configuration for users and clients.
+ *
*/
-object ConfigCommand {
+object ConfigCommand extends Config {
def main(args: Array[String]): Unit = {
val opts = new ConfigCommandOptions(args)
if(args.length == 0)
- CommandLineUtils.printUsageAndDie(opts.parser, "Add/Remove entity config for a topic, client or broker")
+ CommandLineUtils.printUsageAndDie(opts.parser, "Add/Remove entity config for a topic, client, user or broker")
opts.checkArgs()
@@ -57,18 +69,19 @@ object ConfigCommand {
describeConfig(zkUtils, opts)
} catch {
case e: Throwable =>
- println("Error while executing topic command " + e.getMessage)
+ println("Error while executing config command " + e.getMessage)
println(Utils.stackTrace(e))
} finally {
zkUtils.close()
}
}
- def alterConfig(zkUtils: ZkUtils, opts: ConfigCommandOptions, utils: AdminUtilities = AdminUtils) {
+ private[admin] def alterConfig(zkUtils: ZkUtils, opts: ConfigCommandOptions, utils: AdminUtilities = AdminUtils) {
val configsToBeAdded = parseConfigsToBeAdded(opts)
val configsToBeDeleted = parseConfigsToBeDeleted(opts)
- val entityType = opts.options.valueOf(opts.entityType)
- val entityName = opts.options.valueOf(opts.entityName)
+ val entity = parseEntity(opts)
+ val entityType = entity.root.entityType
+ val entityName = entity.fullSanitizedName
warnOnMaxMessagesChange(configsToBeAdded, opts.options.has(opts.forceOpt))
// compile the final set of configs
@@ -77,12 +90,13 @@ object ConfigCommand {
configsToBeDeleted.foreach(config => configs.remove(config))
entityType match {
- case ConfigType.Topic => utils.changeTopicConfig(zkUtils, entityName, configs)
- case ConfigType.Client => utils.changeClientIdConfig(zkUtils, entityName, configs)
+ case ConfigType.Topic => utils.changeTopicConfig(zkUtils, entityName, configs)
+ case ConfigType.Client => utils.changeClientIdConfig(zkUtils, entityName, configs)
+ case ConfigType.User => utils.changeUserOrUserClientIdConfig(zkUtils, entityName, configs)
case ConfigType.Broker => utils.changeBrokerConfig(zkUtils, Seq(parseBroker(entityName)), configs)
case _ => throw new IllegalArgumentException(s"$entityType is not a known entityType. Should be one of ${ConfigType.Topic}, ${ConfigType.Client}, ${ConfigType.Broker}")
}
- println(s"Updated config for EntityType:$entityType => EntityName:'$entityName'.")
+ println(s"Updated config for entity: $entity.")
}
def warnOnMaxMessagesChange(configs: Properties, force: Boolean): Unit = {
@@ -107,17 +121,16 @@ object ConfigCommand {
}
private def describeConfig(zkUtils: ZkUtils, opts: ConfigCommandOptions) {
- val entityType = opts.options.valueOf(opts.entityType)
- val entityNames: Seq[String] =
- if (opts.options.has(opts.entityName))
- Seq(opts.options.valueOf(opts.entityName))
- else
- zkUtils.getAllEntitiesWithConfig(entityType)
-
- for (entityName <- entityNames) {
- val configs = AdminUtils.fetchEntityConfig(zkUtils, entityType, entityName)
- println("Configs for %s:%s are %s"
- .format(entityType, entityName, configs.map(kv => kv._1 + "=" + kv._2).mkString(",")))
+ val configEntity = parseEntity(opts)
+ val describeAllUsers = configEntity.root.entityType == ConfigType.User && !configEntity.root.sanitizedName.isDefined && !configEntity.child.isDefined
+ val entities = configEntity.getAllEntities(zkUtils)
+ for (entity <- entities) {
+ val configs = AdminUtils.fetchEntityConfig(zkUtils, entity.root.entityType, entity.fullSanitizedName)
+ // When describing all users, don't include empty user nodes with only <user, client> quota overrides.
+ if (!configs.isEmpty || !describeAllUsers) {
+ println("Configs for %s are %s"
+ .format(entity, configs.map(kv => kv._1 + "=" + kv._2).mkString(",")))
+ }
}
}
@@ -150,6 +163,115 @@ object ConfigCommand {
Seq.empty
}
+ case class Entity(entityType: String, sanitizedName: Option[String]) {
+ val entityPath = sanitizedName match {
+ case Some(n) => entityType + "/" + n
+ case None => entityType
+ }
+ override def toString: String = {
+ val typeName = entityType match {
+ case ConfigType.User => "user-principal"
+ case ConfigType.Client => "client-id"
+ case ConfigType.Topic => "topic"
+ case t => t
+ }
+ sanitizedName match {
+ case Some(ConfigEntityName.Default) => "default " + typeName
+ case Some(n) =>
+ val desanitized = if (entityType == ConfigType.User) QuotaId.desanitize(n) else n
+ s"$typeName '$desanitized'"
+ case None => entityType
+ }
+ }
+ }
+
+ case class ConfigEntity(root: Entity, child: Option[Entity]) {
+ val fullSanitizedName = root.sanitizedName.getOrElse("") + child.map(s => "/" + s.entityPath).getOrElse("")
+
+ def getAllEntities(zkUtils: ZkUtils) : Seq[ConfigEntity] = {
+ // Describe option examples:
+ // Describe entity with specified name:
+ // --entity-type topics --entity-name topic1 (topic1)
+ // Describe all entities of a type (topics/brokers/users/clients):
+ // --entity-type topics (all topics)
+ // Describe <user, client> quotas:
+ // --entity-type users --entity-name user1 --entity-type clients --entity-name client2 (<user1, client2>)
+ // --entity-type users --entity-name userA --entity-type clients (all clients of userA)
+ // --entity-type users --entity-type clients (all <user, client>s))
+ // Describe default quotas:
+ // --entity-type users --entity-default (Default user)
+ // --entity-type users --entity-default --entity-type clients --entity-default (Default <user, client>)
+ (root.sanitizedName, child) match {
+ case (None, _) =>
+ val rootEntities = zkUtils.getAllEntitiesWithConfig(root.entityType)
+ .map(name => ConfigEntity(Entity(root.entityType, Some(name)), child))
+ child match {
+ case Some (s) =>
+ rootEntities.flatMap(rootEntity =>
+ ConfigEntity(rootEntity.root, Some(Entity(s.entityType, None))).getAllEntities(zkUtils))
+ case None => rootEntities
+ }
+ case (rootName, Some(childEntity)) =>
+ childEntity.sanitizedName match {
+ case Some(subName) => Seq(this)
+ case None =>
+ zkUtils.getAllEntitiesWithConfig(root.entityPath + "/" + childEntity.entityType)
+ .map(name => ConfigEntity(root, Some(Entity(childEntity.entityType, Some(name)))))
+
+ }
+ case (rootName, None) =>
+ Seq(this)
+ }
+ }
+
+ override def toString: String = {
+ root.toString + child.map(s => ", " + s.toString).getOrElse("")
+ }
+ }
+
+ private[admin] def parseEntity(opts: ConfigCommandOptions): ConfigEntity = {
+ val entityTypes = opts.options.valuesOf(opts.entityType)
+ if (entityTypes.head == ConfigType.User || entityTypes.head == ConfigType.Client)
+ parseQuotaEntity(opts)
+ else {
+ // Exactly one entity type and at-most one entity name expected for other entities
+ val name = if (opts.options.has(opts.entityName)) Some(opts.options.valueOf(opts.entityName)) else None
+ ConfigEntity(Entity(entityTypes.head, name), None)
+ }
+ }
+
+ private def parseQuotaEntity(opts: ConfigCommandOptions): ConfigEntity = {
+ val types = opts.options.valuesOf(opts.entityType)
+ val namesIterator = opts.options.valuesOf(opts.entityName).iterator
+ val names = opts.options.specs
+ .filter(spec => spec.options.contains("entity-name") || spec.options.contains("entity-default"))
+ .map(spec => if (spec.options.contains("entity-name")) namesIterator.next else "")
+
+ if (opts.options.has(opts.alterOpt) && names.size != types.size)
+ throw new IllegalArgumentException("--entity-name or --entity-default must be specified with each --entity-type for --alter")
+
+ val reverse = types.size == 2 && types(0) == ConfigType.Client
+ val entityTypes = if (reverse) types.reverse else types.toBuffer
+ val sortedNames = (if (reverse && names.length == 2) names.reverse else names).iterator
+
+ def sanitizeName(entityType: String, name: String) = {
+ if (name.isEmpty)
+ ConfigEntityName.Default
+ else {
+ entityType match {
+ case ConfigType.User => QuotaId.sanitize(name)
+ case ConfigType.Client =>
+ validateChars("Client-id", name)
+ name
+ case _ => throw new IllegalArgumentException("Invalid entity type " + entityType)
+ }
+ }
+ }
+
+ val entities = entityTypes.map(t => Entity(t, if (sortedNames.hasNext) Some(sanitizeName(t, sortedNames.next)) else None))
+ ConfigEntity(entities.head, if (entities.size > 1) Some(entities(1)) else None)
+ }
+
class ConfigCommandOptions(args: Array[String]) {
val parser = new OptionParser
val zkConnectOpt = parser.accepts("zookeeper", "REQUIRED: The connection string for the zookeeper connection in the form host:port. " +
@@ -159,19 +281,23 @@ object ConfigCommand {
.ofType(classOf[String])
val alterOpt = parser.accepts("alter", "Alter the configuration for the entity.")
val describeOpt = parser.accepts("describe", "List configs for the given entity.")
- val entityType = parser.accepts("entity-type", "Type of entity (topics/clients/brokers)")
+ val entityType = parser.accepts("entity-type", "Type of entity (topics/clients/users/brokers)")
.withRequiredArg
.ofType(classOf[String])
- val entityName = parser.accepts("entity-name", "Name of entity (topic name/client id/broker id)")
+ val entityName = parser.accepts("entity-name", "Name of entity (topic name/client id/user principal name/broker id)")
.withRequiredArg
.ofType(classOf[String])
+ val entityDefault = parser.accepts("entity-default", "Default entity name for clients/users (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: " +
"For entity_type '" + ConfigType.Topic + "': " + nl + LogConfig.configNames.map("\t" + _).mkString(nl) + nl +
"For entity_type '" + ConfigType.Broker + "': " + nl + KafkaConfig.dynamicBrokerConfigs.map("\t" + _).mkString(nl) + nl +
- "For entity_type '" + ConfigType.Client + "': " + nl + "\t" + ClientConfigOverride.ProducerOverride
- + nl + "\t" + ClientConfigOverride.ConsumerOverride)
+ "For entity_type '" + ConfigType.Client + "': " + nl + "\t" + QuotaConfigOverride.ProducerOverride
+ + nl + "\t" + QuotaConfigOverride.ConsumerOverride + nl +
+ "For entity_type '" + ConfigType.User + "': " + nl + "\t" + QuotaConfigOverride.ProducerOverride
+ + nl + "\t" + QuotaConfigOverride.ConsumerOverride + 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])
val deleteConfig = parser.accepts("delete-config", "config keys to remove 'k1,k2'")
@@ -194,15 +320,27 @@ object ConfigCommand {
CommandLineUtils.checkRequiredArgs(parser, options, zkConnectOpt, entityType)
CommandLineUtils.checkInvalidArgs(parser, options, alterOpt, Set(describeOpt))
CommandLineUtils.checkInvalidArgs(parser, options, describeOpt, Set(alterOpt, addConfig, deleteConfig))
+ val entityTypeVals = options.valuesOf(entityType)
if(options.has(alterOpt)) {
- require(options.has(entityName), "--entity-name must be specified with --alter")
+ if (entityTypeVals.contains(ConfigType.User) || entityTypeVals.contains(ConfigType.Client)) {
+ if (!options.has(entityName) && !options.has(entityDefault))
+ throw new IllegalArgumentException("--entity-name or --entity-default must be specified with --alter of users/clients")
+ } else if (!options.has(entityName))
+ throw new IllegalArgumentException(s"--entity-name must be specified with --alter of ${entityTypeVals}")
val isAddConfigPresent: Boolean = options.has(addConfig)
val isDeleteConfigPresent: Boolean = options.has(deleteConfig)
if(! isAddConfigPresent && ! isDeleteConfigPresent)
throw new IllegalArgumentException("At least one of --add-config or --delete-config must be specified with --alter")
}
- require(ConfigType.all.contains(options.valueOf(entityType)), s"--entity-type must be one of ${ConfigType.all}")
+ entityTypeVals.foreach(entityTypeVal =>
+ if (!ConfigType.all.contains(entityTypeVal))
+ throw new IllegalArgumentException(s"Invalid entity-type ${entityTypeVal}, --entity-type must be one of ${ConfigType.all}")
+ )
+ if (entityTypeVals.isEmpty)
+ throw new IllegalArgumentException("At least one --entity-type must be specified")
+ else if (entityTypeVals.size > 1 && !entityTypeVals.toSet.equals(Set(ConfigType.User, ConfigType.Client)))
+ throw new IllegalArgumentException(s"Only '${ConfigType.User}' and '${ConfigType.Client}' entity types may be specified together")
}
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/69356fbc/core/src/main/scala/kafka/network/RequestChannel.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/network/RequestChannel.scala b/core/src/main/scala/kafka/network/RequestChannel.scala
index cff7b1a..8aec2d2 100644
--- a/core/src/main/scala/kafka/network/RequestChannel.scala
+++ b/core/src/main/scala/kafka/network/RequestChannel.scala
@@ -25,6 +25,7 @@ import java.util.concurrent._
import com.yammer.metrics.core.Gauge
import kafka.api._
import kafka.metrics.KafkaMetricsGroup
+import kafka.server.QuotaId
import kafka.utils.{Logging, SystemTime}
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.errors.InvalidRequestException
@@ -44,7 +45,9 @@ object RequestChannel extends Logging {
RequestSend.serialize(emptyRequestHeader, emptyProduceRequest.toStruct)
}
- case class Session(principal: KafkaPrincipal, clientAddress: InetAddress)
+ case class Session(principal: KafkaPrincipal, clientAddress: InetAddress) {
+ val sanitizedUser = QuotaId.sanitize(principal.getName)
+ }
case class Request(processor: Int, connectionId: String, session: Session, private var buffer: ByteBuffer, startTimeMs: Long, securityProtocol: SecurityProtocol) {
// These need to be volatile because the readers are in the network thread and the writers are in the request
http://git-wip-us.apache.org/repos/asf/kafka/blob/69356fbc/core/src/main/scala/kafka/server/ClientQuotaManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/ClientQuotaManager.scala b/core/src/main/scala/kafka/server/ClientQuotaManager.scala
index e6cac5d..c4472c6 100644
--- a/core/src/main/scala/kafka/server/ClientQuotaManager.scala
+++ b/core/src/main/scala/kafka/server/ClientQuotaManager.scala
@@ -16,15 +16,19 @@
*/
package kafka.server
+import java.net.{URLEncoder, URLDecoder}
+import java.nio.charset.StandardCharsets
import java.util.concurrent.{ConcurrentHashMap, DelayQueue, TimeUnit}
+import java.util.concurrent.locks.ReentrantReadWriteLock
+
import kafka.utils.{ShutdownableThread, Logging}
import org.apache.kafka.common.MetricName
import org.apache.kafka.common.metrics._
import org.apache.kafka.common.metrics.stats.{Total, Rate, Avg}
-import java.util.concurrent.locks.ReentrantReadWriteLock
-
import org.apache.kafka.common.utils.Time
+import scala.collection.JavaConversions._
+
/**
* Represents the sensors aggregated per client
* @param quotaSensor @Sensor that tracks the quota
@@ -34,7 +38,8 @@ private case class ClientSensors(quotaSensor: Sensor, throttleTimeSensor: Sensor
/**
* Configuration settings for quota management
- * @param quotaBytesPerSecondDefault The default bytes per second quota allocated to any client
+ * @param quotaBytesPerSecondDefault The default bytes per second quota allocated to any client-id if
+ * dynamic defaults or user quotas are not set
* @param numQuotaSamples The number of samples to retain in memory
* @param quotaWindowSizeSeconds The time span of each sample
*
@@ -53,11 +58,72 @@ object ClientQuotaManagerConfig {
val DefaultQuotaWindowSizeSeconds = 1
// Purge sensors after 1 hour of inactivity
val InactiveSensorExpirationTimeSeconds = 3600
+
+ val UnlimitedQuota = Quota.upperBound(Long.MaxValue)
+ val DefaultClientIdQuotaId = QuotaId(None, Some(ConfigEntityName.Default))
+ val DefaultUserQuotaId = QuotaId(Some(ConfigEntityName.Default), None)
+ val DefaultUserClientIdQuotaId = QuotaId(Some(ConfigEntityName.Default), Some(ConfigEntityName.Default))
+}
+
+object QuotaTypes {
+ val NoQuotas = 0
+ val ClientIdQuotaEnabled = 1
+ val UserQuotaEnabled = 2
+ val UserClientIdQuotaEnabled = 4
}
+object QuotaId {
+
+ /**
+ * Sanitizes user principal to a safe value for use in MetricName
+ * and as Zookeeper node name
+ */
+ def sanitize(user: String): String = {
+ val encoded = URLEncoder.encode(user, StandardCharsets.UTF_8.name)
+ val builder = new StringBuilder
+ for (i <- 0 until encoded.length) {
+ encoded.charAt(i) match {
+ case '*' => builder.append("%2A") // Metric ObjectName treats * as pattern
+ case '+' => builder.append("%20") // Space URL-encoded as +, replace with percent encoding
+ case c => builder.append(c)
+ }
+ }
+ builder.toString
+ }
+
+ /**
+ * Decodes sanitized user principal
+ */
+ def desanitize(sanitizedUser: String): String = {
+ URLDecoder.decode(sanitizedUser, StandardCharsets.UTF_8.name)
+ }
+}
+
+case class QuotaId(sanitizedUser: Option[String], clientId: Option[String])
+
+case class QuotaEntity(quotaId: QuotaId, sanitizedUser: String, clientId: String, quota: Quota)
+
/**
* Helper class that records per-client metrics. It is also responsible for maintaining Quota usage statistics
* for all clients.
+ * <p/>
+ * Quotas can be set at <user, client-id>, user or client-id levels. For a given client connection,
+ * the most specific quota matching the connection will be applied. For example, if both a <user, client-id>
+ * and a user quota match a connection, the <user, client-id> quota will be used. Otherwise, user quota takes
+ * precedence over client-id quota. The order of precedence is:
+ * <ul>
+ * <li>/config/users/<user>/clients/<client-id>
+ * <li>/config/users/<user>/clients/<default>
+ * <li>/config/users/<user>
+ * <li>/config/users/<default>/clients/<client-id>
+ * <li>/config/users/<default>/clients/<default>
+ * <li>/config/users/<default>
+ * <li>/config/clients/<client-id>
+ * <li>/config/clients/<default>
+ * </ul>
+ * Quota limits including defaults may be updated dynamically. The implementation is optimized for the case
+ * where a single level of quotas is configured.
+ *
* @param config @ClientQuotaManagerConfig quota configs
* @param metrics @Metrics Metrics instance
* @param apiKey API Key for the request
@@ -67,8 +133,9 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
private val metrics: Metrics,
private val apiKey: QuotaType,
private val time: Time) extends Logging {
- private val overriddenQuota = new ConcurrentHashMap[String, Quota]()
- private val defaultQuota = Quota.upperBound(config.quotaBytesPerSecondDefault)
+ private val overriddenQuota = new ConcurrentHashMap[QuotaId, Quota]()
+ private val staticConfigClientIdQuota = Quota.upperBound(config.quotaBytesPerSecondDefault)
+ private var quotaTypesEnabled = if (config.quotaBytesPerSecondDefault == Long.MaxValue) QuotaTypes.NoQuotas else QuotaTypes.ClientIdQuotaEnabled
private val lock = new ReentrantReadWriteLock()
private val delayQueue = new DelayQueue[ThrottledResponse]()
private val sensorAccessor = new SensorAccess
@@ -107,8 +174,9 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
* @return Number of milliseconds to delay the response in case of Quota violation.
* Zero otherwise
*/
- def recordAndMaybeThrottle(clientId: String, value: Int, callback: Int => Unit): Int = {
- val clientSensors = getOrCreateQuotaSensors(clientId)
+ def recordAndMaybeThrottle(sanitizedUser: String, clientId: String, value: Int, callback: Int => Unit): Int = {
+ val clientQuotaEntity = quotaEntity(sanitizedUser, clientId)
+ val clientSensors = getOrCreateQuotaSensors(clientQuotaEntity)
var throttleTimeMs = 0
try {
clientSensors.quotaSensor.record(value)
@@ -117,8 +185,8 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
} catch {
case qve: QuotaViolationException =>
// Compute the delay
- val clientMetric = metrics.metrics().get(clientRateMetricName(clientId))
- throttleTimeMs = throttleTime(clientMetric, getQuotaMetricConfig(quota(clientId)))
+ val clientMetric = metrics.metrics().get(clientRateMetricName(clientQuotaEntity.sanitizedUser, clientQuotaEntity.clientId))
+ throttleTimeMs = throttleTime(clientMetric, getQuotaMetricConfig(clientQuotaEntity.quota))
clientSensors.throttleTimeSensor.record(throttleTimeMs)
// If delayed, add the element to the delayQueue
delayQueue.add(new ThrottledResponse(time, throttleTimeMs, callback))
@@ -128,6 +196,127 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
throttleTimeMs
}
+ /**
+ * Determines the quota-id for the client with the specified user principal
+ * and client-id and returns the quota entity that encapsulates the quota-id
+ * and the associated quota override or default quota.
+ *
+ */
+ private def quotaEntity(sanitizedUser: String, clientId: String) : QuotaEntity = {
+ quotaTypesEnabled match {
+ case QuotaTypes.NoQuotas | QuotaTypes.ClientIdQuotaEnabled =>
+ val quotaId = QuotaId(None, Some(clientId))
+ var quota = overriddenQuota.get(quotaId)
+ if (quota == null) {
+ quota = overriddenQuota.get(ClientQuotaManagerConfig.DefaultClientIdQuotaId)
+ if (quota == null)
+ quota = staticConfigClientIdQuota
+ }
+ QuotaEntity(quotaId, "", clientId, quota)
+ case QuotaTypes.UserQuotaEnabled =>
+ val quotaId = QuotaId(Some(sanitizedUser), None)
+ var quota = overriddenQuota.get(quotaId)
+ if (quota == null) {
+ quota = overriddenQuota.get(ClientQuotaManagerConfig.DefaultUserQuotaId)
+ if (quota == null)
+ quota = ClientQuotaManagerConfig.UnlimitedQuota
+ }
+ QuotaEntity(quotaId, sanitizedUser, "", quota)
+ case QuotaTypes.UserClientIdQuotaEnabled =>
+ val quotaId = QuotaId(Some(sanitizedUser), Some(clientId))
+ var quota = overriddenQuota.get(quotaId)
+ if (quota == null) {
+ quota = overriddenQuota.get(QuotaId(Some(sanitizedUser), Some(ConfigEntityName.Default)))
+ if (quota == null) {
+ quota = overriddenQuota.get(QuotaId(Some(ConfigEntityName.Default), Some(clientId)))
+ if (quota == null) {
+ quota = overriddenQuota.get(ClientQuotaManagerConfig.DefaultUserClientIdQuotaId)
+ if (quota == null)
+ quota = ClientQuotaManagerConfig.UnlimitedQuota
+ }
+ }
+ }
+ QuotaEntity(quotaId, sanitizedUser, clientId, quota)
+ case _ =>
+ quotaEntityWithMultipleQuotaLevels(sanitizedUser, clientId)
+ }
+ }
+
+ private def quotaEntityWithMultipleQuotaLevels(sanitizedUser: String, clientId: String) : QuotaEntity = {
+ val userClientQuotaId = QuotaId(Some(sanitizedUser), Some(clientId))
+
+ val userQuotaId = QuotaId(Some(sanitizedUser), None)
+ val clientQuotaId = QuotaId(None, Some(clientId))
+ var quotaId = userClientQuotaId
+ var quotaConfigId = userClientQuotaId
+ // 1) /config/users/<user>/clients/<client-id>
+ var quota = overriddenQuota.get(quotaConfigId)
+ if (quota == null) {
+ // 2) /config/users/<user>/clients/<default>
+ quotaId = userClientQuotaId
+ quotaConfigId = QuotaId(Some(sanitizedUser), Some(ConfigEntityName.Default))
+ quota = overriddenQuota.get(quotaConfigId)
+
+ if (quota == null) {
+ // 3) /config/users/<user>
+ quotaId = userQuotaId
+ quotaConfigId = quotaId
+ quota = overriddenQuota.get(quotaConfigId)
+
+ if (quota == null) {
+ // 4) /config/users/<default>/clients/<client-id>
+ quotaId = userClientQuotaId
+ quotaConfigId = QuotaId(Some(ConfigEntityName.Default), Some(clientId))
+ quota = overriddenQuota.get(quotaConfigId)
+
+ if (quota == null) {
+ // 5) /config/users/<default>/clients/<default>
+ quotaId = userClientQuotaId
+ quotaConfigId = QuotaId(Some(ConfigEntityName.Default), Some(ConfigEntityName.Default))
+ quota = overriddenQuota.get(quotaConfigId)
+
+ if (quota == null) {
+ // 6) /config/users/<default>
+ quotaId = userQuotaId
+ quotaConfigId = QuotaId(Some(ConfigEntityName.Default), None)
+ quota = overriddenQuota.get(quotaConfigId)
+
+ if (quota == null) {
+ // 7) /config/clients/<client-id>
+ quotaId = clientQuotaId
+ quotaConfigId = QuotaId(None, Some(clientId))
+ quota = overriddenQuota.get(quotaConfigId)
+
+ if (quota == null) {
+ // 8) /config/clients/<default>
+ quotaId = clientQuotaId
+ quotaConfigId = QuotaId(None, Some(ConfigEntityName.Default))
+ quota = overriddenQuota.get(quotaConfigId)
+
+ if (quota == null) {
+ quotaId = clientQuotaId
+ quotaConfigId = null
+ quota = staticConfigClientIdQuota
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+ val quotaUser = if (quotaId == clientQuotaId) "" else sanitizedUser
+ val quotaClientId = if (quotaId == userQuotaId) "" else clientId
+ QuotaEntity(quotaId, quotaUser, quotaClientId, quota)
+ }
+
+ /**
+ * Returns the quota for the client with the specified (non-encoded) user principal and client-id.
+ */
+ def quota(user: String, clientId: String) = {
+ quotaEntity(QuotaId.sanitize(user), clientId).quota
+ }
+
/*
* This calculates the amount of time needed to bring the metric within quota
* assuming that no new metrics are recorded.
@@ -153,40 +342,35 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
}
}
- /**
- * Returns the quota for the specified clientId
- */
- def quota(clientId: String): Quota =
- if (overriddenQuota.containsKey(clientId)) overriddenQuota.get(clientId) else defaultQuota
-
/*
* This function either returns the sensors for a given client id or creates them if they don't exist
* First sensor of the tuple is the quota enforcement sensor. Second one is the throttle time sensor
*/
- private def getOrCreateQuotaSensors(clientId: String): ClientSensors = {
+ private def getOrCreateQuotaSensors(quotaEntity: QuotaEntity): ClientSensors = {
+ // Names of the sensors to access
ClientSensors(
sensorAccessor.getOrCreate(
- getQuotaSensorName(clientId),
+ getQuotaSensorName(quotaEntity.quotaId),
ClientQuotaManagerConfig.InactiveSensorExpirationTimeSeconds,
lock, metrics,
- () => clientRateMetricName(clientId),
- () => getQuotaMetricConfig(quota(clientId)),
+ () => clientRateMetricName(quotaEntity.sanitizedUser, quotaEntity.clientId),
+ () => getQuotaMetricConfig(quotaEntity.quota),
() => new Rate()
),
- sensorAccessor.getOrCreate(getThrottleTimeSensorName(clientId),
+ sensorAccessor.getOrCreate(getThrottleTimeSensorName(quotaEntity.quotaId),
ClientQuotaManagerConfig.InactiveSensorExpirationTimeSeconds,
lock,
metrics,
- () => metrics.metricName("throttle-time", apiKey.toString, "Tracking average throttle-time per client", "client-id", clientId),
+ () => throttleMetricName(quotaEntity),
() => null,
() => new Avg()
)
)
}
- private def getThrottleTimeSensorName(clientId: String): String = apiKey + "ThrottleTime-" + clientId
+ private def getThrottleTimeSensorName(quotaId: QuotaId): String = apiKey + "ThrottleTime-" + quotaId.sanitizedUser.getOrElse("") + ':' + quotaId.clientId.getOrElse("")
- private def getQuotaSensorName(clientId: String): String = apiKey + "-" + clientId
+ private def getQuotaSensorName(quotaId: QuotaId): String = apiKey + "-" + quotaId.sanitizedUser.getOrElse("") + ':' + quotaId.clientId.getOrElse("")
private def getQuotaMetricConfig(quota: Quota): MetricConfig = {
new MetricConfig()
@@ -196,19 +380,13 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
}
/**
- * Reset quotas to the default value for the given clientId
- * @param clientId client to override
- */
- def resetQuota(clientId: String) = {
- updateQuota(clientId, defaultQuota)
- }
-
- /**
- * Overrides quotas per clientId
- * @param clientId client to override
- * @param quota custom quota to apply
+ * 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 quota custom quota to apply or None if quota override is being removed
*/
- def updateQuota(clientId: String, quota: Quota) = {
+ def updateQuota(sanitizedUser: Option[String], clientId: Option[String], quota: Option[Quota]) {
/*
* 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).
@@ -218,31 +396,85 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
*/
lock.writeLock().lock()
try {
- logger.info(s"Changing quota for clientId $clientId to ${quota.bound()}")
-
- if (quota.equals(defaultQuota))
- this.overriddenQuota.remove(clientId)
- else
- this.overriddenQuota.put(clientId, quota)
-
- // Change the underlying metric config if the sensor has been created.
- // Note the metric could be expired by another thread, so use a local variable and null check.
- val metric = metrics.metrics.get(clientRateMetricName(clientId))
- if (metric != null) {
- logger.info(s"Sensor for clientId $clientId already exists. Changing quota to ${quota.bound()} in MetricConfig")
- metric.config(getQuotaMetricConfig(quota))
+ val quotaId = QuotaId(sanitizedUser, clientId)
+ val userInfo = sanitizedUser match {
+ case Some(ConfigEntityName.Default) => "default user "
+ case Some(user) => "user " + user + " "
+ case None => ""
+ }
+ val clientIdInfo = clientId match {
+ case Some(ConfigEntityName.Default) => "default client-id"
+ case Some(id) => "client-id " + id
+ case None => ""
+ }
+ quota match {
+ case Some(newQuota) =>
+ logger.info(s"Changing ${apiKey} quota for ${userInfo}${clientIdInfo} to ${newQuota.bound}")
+ overriddenQuota.put(quotaId, newQuota)
+ (sanitizedUser, clientId) match {
+ case (Some(u), Some(c)) => quotaTypesEnabled |= QuotaTypes.UserClientIdQuotaEnabled
+ case (Some(u), None) => quotaTypesEnabled |= QuotaTypes.UserQuotaEnabled
+ case (None, Some(c)) => quotaTypesEnabled |= QuotaTypes.ClientIdQuotaEnabled
+ case (None, None) =>
+ }
+ case None =>
+ logger.info(s"Removing ${apiKey} quota for ${userInfo}${clientIdInfo}")
+ overriddenQuota.remove(quotaId)
}
+
+ val quotaMetricName = clientRateMetricName(sanitizedUser.getOrElse(""), clientId.getOrElse(""))
+ val allMetrics = metrics.metrics()
+
+ // If multiple-levels of quotas are defined or if this is a default quota update, traverse metrics
+ // to find all affected values. Otherwise, update just the single matching one.
+ val singleUpdate = quotaTypesEnabled match {
+ case QuotaTypes.NoQuotas | QuotaTypes.ClientIdQuotaEnabled | QuotaTypes.UserQuotaEnabled | QuotaTypes.UserClientIdQuotaEnabled =>
+ !sanitizedUser.filter(_ == ConfigEntityName.Default).isDefined && !clientId.filter(_ == ConfigEntityName.Default).isDefined
+ case _ => false
+ }
+ if (singleUpdate) {
+ // Change the underlying metric config if the sensor has been created
+ val metric = allMetrics.get(quotaMetricName)
+ if (metric != null) {
+ val metricConfigEntity = quotaEntity(sanitizedUser.getOrElse(""), clientId.getOrElse(""))
+ val newQuota = metricConfigEntity.quota
+ logger.info(s"Sensor for ${userInfo}${clientIdInfo} already exists. Changing quota to ${newQuota.bound()} in MetricConfig")
+ metric.config(getQuotaMetricConfig(newQuota))
+ }
+ } else {
+ allMetrics.filterKeys(n => n.name == quotaMetricName.name && n.group == quotaMetricName.group).foreach {
+ case (metricName, metric) =>
+ val userTag = if (metricName.tags.containsKey("user")) metricName.tags.get("user") else ""
+ val clientIdTag = if (metricName.tags.containsKey("client-id")) metricName.tags.get("client-id") else ""
+ val metricConfigEntity = quotaEntity(userTag, clientIdTag)
+ if (metricConfigEntity.quota != metric.config.quota) {
+ val newQuota = metricConfigEntity.quota
+ logger.info(s"Sensor for quota-id ${metricConfigEntity.quotaId} already exists. Setting quota to ${newQuota.bound} in MetricConfig")
+ metric.config(getQuotaMetricConfig(newQuota))
+ }
+ }
+ }
+
} finally {
lock.writeLock().unlock()
}
}
- private def clientRateMetricName(clientId: String): MetricName = {
+ private def clientRateMetricName(sanitizedUser: String, clientId: String): MetricName = {
metrics.metricName("byte-rate", apiKey.toString,
- "Tracking byte-rate per client",
+ "Tracking byte-rate per user/client-id",
+ "user", sanitizedUser,
"client-id", clientId)
}
+ private def throttleMetricName(quotaEntity: QuotaEntity): MetricName = {
+ metrics.metricName("throttle-time",
+ apiKey.toString,
+ "Tracking average throttle-time per user/client-id",
+ "user", quotaEntity.sanitizedUser,
+ "client-id", quotaEntity.clientId)
+ }
+
def shutdown() = {
throttledRequestReaper.shutdown()
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/69356fbc/core/src/main/scala/kafka/server/ConfigHandler.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/ConfigHandler.scala b/core/src/main/scala/kafka/server/ConfigHandler.scala
index 67b74a7..5be9c12 100644
--- a/core/src/main/scala/kafka/server/ConfigHandler.scala
+++ b/core/src/main/scala/kafka/server/ConfigHandler.scala
@@ -96,31 +96,59 @@ class TopicConfigHandler(private val logManager: LogManager, kafkaConfig: KafkaC
}
}
-object ClientConfigOverride {
+object QuotaConfigOverride {
val ProducerOverride = "producer_byte_rate"
val ConsumerOverride = "consumer_byte_rate"
}
/**
- * The ClientIdConfigHandler will process clientId config changes in ZK.
- * The callback provides the clientId and the full properties set read from ZK.
- * This implementation reports the overrides to the respective ClientQuotaManager objects
- */
-class ClientIdConfigHandler(private val quotaManagers: QuotaManagers) extends ConfigHandler {
+ * Handles <client-id>, <user> or <user, client-id> quota config updates in ZK.
+ * This implementation reports the overrides to the respective ClientQuotaManager objects
+ */
+class QuotaConfigHandler(private val quotaManagers: QuotaManagers) {
+
+ def updateQuotaConfig(sanitizedUser: Option[String], clientId: Option[String], config: Properties) {
+ val producerQuota =
+ if (config.containsKey(QuotaConfigOverride.ProducerOverride))
+ Some(new Quota(config.getProperty(QuotaConfigOverride.ProducerOverride).toLong, true))
+ else
+ None
+ quotaManagers.produce.updateQuota(sanitizedUser, clientId, producerQuota)
+ val consumerQuota =
+ if (config.containsKey(QuotaConfigOverride.ConsumerOverride))
+ Some(new Quota(config.getProperty(QuotaConfigOverride.ConsumerOverride).toLong, true))
+ else
+ None
+ quotaManagers.fetch.updateQuota(sanitizedUser, clientId, consumerQuota)
+ }
+}
+
+/**
+ * The ClientIdConfigHandler will process clientId config changes in ZK.
+ * The callback provides the clientId and the full properties set read from ZK.
+ */
+class ClientIdConfigHandler(private val quotaManagers: QuotaManagers) extends QuotaConfigHandler(quotaManagers) with ConfigHandler {
+
def processConfigChanges(clientId: String, clientConfig: Properties) {
- if (clientConfig.containsKey(ClientConfigOverride.ProducerOverride)) {
- quotaManagers.produce.updateQuota(clientId,
- new Quota(clientConfig.getProperty(ClientConfigOverride.ProducerOverride).toLong, true))
- } else {
- quotaManagers.fetch.resetQuota(clientId)
- }
+ updateQuotaConfig(None, Some(clientId), clientConfig)
+ }
+}
- if (clientConfig.containsKey(ClientConfigOverride.ConsumerOverride)) {
- quotaManagers.fetch.updateQuota(clientId,
- new Quota(clientConfig.getProperty(ClientConfigOverride.ConsumerOverride).toLong, true))
- } else {
- quotaManagers.produce.resetQuota(clientId)
- }
+/**
+ * The UserConfigHandler will process <user> and <user, client-id> quota changes in ZK.
+ * The callback provides the node name containing sanitized user principal, client-id if this is
+ * a <user, client-id> update and the full properties set read from ZK.
+ */
+class UserConfigHandler(private val quotaManagers: QuotaManagers) extends QuotaConfigHandler(quotaManagers) with ConfigHandler {
+
+ def processConfigChanges(quotaEntityPath: String, config: Properties) {
+ // Entity path is <user> or <user>/clients/<client>
+ val entities = quotaEntityPath.split("/")
+ if (entities.length != 1 && entities.length != 3)
+ throw new IllegalArgumentException("Invalid quota entity path: " + quotaEntityPath);
+ val sanitizedUser = entities(0)
+ val clientId = if (entities.length == 3) Some(entities(2)) else None
+ updateQuotaConfig(Some(sanitizedUser), clientId, config)
}
}
@@ -151,4 +179,4 @@ object ThrottledReplicaValidator extends Validator {
private def isValid(proposed: String): Boolean = {
proposed.trim.equals("*") || proposed.trim.matches("([0-9]+:[0-9]+)?(,[0-9]+:[0-9]+)*")
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/69356fbc/core/src/main/scala/kafka/server/DynamicConfigManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/DynamicConfigManager.scala b/core/src/main/scala/kafka/server/DynamicConfigManager.scala
index 556534a..b31d838 100644
--- a/core/src/main/scala/kafka/server/DynamicConfigManager.scala
+++ b/core/src/main/scala/kafka/server/DynamicConfigManager.scala
@@ -37,8 +37,13 @@ import org.I0Itec.zkclient.{IZkStateListener, IZkChildListener, ZkClient}
object ConfigType {
val Topic = "topics"
val Client = "clients"
+ val User = "users"
val Broker = "brokers"
- val all = Seq(Topic, Client, Broker)
+ val all = Seq(Topic, Client, User, Broker)
+}
+
+object ConfigEntityName {
+ val Default = "<default>"
}
/**
@@ -48,7 +53,9 @@ object ConfigType {
*
* Config is stored under the path: /config/entityType/entityName
* E.g. /config/topics/<topic_name> and /config/clients/<clientId>
- * This znode stores the overrides for this entity (but no defaults) in properties format.
+ * This znode stores the overrides for this entity in properties format with defaults stored using entityName "<default>".
+ * Multiple entity names may be specified (eg. <user, client-id> quotas) using a hierarchical path:
+ * E.g. /config/users/<user>/clients/<clientId>
*
* To avoid watching all topics for changes instead we have a notification path
* /config/changes
@@ -57,8 +64,10 @@ object ConfigType {
* To update a config we first update the config properties. Then we create a new sequential
* znode under the change path which contains the name of the entityType and entityName that was updated, say
* /config/changes/config_change_13321
- * The sequential znode contains data in this format: {"version" : 1, "entityType":"topic/client", "entityName" : "topic_name/client_id"}
+ * The sequential znode contains data in this format: {"version" : 1, "entity_type":"topic/client", "entity_name" : "topic_name/client_id"}
* This is just a notification--the actual config change is stored only once under the /config/entityType/entityName path.
+ * Version 2 of notifications has the format: {"version" : 2, "entity_path":"entity_type/entity_name"}
+ * Multiple entities may be specified as a hierarchical path (eg. users/<user>/clients/<clientId>).
*
* This will fire a watcher on all brokers. This watcher works as follows. It reads all the config change notifications.
* It keeps track of the highest config change suffix number it has applied previously. For any previously applied change it finds
@@ -89,30 +98,60 @@ class DynamicConfigManager(private val zkUtils: ZkUtils,
case Some(mapAnon: Map[_, _]) =>
val map = mapAnon collect
{ case (k: String, v: Any) => k -> v }
- require(map("version") == 1)
-
- val entityType = map.get("entity_type") match {
- case Some(ConfigType.Topic) => ConfigType.Topic
- case Some(ConfigType.Client) => ConfigType.Client
- case Some(ConfigType.Broker) => ConfigType.Broker
- case _ => throw new IllegalArgumentException(s"Config change notification must have 'entity_type' set to one of ${ConfigType.all}. Received: $json")
- }
- val entity = map.get("entity_name") match {
- case Some(value: String) => value
- case _ => throw new IllegalArgumentException("Config change notification does not specify 'entity_name'. Received: " + json)
+ map("version") match {
+ case 1 => processEntityConfigChangeVersion1(json, map)
+ case 2 => processEntityConfigChangeVersion2(json, map)
+ case _ => throw new IllegalArgumentException("Config change notification has an unsupported version " + map("version") +
+ "Supported versions are 1 and 2.")
}
- val entityConfig = AdminUtils.fetchEntityConfig(zkUtils, entityType, entity)
- logger.info(s"Processing override for entityType: $entityType, entity: $entity with config: $entityConfig")
- configHandlers(entityType).processConfigChanges(entity, entityConfig)
case o => throw new IllegalArgumentException("Config change notification has an unexpected value. The format is:" +
- "{\"version\" : 1," +
- " \"entity_type\":\"topic/client\"," +
- " \"entity_name\" : \"topic_name/client_id\"}." +
+ "{\"version\" : 1, \"entity_type\":\"topics/clients\", \"entity_name\" : \"topic_name/client_id\"}." + " or " +
+ "{\"version\" : 2, \"entity_path\":\"entity_type/entity_name\"}." +
" Received: " + json)
}
}
+
+ private def processEntityConfigChangeVersion1(json: String, map: Map[String, Any]) {
+
+ val entityType = map.get("entity_type") match {
+ case Some(ConfigType.Topic) => ConfigType.Topic
+ case Some(ConfigType.Client) => ConfigType.Client
+ case _ => throw new IllegalArgumentException("Version 1 config change notification must have 'entity_type' set to 'clients' or 'topics'." +
+ " Received: " + json)
+ }
+
+ val entity = map.get("entity_name") match {
+ case Some(value: String) => value
+ case _ => throw new IllegalArgumentException("Version 1 config change notification does not specify 'entity_name'. Received: " + json)
+ }
+
+ val entityConfig = AdminUtils.fetchEntityConfig(zkUtils, entityType, entity)
+ logger.info(s"Processing override for entityType: $entityType, entity: $entity with config: $entityConfig")
+ configHandlers(entityType).processConfigChanges(entity, entityConfig)
+
+ }
+
+ private def processEntityConfigChangeVersion2(json: String, map: Map[String, Any]) {
+
+ val entityPath = map.get("entity_path") match {
+ case Some(value: String) => value
+ case _ => throw new IllegalArgumentException("Version 2 config change notification does not specify 'entity_path'. Received: " + json)
+ }
+
+ val index = entityPath.indexOf('/')
+ val rootEntityType = entityPath.substring(0, index)
+ if (index < 0 || !configHandlers.contains(rootEntityType))
+ throw new IllegalArgumentException("Version 2 config change notification must have 'entity_path' starting with 'clients/', 'topics/' or 'users/'." +
+ " Received: " + json)
+ val fullSanitizedEntityName = entityPath.substring(index + 1)
+
+ val entityConfig = AdminUtils.fetchEntityConfig(zkUtils, rootEntityType, fullSanitizedEntityName)
+ logger.info(s"Processing override for entityPath: $entityPath with config: $entityConfig")
+ configHandlers(rootEntityType).processConfigChanges(fullSanitizedEntityName, entityConfig)
+
+ }
}
private val configChangeListener = new ZkNodeChangeNotificationListener(zkUtils, ZkUtils.EntityConfigChangesPath, AdminUtils.EntityConfigChangeZnodePrefix, ConfigChangedNotificationHandler)
@@ -122,5 +161,20 @@ class DynamicConfigManager(private val zkUtils: ZkUtils,
*/
def startup(): Unit = {
configChangeListener.init()
+
+ // Apply all existing client/user configs to the ClientIdConfigHandler/UserConfigHandler to bootstrap the overrides
+ configHandlers.foreach {
+ case (ConfigType.User, handler) =>
+ AdminUtils.fetchAllEntityConfigs(zkUtils, ConfigType.User).foreach {
+ case (sanitizedUser, properties) => handler.processConfigChanges(sanitizedUser, properties)
+ }
+ AdminUtils.fetchAllChildEntityConfigs(zkUtils, ConfigType.User, ConfigType.Client).foreach {
+ case (sanitizedUserClientId, properties) => handler.processConfigChanges(sanitizedUserClientId, properties)
+ }
+ case (configType, handler) =>
+ AdminUtils.fetchAllEntityConfigs(zkUtils, configType).foreach {
+ case (entityName, properties) => handler.processConfigChanges(entityName, properties)
+ }
+ }
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/69356fbc/core/src/main/scala/kafka/server/KafkaApis.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala
index 3008426..d3ba5ef 100644
--- a/core/src/main/scala/kafka/server/KafkaApis.scala
+++ b/core/src/main/scala/kafka/server/KafkaApis.scala
@@ -396,6 +396,7 @@ class KafkaApis(val requestChannel: RequestChannel,
request.apiRemoteCompleteTimeMs = SystemTime.milliseconds
quotas.produce.recordAndMaybeThrottle(
+ request.session.sanitizedUser,
request.header.clientId,
numBytesAppended,
produceResponseCallback)
@@ -494,7 +495,7 @@ class KafkaApis(val requestChannel: RequestChannel,
fetchResponseCallback(0)
} else {
val size = FetchResponse.responseSize(mergedPartitionData.groupBy(_._1.topic), fetchRequest.versionId)
- quotas.fetch.recordAndMaybeThrottle(fetchRequest.clientId, size, fetchResponseCallback)
+ quotas.fetch.recordAndMaybeThrottle(request.session.sanitizedUser, fetchRequest.clientId, size, fetchResponseCallback)
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/69356fbc/core/src/main/scala/kafka/server/KafkaConfig.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index 3671297..b37be5b 100755
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -523,8 +523,10 @@ object KafkaConfig {
"or this timeout is reached. This is similar to the producer request timeout."
val OffsetCommitRequiredAcksDoc = "The required acks before the commit can be accepted. In general, the default (-1) should not be overridden"
/** ********* Quota Configuration ***********/
- val ProducerQuotaBytesPerSecondDefaultDoc = "Any producer distinguished by clientId will get throttled if it produces more bytes than this value per-second"
- val ConsumerQuotaBytesPerSecondDefaultDoc = "Any consumer distinguished by clientId/consumer group will get throttled if it fetches more bytes than this value per-second"
+ val ProducerQuotaBytesPerSecondDefaultDoc = "DEPRECATED: Used only when dynamic default quotas are not configured for <user>, <client-id> or <user, client-id> in Zookeeper. " +
+ "Any producer distinguished by clientId will get throttled if it produces more bytes than this value per-second"
+ val ConsumerQuotaBytesPerSecondDefaultDoc = "DEPRECATED: Used only when dynamic default quotas are not configured for <user, <client-id> or <user, client-id> in Zookeeper. " +
+ "Any consumer distinguished by clientId/consumer group will get throttled if it fetches more bytes than this value per-second"
val NumQuotaSamplesDoc = "The number of samples to retain in memory for client quotas"
val NumReplicationQuotaSamplesDoc = "The number of samples to retain in memory for replication quotas"
val QuotaWindowSizeSecondsDoc = "The time span of each sample for client quotas"
http://git-wip-us.apache.org/repos/asf/kafka/blob/69356fbc/core/src/main/scala/kafka/server/KafkaServer.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala
index db92cb8..5055c87 100755
--- a/core/src/main/scala/kafka/server/KafkaServer.scala
+++ b/core/src/main/scala/kafka/server/KafkaServer.scala
@@ -251,14 +251,9 @@ class KafkaServer(val config: KafkaConfig, time: Time = SystemTime, threadNamePr
/* start dynamic config manager */
dynamicConfigHandlers = Map[String, ConfigHandler](ConfigType.Topic -> new TopicConfigHandler(logManager, config, quotaManagers),
ConfigType.Client -> new ClientIdConfigHandler(quotaManagers),
+ ConfigType.User -> new UserConfigHandler(quotaManagers),
ConfigType.Broker -> new BrokerConfigHandler(config, quotaManagers))
- // Apply all existing client configs to the ClientIdConfigHandler to bootstrap the overrides
- // TODO: Move this logic to DynamicConfigManager
- AdminUtils.fetchAllEntityConfigs(zkUtils, ConfigType.Client).foreach {
- case (clientId, properties) => dynamicConfigHandlers(ConfigType.Client).processConfigChanges(clientId, properties)
- }
-
// Create the config manager. start listening to notifications
dynamicConfigManager = new DynamicConfigManager(zkUtils, dynamicConfigHandlers)
dynamicConfigManager.startup()
http://git-wip-us.apache.org/repos/asf/kafka/blob/69356fbc/core/src/main/scala/kafka/utils/ZkUtils.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/utils/ZkUtils.scala b/core/src/main/scala/kafka/utils/ZkUtils.scala
index 503ed54..96779ff 100644
--- a/core/src/main/scala/kafka/utils/ZkUtils.scala
+++ b/core/src/main/scala/kafka/utils/ZkUtils.scala
@@ -119,6 +119,9 @@ object ZkUtils {
def getEntityConfigPath(entityType: String, entity: String): String =
getEntityConfigRootPath(entityType) + "/" + entity
+ def getEntityConfigPath(entityPath: String): String =
+ ZkUtils.EntityConfigPath + "/" + entityPath
+
def getDeleteTopicPath(topic: String): String =
DeleteTopicsPath + "/" + topic
http://git-wip-us.apache.org/repos/asf/kafka/blob/69356fbc/core/src/test/scala/integration/kafka/api/BaseQuotaTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/integration/kafka/api/BaseQuotaTest.scala b/core/src/test/scala/integration/kafka/api/BaseQuotaTest.scala
new file mode 100644
index 0000000..c9b7787
--- /dev/null
+++ b/core/src/test/scala/integration/kafka/api/BaseQuotaTest.scala
@@ -0,0 +1,195 @@
+/**
+ * Licensed 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 kafka.api
+
+import java.util.Properties
+
+import kafka.server.{QuotaConfigOverride, KafkaConfig, KafkaServer, QuotaId}
+import kafka.utils.TestUtils
+import org.apache.kafka.clients.consumer.{ConsumerConfig, KafkaConsumer}
+import org.apache.kafka.clients.producer._
+import org.apache.kafka.clients.producer.internals.ErrorLoggingCallback
+import org.apache.kafka.common.MetricName
+import org.apache.kafka.common.metrics.{Quota, KafkaMetric}
+import org.apache.kafka.common.protocol.ApiKeys
+import org.junit.Assert._
+import org.junit.{After, Before, Test}
+
+import scala.collection.JavaConversions._
+import scala.collection.JavaConverters._
+import scala.collection.Map
+import scala.collection.mutable
+
+abstract class BaseQuotaTest extends IntegrationTestHarness {
+
+ def userPrincipal : String
+ def producerQuotaId : QuotaId
+ def consumerQuotaId : QuotaId
+ def overrideQuotas(producerQuota: Long, consumerQuota: Long)
+ def removeQuotaOverrides()
+
+ override val serverCount = 2
+ val producerCount = 1
+ val consumerCount = 1
+
+ private val producerBufferSize = 300000
+ protected val producerClientId = "QuotasTestProducer-1"
+ protected val consumerClientId = "QuotasTestConsumer-1"
+
+ this.serverConfig.setProperty(KafkaConfig.ControlledShutdownEnableProp, "false")
+ this.serverConfig.setProperty(KafkaConfig.OffsetsTopicReplicationFactorProp, "2")
+ this.serverConfig.setProperty(KafkaConfig.OffsetsTopicPartitionsProp, "1")
+ this.serverConfig.setProperty(KafkaConfig.GroupMinSessionTimeoutMsProp, "100")
+ this.serverConfig.setProperty(KafkaConfig.GroupMaxSessionTimeoutMsProp, "30000")
+ this.producerConfig.setProperty(ProducerConfig.ACKS_CONFIG, "0")
+ this.producerConfig.setProperty(ProducerConfig.BUFFER_MEMORY_CONFIG, producerBufferSize.toString)
+ this.producerConfig.setProperty(ProducerConfig.CLIENT_ID_CONFIG, producerClientId)
+ this.consumerConfig.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "QuotasTest")
+ this.consumerConfig.setProperty(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 4096.toString)
+ this.consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
+ this.consumerConfig.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, consumerClientId)
+
+ // Low enough quota that a producer sending a small payload in a tight loop should get throttled
+ val defaultProducerQuota = 8000
+ val defaultConsumerQuota = 2500
+
+ var leaderNode: KafkaServer = null
+ var followerNode: KafkaServer = null
+ private val topic1 = "topic-1"
+
+ @Before
+ override def setUp() {
+ super.setUp()
+
+ val numPartitions = 1
+ val leaders = TestUtils.createTopic(zkUtils, topic1, numPartitions, serverCount, servers)
+ leaderNode = if (leaders(0).get == servers.head.config.brokerId) servers.head else servers(1)
+ followerNode = if (leaders(0).get != servers.head.config.brokerId) servers.head else servers(1)
+ assertTrue("Leader of all partitions of the topic should exist", leaders.values.forall(leader => leader.isDefined))
+ }
+
+ @After
+ override def tearDown() {
+ super.tearDown()
+ }
+
+ @Test
+ def testThrottledProducerConsumer() {
+ val allMetrics: mutable.Map[MetricName, KafkaMetric] = leaderNode.metrics.metrics().asScala
+
+ val numRecords = 1000
+ produce(producers.head, numRecords)
+
+ val producerMetricName = throttleMetricName(ApiKeys.PRODUCE, producerQuotaId)
+ assertTrue("Should have been throttled", allMetrics(producerMetricName).value() > 0)
+
+ // Consumer should read in a bursty manner and get throttled immediately
+ consume(consumers.head, numRecords)
+ val consumerMetricName = throttleMetricName(ApiKeys.FETCH, consumerQuotaId)
+ assertTrue("Should have been throttled", allMetrics(consumerMetricName).value() > 0)
+ }
+
+ @Test
+ def testProducerConsumerOverrideUnthrottled() {
+ // Give effectively unlimited quota for producer and consumer
+ val props = new Properties()
+ props.put(QuotaConfigOverride.ProducerOverride, Long.MaxValue.toString)
+ props.put(QuotaConfigOverride.ConsumerOverride, Long.MaxValue.toString)
+
+ overrideQuotas(Long.MaxValue, Long.MaxValue)
+ waitForQuotaUpdate(Long.MaxValue, Long.MaxValue)
+
+ val allMetrics: mutable.Map[MetricName, KafkaMetric] = leaderNode.metrics.metrics().asScala
+ val numRecords = 1000
+ produce(producers.head, numRecords)
+ val producerMetricName = throttleMetricName(ApiKeys.PRODUCE, producerQuotaId)
+ assertEquals("Should not have been throttled", 0.0, allMetrics(producerMetricName).value(), 0.0)
+
+ // The "client" consumer does not get throttled.
+ consume(consumers.head, numRecords)
+ val consumerMetricName = throttleMetricName(ApiKeys.FETCH, consumerQuotaId)
+ assertEquals("Should not have been throttled", 0.0, allMetrics(consumerMetricName).value(), 0.0)
+ }
+
+ @Test
+ def testQuotaOverrideDelete() {
+ // Override producer and consumer quotas to unlimited
+ overrideQuotas(Long.MaxValue, Long.MaxValue)
+
+ val allMetrics: mutable.Map[MetricName, KafkaMetric] = leaderNode.metrics.metrics().asScala
+ val numRecords = 1000
+ produce(producers.head, numRecords)
+ assertTrue("Should not have been throttled", allMetrics(throttleMetricName(ApiKeys.PRODUCE, producerQuotaId)).value() == 0)
+ consume(consumers.head, numRecords)
+ assertTrue("Should not have been throttled", allMetrics(throttleMetricName(ApiKeys.FETCH, consumerQuotaId)).value() == 0)
+
+ // Delete producer and consumer quota overrides. Consumer and producer should now be
+ // throttled since broker defaults are very small
+ removeQuotaOverrides()
+ produce(producers.head, numRecords)
+
+ assertTrue("Should have been throttled", allMetrics(throttleMetricName(ApiKeys.PRODUCE, producerQuotaId)).value() > 0)
+ consume(consumers.head, numRecords)
+ assertTrue("Should have been throttled", allMetrics(throttleMetricName(ApiKeys.FETCH, consumerQuotaId)).value() > 0)
+ }
+
+ def produce(p: KafkaProducer[Array[Byte], Array[Byte]], count: Int): Int = {
+ var numBytesProduced = 0
+ for (i <- 0 to count) {
+ val payload = i.toString.getBytes
+ numBytesProduced += payload.length
+ p.send(new ProducerRecord[Array[Byte], Array[Byte]](topic1, null, null, payload),
+ new ErrorLoggingCallback(topic1, null, null, true)).get()
+ Thread.sleep(1)
+ }
+ numBytesProduced
+ }
+
+ def consume(consumer: KafkaConsumer[Array[Byte], Array[Byte]], numRecords: Int) {
+ consumer.subscribe(List(topic1))
+ var numConsumed = 0
+ while (numConsumed < numRecords) {
+ for (cr <- consumer.poll(100)) {
+ numConsumed += 1
+ }
+ }
+ }
+
+ def waitForQuotaUpdate(producerQuota: Long, consumerQuota: Long) {
+ TestUtils.retry(10000) {
+ val quotaManagers = leaderNode.apis.quotas
+ val overrideProducerQuota = quotaManagers.produce.quota(userPrincipal, producerClientId)
+ val overrideConsumerQuota = quotaManagers.fetch.quota(userPrincipal, consumerClientId)
+
+ assertEquals(s"ClientId $producerClientId of user $userPrincipal must have producer quota", Quota.upperBound(producerQuota), overrideProducerQuota)
+ assertEquals(s"ClientId $consumerClientId of user $userPrincipal must have consumer quota", Quota.upperBound(consumerQuota), overrideConsumerQuota)
+ }
+ }
+
+ private def throttleMetricName(apiKey: ApiKeys, quotaId: QuotaId): MetricName = {
+ leaderNode.metrics.metricName("throttle-time",
+ apiKey.name,
+ "Tracking throttle-time per user/client-id",
+ "user", quotaId.sanitizedUser.getOrElse(""),
+ "client-id", quotaId.clientId.getOrElse(""))
+ }
+
+ def quotaProperties(producerQuota: Long, consumerQuota: Long): Properties = {
+ val props = new Properties()
+ props.put(QuotaConfigOverride.ProducerOverride, producerQuota.toString)
+ props.put(QuotaConfigOverride.ConsumerOverride, consumerQuota.toString)
+ props
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/69356fbc/core/src/test/scala/integration/kafka/api/ClientIdQuotaTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/integration/kafka/api/ClientIdQuotaTest.scala b/core/src/test/scala/integration/kafka/api/ClientIdQuotaTest.scala
new file mode 100644
index 0000000..7477f7f
--- /dev/null
+++ b/core/src/test/scala/integration/kafka/api/ClientIdQuotaTest.scala
@@ -0,0 +1,55 @@
+/**
+ * Licensed 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 kafka.api
+
+import java.util.Properties
+
+import kafka.admin.AdminUtils
+import kafka.server.{KafkaConfig, QuotaConfigOverride, QuotaId}
+import org.apache.kafka.common.security.auth.KafkaPrincipal
+import org.junit.Before
+
+class ClientIdQuotaTest extends BaseQuotaTest {
+
+ override val userPrincipal = KafkaPrincipal.ANONYMOUS.getName
+ override val producerQuotaId = QuotaId(None, Some(producerClientId))
+ override val consumerQuotaId = QuotaId(None, Some(consumerClientId))
+
+ @Before
+ override def setUp() {
+ this.serverConfig.setProperty(KafkaConfig.ProducerQuotaBytesPerSecondDefaultProp, defaultProducerQuota.toString)
+ this.serverConfig.setProperty(KafkaConfig.ConsumerQuotaBytesPerSecondDefaultProp, defaultConsumerQuota.toString)
+ super.setUp()
+ }
+
+ override def overrideQuotas(producerQuota: Long, consumerQuota: Long) {
+ val producerProps = new Properties()
+ producerProps.put(QuotaConfigOverride.ProducerOverride, producerQuota.toString)
+ updateQuotaOverride(producerClientId, producerProps)
+
+ val consumerProps = new Properties()
+ consumerProps.put(QuotaConfigOverride.ConsumerOverride, consumerQuota.toString)
+ updateQuotaOverride(consumerClientId, consumerProps)
+ }
+ override def removeQuotaOverrides() {
+ val emptyProps = new Properties
+ updateQuotaOverride(producerClientId, emptyProps)
+ updateQuotaOverride(consumerClientId, emptyProps)
+ }
+
+ private def updateQuotaOverride(clientId: String, properties: Properties) {
+ AdminUtils.changeClientIdConfig(zkUtils, clientId, properties)
+ }
+}