You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mi...@apache.org on 2024/01/18 15:15:20 UTC
(kafka) branch trunk updated: KAFKA-15853: Move ClientQuotaManagerConfig outside of core (#15159)
This is an automated email from the ASF dual-hosted git repository.
mimaison pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 2f2a0d799ab KAFKA-15853: Move ClientQuotaManagerConfig outside of core (#15159)
2f2a0d799ab is described below
commit 2f2a0d799abb22ffdf3c76035261ac6589b7e7f2
Author: Omnia Ibrahim <o....@gmail.com>
AuthorDate: Thu Jan 18 15:15:13 2024 +0000
KAFKA-15853: Move ClientQuotaManagerConfig outside of core (#15159)
Reviewers: Mickael Maison <mi...@gmail.com>, Nikolay <NI...@gmail.com>
---
.../scala/kafka/server/ClientQuotaManager.scala | 19 +--------
.../kafka/server/ClientRequestQuotaManager.scala | 1 +
.../server/ControllerMutationQuotaManager.scala | 1 +
core/src/main/scala/kafka/server/KafkaConfig.scala | 10 ++---
.../src/main/scala/kafka/server/QuotaFactory.scala | 14 +++----
.../integration/kafka/api/BaseQuotaTest.scala | 5 ++-
.../unit/kafka/server/ClientQuotaManagerTest.scala | 20 +++++-----
.../server/ClientRequestQuotaManagerTest.scala | 4 +-
.../ControllerMutationQuotaManagerTest.scala | 7 +---
.../server/ThrottledChannelExpirationTest.scala | 4 +-
.../server/config/ClientQuotaManagerConfig.java | 45 ++++++++++++++++++++++
11 files changed, 78 insertions(+), 52 deletions(-)
diff --git a/core/src/main/scala/kafka/server/ClientQuotaManager.scala b/core/src/main/scala/kafka/server/ClientQuotaManager.scala
index 299525cf459..d9cc1c971e7 100644
--- a/core/src/main/scala/kafka/server/ClientQuotaManager.scala
+++ b/core/src/main/scala/kafka/server/ClientQuotaManager.scala
@@ -30,7 +30,7 @@ import org.apache.kafka.common.metrics.Metrics
import org.apache.kafka.common.metrics.stats.{Avg, CumulativeSum, Rate}
import org.apache.kafka.common.security.auth.KafkaPrincipal
import org.apache.kafka.common.utils.{Sanitizer, Time}
-import org.apache.kafka.server.config.ConfigEntityName
+import org.apache.kafka.server.config.{ConfigEntityName, ClientQuotaManagerConfig}
import org.apache.kafka.server.quota.{ClientQuotaCallback, ClientQuotaEntity, ClientQuotaType}
import org.apache.kafka.server.util.ShutdownableThread
@@ -44,23 +44,6 @@ import scala.jdk.CollectionConverters._
*/
case class ClientSensors(metricTags: Map[String, String], quotaSensor: Sensor, throttleTimeSensor: Sensor)
-/**
- * Configuration settings for quota management
- * @param numQuotaSamples The number of samples to retain in memory
- * @param quotaWindowSizeSeconds The time span of each sample
- *
- */
-case class ClientQuotaManagerConfig(numQuotaSamples: Int =
- ClientQuotaManagerConfig.DefaultNumQuotaSamples,
- quotaWindowSizeSeconds: Int =
- ClientQuotaManagerConfig.DefaultQuotaWindowSizeSeconds)
-
-object ClientQuotaManagerConfig {
- // Always have 10 whole windows + 1 current window
- val DefaultNumQuotaSamples = 11
- val DefaultQuotaWindowSizeSeconds = 1
-}
-
object QuotaTypes {
val NoQuotas = 0
val ClientIdQuotaEnabled = 1
diff --git a/core/src/main/scala/kafka/server/ClientRequestQuotaManager.scala b/core/src/main/scala/kafka/server/ClientRequestQuotaManager.scala
index d330210b9d5..9885565502a 100644
--- a/core/src/main/scala/kafka/server/ClientRequestQuotaManager.scala
+++ b/core/src/main/scala/kafka/server/ClientRequestQuotaManager.scala
@@ -23,6 +23,7 @@ import org.apache.kafka.common.MetricName
import org.apache.kafka.common.metrics._
import org.apache.kafka.common.metrics.stats.Rate
import org.apache.kafka.common.utils.Time
+import org.apache.kafka.server.config.ClientQuotaManagerConfig
import org.apache.kafka.server.quota.ClientQuotaCallback
import scala.jdk.CollectionConverters._
diff --git a/core/src/main/scala/kafka/server/ControllerMutationQuotaManager.scala b/core/src/main/scala/kafka/server/ControllerMutationQuotaManager.scala
index e21c4699bcd..b12d5aa4595 100644
--- a/core/src/main/scala/kafka/server/ControllerMutationQuotaManager.scala
+++ b/core/src/main/scala/kafka/server/ControllerMutationQuotaManager.scala
@@ -28,6 +28,7 @@ import org.apache.kafka.common.metrics.stats.TokenBucket
import org.apache.kafka.common.protocol.Errors
import org.apache.kafka.common.utils.Time
import org.apache.kafka.server.quota.ClientQuotaCallback
+import org.apache.kafka.server.config.ClientQuotaManagerConfig
import scala.jdk.CollectionConverters._
diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index 3c26b04cf2f..6bbf4968bc5 100755
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -48,7 +48,7 @@ import org.apache.kafka.server.ProcessRole
import org.apache.kafka.server.authorizer.Authorizer
import org.apache.kafka.server.common.{MetadataVersion, MetadataVersionValidator}
import org.apache.kafka.server.common.MetadataVersion._
-import org.apache.kafka.server.config.{ReplicationQuotaManagerConfig, ServerTopicConfigSynonyms}
+import org.apache.kafka.server.config.{ClientQuotaManagerConfig, ReplicationQuotaManagerConfig, ServerTopicConfigSynonyms}
import org.apache.kafka.storage.internals.log.{LogConfig, ProducerStateManagerConfig}
import org.apache.kafka.storage.internals.log.LogConfig.MessageFormatVersion
import org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig
@@ -213,14 +213,14 @@ object Defaults {
val FetchMaxBytes = 55 * 1024 * 1024
/** ********* Quota Configuration ***********/
- val NumQuotaSamples: Int = ClientQuotaManagerConfig.DefaultNumQuotaSamples
- val QuotaWindowSizeSeconds: Int = ClientQuotaManagerConfig.DefaultQuotaWindowSizeSeconds
+ val NumQuotaSamples: Int = ClientQuotaManagerConfig.DEFAULT_NUM_QUOTA_SAMPLES
+ val QuotaWindowSizeSeconds: Int = ClientQuotaManagerConfig.DEFAULT_QUOTA_WINDOW_SIZE_SECONDS
val NumReplicationQuotaSamples: Int = ReplicationQuotaManagerConfig.DEFAULT_NUM_QUOTA_SAMPLES
val ReplicationQuotaWindowSizeSeconds: Int = ReplicationQuotaManagerConfig.DEFAULT_QUOTA_WINDOW_SIZE_SECONDS
val NumAlterLogDirsReplicationQuotaSamples: Int = ReplicationQuotaManagerConfig.DEFAULT_NUM_QUOTA_SAMPLES
val AlterLogDirsReplicationQuotaWindowSizeSeconds: Int = ReplicationQuotaManagerConfig.DEFAULT_QUOTA_WINDOW_SIZE_SECONDS
- val NumControllerQuotaSamples: Int = ClientQuotaManagerConfig.DefaultNumQuotaSamples
- val ControllerQuotaWindowSizeSeconds: Int = ClientQuotaManagerConfig.DefaultQuotaWindowSizeSeconds
+ val NumControllerQuotaSamples: Int = ClientQuotaManagerConfig.DEFAULT_NUM_QUOTA_SAMPLES
+ val ControllerQuotaWindowSizeSeconds: Int = ClientQuotaManagerConfig.DEFAULT_QUOTA_WINDOW_SIZE_SECONDS
/** ********* Kafka Metrics Configuration ***********/
val MetricNumSamples = 2
diff --git a/core/src/main/scala/kafka/server/QuotaFactory.scala b/core/src/main/scala/kafka/server/QuotaFactory.scala
index 0dc4dbe6e1e..2a828f777d3 100644
--- a/core/src/main/scala/kafka/server/QuotaFactory.scala
+++ b/core/src/main/scala/kafka/server/QuotaFactory.scala
@@ -22,7 +22,7 @@ import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.metrics.Metrics
import org.apache.kafka.server.quota.ClientQuotaCallback
import org.apache.kafka.common.utils.Time
-import org.apache.kafka.server.config.ReplicationQuotaManagerConfig
+import org.apache.kafka.server.config.{ClientQuotaManagerConfig, ReplicationQuotaManagerConfig}
import org.apache.kafka.server.quota.ClientQuotaType
object QuotaType {
@@ -90,16 +90,16 @@ object QuotaFactory extends Logging {
}
def clientConfig(cfg: KafkaConfig): ClientQuotaManagerConfig = {
- ClientQuotaManagerConfig(
- numQuotaSamples = cfg.numQuotaSamples,
- quotaWindowSizeSeconds = cfg.quotaWindowSizeSeconds
+ new ClientQuotaManagerConfig(
+ cfg.numQuotaSamples,
+ cfg.quotaWindowSizeSeconds
)
}
def clientControllerMutationConfig(cfg: KafkaConfig): ClientQuotaManagerConfig = {
- ClientQuotaManagerConfig(
- numQuotaSamples = cfg.numControllerQuotaSamples,
- quotaWindowSizeSeconds = cfg.controllerQuotaWindowSizeSeconds
+ new ClientQuotaManagerConfig(
+ cfg.numControllerQuotaSamples,
+ cfg.controllerQuotaWindowSizeSeconds
)
}
diff --git a/core/src/test/scala/integration/kafka/api/BaseQuotaTest.scala b/core/src/test/scala/integration/kafka/api/BaseQuotaTest.scala
index b9a7afe442a..74e6756090c 100644
--- a/core/src/test/scala/integration/kafka/api/BaseQuotaTest.scala
+++ b/core/src/test/scala/integration/kafka/api/BaseQuotaTest.scala
@@ -19,7 +19,7 @@ import java.util.concurrent.TimeUnit
import java.util.{Collections, HashMap, Properties}
import com.yammer.metrics.core.{Histogram, Meter}
import kafka.api.QuotaTestClients._
-import kafka.server.{ClientQuotaManager, ClientQuotaManagerConfig, KafkaBroker, KafkaConfig, QuotaType}
+import kafka.server.{ClientQuotaManager, KafkaBroker, KafkaConfig, QuotaType}
import kafka.utils.{TestInfoUtils, TestUtils}
import org.apache.kafka.clients.admin.Admin
import org.apache.kafka.clients.consumer.{Consumer, ConsumerConfig}
@@ -33,6 +33,7 @@ import org.apache.kafka.common.quota.ClientQuotaAlteration
import org.apache.kafka.common.quota.ClientQuotaEntity
import org.apache.kafka.common.security.auth.KafkaPrincipal
import org.apache.kafka.server.metrics.KafkaYammerMetrics
+import org.apache.kafka.server.config.ClientQuotaManagerConfig
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{BeforeEach, TestInfo}
import org.junit.jupiter.params.ParameterizedTest
@@ -182,7 +183,7 @@ abstract class BaseQuotaTest extends IntegrationTestHarness {
assertTrue(throttled, "Should have been throttled")
quotaTestClients.verifyConsumerClientThrottleTimeMetric(expectThrottle = true,
- Some(ClientQuotaManagerConfig.DefaultQuotaWindowSizeSeconds * 1000.0))
+ Some(ClientQuotaManagerConfig.DEFAULT_QUOTA_WINDOW_SIZE_SECONDS * 1000.0))
val exemptMetric = quotaTestClients.exemptRequestMetric
assertNotNull(exemptMetric, "Exempt requests not recorded")
diff --git a/core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala
index 7fb9d83736a..1dd1823b0f1 100644
--- a/core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala
@@ -17,19 +17,17 @@
package kafka.server
import java.net.InetAddress
-
import kafka.network.RequestChannel.Session
import kafka.server.QuotaType._
import org.apache.kafka.common.metrics.Quota
import org.apache.kafka.common.security.auth.KafkaPrincipal
import org.apache.kafka.common.utils.Sanitizer
-import org.apache.kafka.server.config.ConfigEntityName
-
+import org.apache.kafka.server.config.{ClientQuotaManagerConfig, ConfigEntityName}
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.Test
class ClientQuotaManagerTest extends BaseClientQuotaManagerTest {
- private val config = ClientQuotaManagerConfig()
+ private val config = new ClientQuotaManagerConfig()
private def testQuotaParsing(config: ClientQuotaManagerConfig, client1: UserClient, client2: UserClient, randomClient: UserClient, defaultConfigClient: UserClient): Unit = {
val clientQuotaManager = new ClientQuotaManager(config, metrics, Produce, time, "")
@@ -101,7 +99,7 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest {
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()
+ val config = new ClientQuotaManagerConfig()
testQuotaParsing(config, client1, client2, randomClient, defaultConfigClient)
}
@@ -115,7 +113,7 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest {
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()
+ val config = new ClientQuotaManagerConfig()
testQuotaParsing(config, client1, client2, randomClient, defaultConfigClient)
}
@@ -161,7 +159,7 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest {
@Test
def testGetMaxValueInQuotaWindowWithNonDefaultQuotaWindow(): Unit = {
val numFullQuotaWindows = 3 // 3 seconds window (vs. 10 seconds default)
- val nonDefaultConfig = ClientQuotaManagerConfig(numQuotaSamples = numFullQuotaWindows + 1)
+ val nonDefaultConfig = new ClientQuotaManagerConfig(numFullQuotaWindows + 1)
val clientQuotaManager = new ClientQuotaManager(nonDefaultConfig, metrics, Fetch, time, "")
val userSession = Session(new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "userA"), InetAddress.getLocalHost)
@@ -180,7 +178,7 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest {
@Test
def testSetAndRemoveDefaultUserQuota(): Unit = {
// quotaTypesEnabled will be QuotaTypes.NoQuotas initially
- val clientQuotaManager = new ClientQuotaManager(ClientQuotaManagerConfig(),
+ val clientQuotaManager = new ClientQuotaManager(new ClientQuotaManagerConfig(),
metrics, Produce, time, "")
try {
@@ -202,7 +200,7 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest {
@Test
def testSetAndRemoveUserQuota(): Unit = {
// quotaTypesEnabled will be QuotaTypes.NoQuotas initially
- val clientQuotaManager = new ClientQuotaManager(ClientQuotaManagerConfig(),
+ val clientQuotaManager = new ClientQuotaManager(new ClientQuotaManagerConfig(),
metrics, Produce, time, "")
try {
@@ -221,7 +219,7 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest {
@Test
def testSetAndRemoveUserClientQuota(): Unit = {
// quotaTypesEnabled will be QuotaTypes.NoQuotas initially
- val clientQuotaManager = new ClientQuotaManager(ClientQuotaManagerConfig(),
+ val clientQuotaManager = new ClientQuotaManager(new ClientQuotaManagerConfig(),
metrics, Produce, time, "")
try {
@@ -239,7 +237,7 @@ class ClientQuotaManagerTest extends BaseClientQuotaManagerTest {
@Test
def testQuotaConfigPrecedence(): Unit = {
- val clientQuotaManager = new ClientQuotaManager(ClientQuotaManagerConfig(),
+ val clientQuotaManager = new ClientQuotaManager(new ClientQuotaManagerConfig(),
metrics, Produce, time, "")
try {
diff --git a/core/src/test/scala/unit/kafka/server/ClientRequestQuotaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ClientRequestQuotaManagerTest.scala
index db2dceae193..31a3e6e753d 100644
--- a/core/src/test/scala/unit/kafka/server/ClientRequestQuotaManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ClientRequestQuotaManagerTest.scala
@@ -18,12 +18,12 @@ package kafka.server
import kafka.server.QuotaType.Request
import org.apache.kafka.common.metrics.Quota
-
+import org.apache.kafka.server.config.ClientQuotaManagerConfig
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.Test
class ClientRequestQuotaManagerTest extends BaseClientQuotaManagerTest {
- private val config = ClientQuotaManagerConfig()
+ private val config = new ClientQuotaManagerConfig()
@Test
def testRequestPercentageQuotaViolation(): Unit = {
diff --git a/core/src/test/scala/unit/kafka/server/ControllerMutationQuotaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ControllerMutationQuotaManagerTest.scala
index 5479e15b786..dea6d554f1a 100644
--- a/core/src/test/scala/unit/kafka/server/ControllerMutationQuotaManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ControllerMutationQuotaManagerTest.scala
@@ -17,7 +17,6 @@
package kafka.server
import java.util.concurrent.TimeUnit
-
import kafka.server.QuotaType.ControllerMutation
import org.apache.kafka.common.errors.ThrottlingQuotaExceededException
import org.apache.kafka.common.metrics.MetricConfig
@@ -26,6 +25,7 @@ import org.apache.kafka.common.metrics.Quota
import org.apache.kafka.common.metrics.QuotaViolationException
import org.apache.kafka.common.metrics.stats.TokenBucket
import org.apache.kafka.common.utils.MockTime
+import org.apache.kafka.server.config.ClientQuotaManagerConfig
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Assertions.assertFalse
@@ -115,10 +115,7 @@ class ControllerMutationQuotaManagerTest extends BaseClientQuotaManagerTest {
private val User = "ANONYMOUS"
private val ClientId = "test-client"
- private val config = ClientQuotaManagerConfig(
- numQuotaSamples = 10,
- quotaWindowSizeSeconds = 1
- )
+ private val config = new ClientQuotaManagerConfig(10, 1)
private def withQuotaManager(f: ControllerMutationQuotaManager => Unit): Unit = {
val quotaManager = new ControllerMutationQuotaManager(config, metrics, time,"", None)
diff --git a/core/src/test/scala/unit/kafka/server/ThrottledChannelExpirationTest.scala b/core/src/test/scala/unit/kafka/server/ThrottledChannelExpirationTest.scala
index 15ad22d9725..360614e93a4 100644
--- a/core/src/test/scala/unit/kafka/server/ThrottledChannelExpirationTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ThrottledChannelExpirationTest.scala
@@ -20,9 +20,9 @@ package kafka.server
import java.util.Collections
import java.util.concurrent.{DelayQueue, TimeUnit}
-
import org.apache.kafka.common.metrics.MetricConfig
import org.apache.kafka.common.utils.MockTime
+import org.apache.kafka.server.config.ClientQuotaManagerConfig
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{BeforeEach, Test}
@@ -51,7 +51,7 @@ class ThrottledChannelExpirationTest {
@Test
def testCallbackInvocationAfterExpiration(): Unit = {
- val clientMetrics = new ClientQuotaManager(ClientQuotaManagerConfig(), metrics, QuotaType.Produce, time, "")
+ val clientMetrics = new ClientQuotaManager(new ClientQuotaManagerConfig(), metrics, QuotaType.Produce, time, "")
val delayQueue = new DelayQueue[ThrottledChannel]()
val reaper = new clientMetrics.ThrottledChannelReaper(delayQueue, "")
diff --git a/server/src/main/java/org/apache/kafka/server/config/ClientQuotaManagerConfig.java b/server/src/main/java/org/apache/kafka/server/config/ClientQuotaManagerConfig.java
new file mode 100644
index 00000000000..3ac9b5bd119
--- /dev/null
+++ b/server/src/main/java/org/apache/kafka/server/config/ClientQuotaManagerConfig.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.server.config;
+
+public class ClientQuotaManagerConfig {
+ // Always have 10 whole windows + 1 current window
+ public static final int DEFAULT_NUM_QUOTA_SAMPLES = 11;
+ public static final int DEFAULT_QUOTA_WINDOW_SIZE_SECONDS = 1;
+
+ public final int numQuotaSamples;
+ public final int quotaWindowSizeSeconds;
+
+ /**
+ * Configuration settings for quota management
+ *
+ * @param numQuotaSamples The number of samples to retain in memory
+ * @param quotaWindowSizeSeconds The time span of each sample
+ */
+ public ClientQuotaManagerConfig(int numQuotaSamples, int quotaWindowSizeSeconds) {
+ this.numQuotaSamples = numQuotaSamples;
+ this.quotaWindowSizeSeconds = quotaWindowSizeSeconds;
+ }
+
+ public ClientQuotaManagerConfig() {
+ this(DEFAULT_NUM_QUOTA_SAMPLES, DEFAULT_QUOTA_WINDOW_SIZE_SECONDS);
+ }
+
+ public ClientQuotaManagerConfig(int numQuotaSamples) {
+ this(numQuotaSamples, DEFAULT_QUOTA_WINDOW_SIZE_SECONDS);
+ }
+}