You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2024/04/24 05:05:24 UTC
(kafka) branch trunk updated: KAFKA-15853 Move quota configs into server-common package (#15774)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new cfe5ab5cf23 KAFKA-15853 Move quota configs into server-common package (#15774)
cfe5ab5cf23 is described below
commit cfe5ab5cf23543392a240a7abc38279f343501c3
Author: Omnia Ibrahim <o....@gmail.com>
AuthorDate: Wed Apr 24 06:05:18 2024 +0100
KAFKA-15853 Move quota configs into server-common package (#15774)
Reviewers: Mickael Maison <mi...@gmail.com>, Chia-Ping Tsai <ch...@gmail.com>
---
checkstyle/import-control-metadata.xml | 1 +
.../common/config/internals/QuotaConfigs.java | 98 --------------
.../src/main/scala/kafka/admin/ConfigCommand.scala | 3 +-
.../main/scala/kafka/network/SocketServer.scala | 2 +-
.../main/scala/kafka/server/ConfigHandler.scala | 19 ++-
.../main/scala/kafka/server/DynamicConfig.scala | 37 +----
core/src/main/scala/kafka/server/KafkaConfig.scala | 76 ++++-------
.../src/main/scala/kafka/server/QuotaFactory.scala | 4 +-
.../main/scala/kafka/server/ZkAdminManager.scala | 3 +-
.../metadata/ClientQuotaMetadataManager.scala | 3 +-
.../integration/kafka/api/BaseQuotaTest.scala | 5 +-
.../kafka/api/CustomQuotaCallbackTest.scala | 5 +-
.../kafka/api/PlaintextAdminIntegrationTest.scala | 66 ++++-----
.../kafka/network/DynamicConnectionQuotaTest.scala | 6 +-
.../unit/kafka/admin/ReplicationQuotaUtils.scala | 21 ++-
.../unit/kafka/network/ConnectionQuotasTest.scala | 7 +-
.../unit/kafka/network/SocketServerTest.scala | 5 +-
.../kafka/server/ClientQuotasRequestTest.scala | 140 +++++++++----------
.../kafka/server/ControllerMutationQuotaTest.scala | 7 +-
.../kafka/server/DynamicConfigChangeTest.scala | 22 ++-
.../unit/kafka/server/DynamicConfigTest.scala | 7 +-
.../scala/unit/kafka/server/KafkaConfigTest.scala | 10 +-
.../KafkaMetricReporterExceptionHandlingTest.scala | 2 +-
.../unit/kafka/server/ReplicationQuotasTest.scala | 19 ++-
.../scala/unit/kafka/server/RequestQuotaTest.scala | 3 +-
.../test/scala/unit/kafka/utils/TestUtils.scala | 15 +--
.../scala/unit/kafka/zk/AdminZkClientTest.scala | 24 ++--
.../zk/migration/ZkConfigMigrationClientTest.scala | 4 +-
.../controller/ClientQuotaControlManager.java | 2 +-
.../controller/ClientQuotaControlManagerTest.java | 3 +-
.../apache/kafka/image/ClientQuotasImageTest.java | 3 +-
.../apache/kafka/server/config/QuotaConfigs.java | 150 +++++++++++++++++++++
.../server/config/ClientQuotaManagerConfig.java | 9 +-
.../org/apache/kafka/server/config/Defaults.java | 10 --
.../config/ReplicationQuotaManagerConfig.java | 10 +-
.../kafka/storage/internals/log/LogConfig.java | 32 ++---
.../tools/reassign/ReassignPartitionsCommand.java | 33 ++---
.../org/apache/kafka/tools/ToolsTestUtils.java | 15 +--
.../ReassignPartitionsIntegrationTest.java | 18 ++-
.../tools/reassign/ReassignPartitionsUnitTest.java | 17 +--
40 files changed, 422 insertions(+), 494 deletions(-)
diff --git a/checkstyle/import-control-metadata.xml b/checkstyle/import-control-metadata.xml
index 897932492b1..0d1088d7c3e 100644
--- a/checkstyle/import-control-metadata.xml
+++ b/checkstyle/import-control-metadata.xml
@@ -119,6 +119,7 @@
<allow pkg="org.apache.kafka.clients.admin" />
<allow pkg="org.apache.kafka.raft" />
<allow pkg="org.apache.kafka.server.common" />
+ <allow pkg="org.apache.kafka.server.config" />
<allow pkg="org.apache.kafka.server.fault" />
<allow pkg="org.apache.kafka.server.util" />
<allow pkg="org.apache.kafka.snapshot" />
diff --git a/clients/src/main/java/org/apache/kafka/common/config/internals/QuotaConfigs.java b/clients/src/main/java/org/apache/kafka/common/config/internals/QuotaConfigs.java
deleted file mode 100644
index e382cb0c172..00000000000
--- a/clients/src/main/java/org/apache/kafka/common/config/internals/QuotaConfigs.java
+++ /dev/null
@@ -1,98 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.kafka.common.config.internals;
-
-import org.apache.kafka.common.config.ConfigDef;
-import org.apache.kafka.common.security.scram.internals.ScramMechanism;
-
-import java.util.Arrays;
-import java.util.HashSet;
-import java.util.Set;
-
-/**
- * Define the dynamic quota configs. Note that these are not normal configurations that exist in properties files. They
- * only exist dynamically in the controller (or ZK, depending on which mode the cluster is running).
- */
-public class QuotaConfigs {
- public static final String PRODUCER_BYTE_RATE_OVERRIDE_CONFIG = "producer_byte_rate";
- public static final String CONSUMER_BYTE_RATE_OVERRIDE_CONFIG = "consumer_byte_rate";
- public static final String REQUEST_PERCENTAGE_OVERRIDE_CONFIG = "request_percentage";
- public static final String CONTROLLER_MUTATION_RATE_OVERRIDE_CONFIG = "controller_mutation_rate";
- public static final String IP_CONNECTION_RATE_OVERRIDE_CONFIG = "connection_creation_rate";
-
- public static final String PRODUCER_BYTE_RATE_DOC = "A rate representing the upper bound (bytes/sec) for producer traffic.";
- public static final String CONSUMER_BYTE_RATE_DOC = "A rate representing the upper bound (bytes/sec) for consumer traffic.";
- public static final String REQUEST_PERCENTAGE_DOC = "A percentage representing the upper bound of time spent for processing requests.";
- public static final String CONTROLLER_MUTATION_RATE_DOC = "The rate at which mutations are accepted for the create " +
- "topics request, the create partitions request and the delete topics request. The rate is accumulated by " +
- "the number of partitions created or deleted.";
- public static final String IP_CONNECTION_RATE_DOC = "An int representing the upper bound of connections accepted " +
- "for the specified IP.";
-
- public static final int IP_CONNECTION_RATE_DEFAULT = Integer.MAX_VALUE;
-
- private final static Set<String> USER_AND_CLIENT_QUOTA_NAMES = new HashSet<>(Arrays.asList(
- PRODUCER_BYTE_RATE_OVERRIDE_CONFIG,
- CONSUMER_BYTE_RATE_OVERRIDE_CONFIG,
- REQUEST_PERCENTAGE_OVERRIDE_CONFIG,
- CONTROLLER_MUTATION_RATE_OVERRIDE_CONFIG
- ));
-
- private static void buildUserClientQuotaConfigDef(ConfigDef configDef) {
- configDef.define(PRODUCER_BYTE_RATE_OVERRIDE_CONFIG, ConfigDef.Type.LONG, Long.MAX_VALUE,
- ConfigDef.Importance.MEDIUM, PRODUCER_BYTE_RATE_DOC);
-
- configDef.define(CONSUMER_BYTE_RATE_OVERRIDE_CONFIG, ConfigDef.Type.LONG, Long.MAX_VALUE,
- ConfigDef.Importance.MEDIUM, CONSUMER_BYTE_RATE_DOC);
-
- configDef.define(REQUEST_PERCENTAGE_OVERRIDE_CONFIG, ConfigDef.Type.DOUBLE,
- Integer.valueOf(Integer.MAX_VALUE).doubleValue(),
- ConfigDef.Importance.MEDIUM, REQUEST_PERCENTAGE_DOC);
-
- configDef.define(CONTROLLER_MUTATION_RATE_OVERRIDE_CONFIG, ConfigDef.Type.DOUBLE,
- Integer.valueOf(Integer.MAX_VALUE).doubleValue(),
- ConfigDef.Importance.MEDIUM, CONTROLLER_MUTATION_RATE_DOC);
- }
-
- public static boolean isClientOrUserConfig(String name) {
- return USER_AND_CLIENT_QUOTA_NAMES.contains(name);
- }
-
- public static ConfigDef userAndClientQuotaConfigs() {
- ConfigDef configDef = new ConfigDef();
- buildUserClientQuotaConfigDef(configDef);
- return configDef;
- }
-
- public static ConfigDef scramMechanismsPlusUserAndClientQuotaConfigs() {
- ConfigDef configDef = new ConfigDef();
- ScramMechanism.mechanismNames().forEach(mechanismName -> {
- configDef.define(mechanismName, ConfigDef.Type.STRING, null, ConfigDef.Importance.MEDIUM,
- "User credentials for SCRAM mechanism " + mechanismName);
- });
- buildUserClientQuotaConfigDef(configDef);
- return configDef;
- }
-
- public static ConfigDef ipConfigs() {
- ConfigDef configDef = new ConfigDef();
- configDef.define(IP_CONNECTION_RATE_OVERRIDE_CONFIG, ConfigDef.Type.INT, Integer.MAX_VALUE,
- ConfigDef.Range.atLeast(0), ConfigDef.Importance.MEDIUM, IP_CONNECTION_RATE_DOC);
- return configDef;
- }
-}
diff --git a/core/src/main/scala/kafka/admin/ConfigCommand.scala b/core/src/main/scala/kafka/admin/ConfigCommand.scala
index 644043f7610..3bb0c62b288 100644
--- a/core/src/main/scala/kafka/admin/ConfigCommand.scala
+++ b/core/src/main/scala/kafka/admin/ConfigCommand.scala
@@ -22,7 +22,6 @@ import java.util.concurrent.TimeUnit
import java.util.{Collections, Properties}
import joptsimple._
import kafka.server.{DynamicBrokerConfig, DynamicConfig, KafkaConfig}
-import kafka.server.DynamicConfig.QuotaConfigs
import kafka.utils.Implicits._
import kafka.utils.{Exit, Logging}
import kafka.zk.{AdminZkClient, KafkaZkClient}
@@ -35,7 +34,7 @@ import org.apache.kafka.common.quota.{ClientQuotaAlteration, ClientQuotaEntity,
import org.apache.kafka.common.security.JaasUtils
import org.apache.kafka.common.security.scram.internals.{ScramCredentialUtils, ScramFormatter, ScramMechanism}
import org.apache.kafka.common.utils.{Sanitizer, Time, Utils}
-import org.apache.kafka.server.config.{ConfigType, ZooKeeperInternals, ZkConfigs}
+import org.apache.kafka.server.config.{ConfigType, QuotaConfigs, ZkConfigs, ZooKeeperInternals}
import org.apache.kafka.security.{PasswordEncoder, PasswordEncoderConfigs}
import org.apache.kafka.server.util.{CommandDefaultOptions, CommandLineUtils}
import org.apache.kafka.storage.internals.log.LogConfig
diff --git a/core/src/main/scala/kafka/network/SocketServer.scala b/core/src/main/scala/kafka/network/SocketServer.scala
index 40b68cd6fb2..8ef05f8027a 100644
--- a/core/src/main/scala/kafka/network/SocketServer.scala
+++ b/core/src/main/scala/kafka/network/SocketServer.scala
@@ -34,7 +34,6 @@ import kafka.server.{ApiVersionManager, BrokerReconfigurable, KafkaConfig}
import org.apache.kafka.common.message.ApiMessageType.ListenerType
import kafka.utils._
import org.apache.kafka.common.config.ConfigException
-import org.apache.kafka.common.config.internals.QuotaConfigs
import org.apache.kafka.common.errors.InvalidRequestException
import org.apache.kafka.common.memory.{MemoryPool, SimpleMemoryPool}
import org.apache.kafka.common.metrics._
@@ -48,6 +47,7 @@ import org.apache.kafka.common.utils.{KafkaThread, LogContext, Time, Utils}
import org.apache.kafka.common.{Endpoint, KafkaException, MetricName, Reconfigurable}
import org.apache.kafka.network.SocketServerConfigs
import org.apache.kafka.security.CredentialProvider
+import org.apache.kafka.server.config.QuotaConfigs
import org.apache.kafka.server.metrics.KafkaMetricsGroup
import org.apache.kafka.server.util.FutureUtils
import org.slf4j.event.Level
diff --git a/core/src/main/scala/kafka/server/ConfigHandler.scala b/core/src/main/scala/kafka/server/ConfigHandler.scala
index 206e31ea499..1d5702e76e4 100644
--- a/core/src/main/scala/kafka/server/ConfigHandler.scala
+++ b/core/src/main/scala/kafka/server/ConfigHandler.scala
@@ -19,7 +19,6 @@ package kafka.server
import java.net.{InetAddress, UnknownHostException}
import java.util.{Collections, Properties}
-import DynamicConfig.Broker._
import kafka.controller.KafkaController
import kafka.log.UnifiedLog
import kafka.network.ConnectionQuotas
@@ -27,16 +26,14 @@ import kafka.server.Constants._
import kafka.server.QuotaFactory.QuotaManagers
import kafka.utils.Implicits._
import kafka.utils.Logging
-import org.apache.kafka.server.config.ReplicationConfigs
+import org.apache.kafka.server.config.{ReplicationConfigs, QuotaConfigs, ZooKeeperInternals}
import org.apache.kafka.common.config.TopicConfig
-import org.apache.kafka.common.config.internals.QuotaConfigs
import org.apache.kafka.common.metrics.Quota
import org.apache.kafka.common.metrics.Quota._
import org.apache.kafka.common.utils.Sanitizer
import org.apache.kafka.security.CredentialProvider
import org.apache.kafka.server.ClientMetricsManager
-import org.apache.kafka.server.config.ZooKeeperInternals
-import org.apache.kafka.storage.internals.log.{LogConfig, ThrottledReplicaListValidator}
+import org.apache.kafka.storage.internals.log.ThrottledReplicaListValidator
import org.apache.kafka.storage.internals.log.LogConfig.MessageFormatVersion
import scala.annotation.nowarn
@@ -106,8 +103,8 @@ class TopicConfigHandler(private val replicaManager: ReplicaManager,
debug(s"Removing $prop from broker ${kafkaConfig.brokerId} for topic $topic")
}
}
- updateThrottledList(LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, quotas.leader)
- updateThrottledList(LogConfig.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG, quotas.follower)
+ updateThrottledList(QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, quotas.leader)
+ updateThrottledList(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG, quotas.follower)
if (Try(topicConfig.getProperty(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG).toBoolean).getOrElse(false)) {
kafkaController.foreach(_.enableTopicUncleanLeaderElection(topic))
@@ -245,15 +242,15 @@ class BrokerConfigHandler(private val brokerConfig: KafkaConfig,
if (properties.containsKey(prop))
properties.getProperty(prop).toLong
else
- DefaultReplicationThrottledRate
+ QuotaConfigs.QUOTA_BYTES_PER_SECOND_DEFAULT
}
if (brokerId == ZooKeeperInternals.DEFAULT_STRING)
brokerConfig.dynamicConfig.updateDefaultConfig(properties)
else if (brokerConfig.brokerId == brokerId.trim.toInt) {
brokerConfig.dynamicConfig.updateBrokerConfig(brokerConfig.brokerId, properties)
- quotaManagers.leader.updateQuota(upperBound(getOrDefault(LeaderReplicationThrottledRateProp).toDouble))
- quotaManagers.follower.updateQuota(upperBound(getOrDefault(FollowerReplicationThrottledRateProp).toDouble))
- quotaManagers.alterLogDirs.updateQuota(upperBound(getOrDefault(ReplicaAlterLogDirsIoMaxBytesPerSecondProp).toDouble))
+ quotaManagers.leader.updateQuota(upperBound(getOrDefault(QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG).toDouble))
+ quotaManagers.follower.updateQuota(upperBound(getOrDefault(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG).toDouble))
+ quotaManagers.alterLogDirs.updateQuota(upperBound(getOrDefault(QuotaConfigs.REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_CONFIG).toDouble))
}
}
}
diff --git a/core/src/main/scala/kafka/server/DynamicConfig.scala b/core/src/main/scala/kafka/server/DynamicConfig.scala
index 4fbbacd8f34..5a5e3893179 100644
--- a/core/src/main/scala/kafka/server/DynamicConfig.scala
+++ b/core/src/main/scala/kafka/server/DynamicConfig.scala
@@ -23,8 +23,7 @@ import org.apache.kafka.common.config.ConfigDef
import org.apache.kafka.common.config.ConfigDef.Importance._
import org.apache.kafka.common.config.ConfigDef.Range._
import org.apache.kafka.common.config.ConfigDef.Type._
-import org.apache.kafka.server.config.{ReplicationQuotaManagerConfig, ZooKeeperInternals}
-import org.apache.kafka.storage.internals.log.LogConfig
+import org.apache.kafka.server.config.{QuotaConfigs, ZooKeeperInternals}
import java.util
import scala.jdk.CollectionConverters._
@@ -36,30 +35,12 @@ import scala.jdk.CollectionConverters._
object DynamicConfig {
object Broker {
- // Properties
- val LeaderReplicationThrottledRateProp = "leader.replication.throttled.rate"
- val FollowerReplicationThrottledRateProp = "follower.replication.throttled.rate"
- val ReplicaAlterLogDirsIoMaxBytesPerSecondProp = "replica.alter.log.dirs.io.max.bytes.per.second"
-
- // Defaults
- val DefaultReplicationThrottledRate = ReplicationQuotaManagerConfig.DEFAULT_QUOTA_BYTES_PER_SECOND
-
- // Documentation
- val LeaderReplicationThrottledRateDoc = "A long representing the upper bound (bytes/sec) on replication traffic for leaders enumerated in the " +
- s"property ${LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG} (for each topic). This property can be only set dynamically. It is suggested that the " +
- s"limit be kept above 1MB/s for accurate behaviour."
- val FollowerReplicationThrottledRateDoc = "A long representing the upper bound (bytes/sec) on replication traffic for followers enumerated in the " +
- s"property ${LogConfig.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG} (for each topic). This property can be only set dynamically. It is suggested that the " +
- s"limit be kept above 1MB/s for accurate behaviour."
- val ReplicaAlterLogDirsIoMaxBytesPerSecondDoc = "A long representing the upper bound (bytes/sec) on disk IO used for moving replica between log directories on the same broker. " +
- s"This property can be only set dynamically. It is suggested that the limit be kept above 1MB/s for accurate behaviour."
-
// Definitions
val brokerConfigDef = new ConfigDef()
// Round minimum value down, to make it easier for users.
- .define(LeaderReplicationThrottledRateProp, LONG, DefaultReplicationThrottledRate, atLeast(0), MEDIUM, LeaderReplicationThrottledRateDoc)
- .define(FollowerReplicationThrottledRateProp, LONG, DefaultReplicationThrottledRate, atLeast(0), MEDIUM, FollowerReplicationThrottledRateDoc)
- .define(ReplicaAlterLogDirsIoMaxBytesPerSecondProp, LONG, DefaultReplicationThrottledRate, atLeast(0), MEDIUM, ReplicaAlterLogDirsIoMaxBytesPerSecondDoc)
+ .define(QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG, LONG, QuotaConfigs.QUOTA_BYTES_PER_SECOND_DEFAULT, atLeast(0), MEDIUM, QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_DOC)
+ .define(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG, LONG, QuotaConfigs.QUOTA_BYTES_PER_SECOND_DEFAULT, atLeast(0), MEDIUM, QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_DOC)
+ .define(QuotaConfigs.REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_CONFIG, LONG, QuotaConfigs.QUOTA_BYTES_PER_SECOND_DEFAULT, atLeast(0), MEDIUM, QuotaConfigs.REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_DOC)
DynamicBrokerConfig.addDynamicConfigs(brokerConfigDef)
val nonDynamicProps = KafkaConfig.configNames.toSet -- brokerConfigDef.names.asScala
@@ -68,12 +49,8 @@ object DynamicConfig {
def validate(props: Properties) = DynamicConfig.validate(brokerConfigDef, props, customPropsAllowed = true)
}
- object QuotaConfigs {
- def isClientOrUserQuotaConfig(name: String): Boolean = org.apache.kafka.common.config.internals.QuotaConfigs.isClientOrUserConfig(name)
- }
-
object Client {
- private val clientConfigs = org.apache.kafka.common.config.internals.QuotaConfigs.userAndClientQuotaConfigs()
+ private val clientConfigs = QuotaConfigs.userAndClientQuotaConfigs()
def configKeys = clientConfigs.configKeys
@@ -83,7 +60,7 @@ object DynamicConfig {
}
object User {
- private val userConfigs = org.apache.kafka.common.config.internals.QuotaConfigs.scramMechanismsPlusUserAndClientQuotaConfigs()
+ private val userConfigs = QuotaConfigs.scramMechanismsPlusUserAndClientQuotaConfigs()
def configKeys = userConfigs.configKeys
@@ -93,7 +70,7 @@ object DynamicConfig {
}
object Ip {
- private val ipConfigs = org.apache.kafka.common.config.internals.QuotaConfigs.ipConfigs()
+ private val ipConfigs = QuotaConfigs.ipConfigs()
def configKeys = ipConfigs.configKeys
diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index 9ef24228d1d..47fe2910b35 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.{Defaults, KafkaSecurityConfigs, ReplicationConfigs, ServerLogConfigs, ServerTopicConfigSynonyms, ZkConfigs}
+import org.apache.kafka.server.config.{Defaults, KafkaSecurityConfigs, QuotaConfigs, ReplicationConfigs, ServerLogConfigs, ServerTopicConfigSynonyms, ZkConfigs}
import org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig
import org.apache.kafka.server.record.BrokerCompressionType
import org.apache.kafka.server.util.Csv
@@ -106,6 +106,8 @@ object KafkaConfig {
val RequestTimeoutMsProp = CommonClientConfigs.REQUEST_TIMEOUT_MS_CONFIG
val ConnectionSetupTimeoutMsProp = CommonClientConfigs.SOCKET_CONNECTION_SETUP_TIMEOUT_MS_CONFIG
val ConnectionSetupTimeoutMaxMsProp = CommonClientConfigs.SOCKET_CONNECTION_SETUP_TIMEOUT_MAX_MS_CONFIG
+ val DeleteTopicEnableProp = "delete.topic.enable"
+ val CompressionTypeProp = ServerTopicConfigSynonyms.serverSynonym(TopicConfig.COMPRESSION_TYPE_CONFIG)
/** KRaft mode configs */
val ProcessRolesProp = "process.roles"
@@ -153,20 +155,6 @@ object KafkaConfig {
/** ********* Request Limit Configuration **************/
val MaxRequestPartitionSizeLimit = "max.request.partition.size.limit"
- /** ********* Quota Configuration ***********/
- val NumQuotaSamplesProp = "quota.window.num"
- val NumReplicationQuotaSamplesProp = "replication.quota.window.num"
- val NumAlterLogDirsReplicationQuotaSamplesProp = "alter.log.dirs.replication.quota.window.num"
- val NumControllerQuotaSamplesProp = "controller.quota.window.num"
- val QuotaWindowSizeSecondsProp = "quota.window.size.seconds"
- val ReplicationQuotaWindowSizeSecondsProp = "replication.quota.window.size.seconds"
- val AlterLogDirsReplicationQuotaWindowSizeSecondsProp = "alter.log.dirs.replication.quota.window.size.seconds"
- val ControllerQuotaWindowSizeSecondsProp = "controller.quota.window.size.seconds"
- val ClientQuotaCallbackClassProp = "client.quota.callback.class"
-
- val DeleteTopicEnableProp = "delete.topic.enable"
- val CompressionTypeProp = ServerTopicConfigSynonyms.serverSynonym(TopicConfig.COMPRESSION_TYPE_CONFIG)
-
/** ********* Kafka Metrics Configuration ***********/
val MetricSampleWindowMsProp = CommonClientConfigs.METRICS_SAMPLE_WINDOW_MS_CONFIG
val MetricNumSamplesProp: String = CommonClientConfigs.METRICS_NUM_SAMPLES_CONFIG
@@ -211,6 +199,10 @@ object KafkaConfig {
val RequestTimeoutMsDoc = CommonClientConfigs.REQUEST_TIMEOUT_MS_DOC
val ConnectionSetupTimeoutMsDoc = CommonClientConfigs.SOCKET_CONNECTION_SETUP_TIMEOUT_MS_DOC
val ConnectionSetupTimeoutMaxMsDoc = CommonClientConfigs.SOCKET_CONNECTION_SETUP_TIMEOUT_MAX_MS_DOC
+ val DeleteTopicEnableDoc = "Enables delete topic. Delete topic through the admin tool will have no effect if this config is turned off"
+ val CompressionTypeDoc = "Specify the final compression type for a given topic. This configuration accepts the standard compression codecs " +
+ "('gzip', 'snappy', 'lz4', 'zstd'). It additionally accepts 'uncompressed' which is equivalent to no compression; and " +
+ "'producer' which means retain the original compression codec set by the producer."
/** KRaft mode configs */
val ProcessRolesDoc = "The roles that this process plays: 'broker', 'controller', or 'broker,controller' if it is both. " +
@@ -274,26 +266,6 @@ object KafkaConfig {
/** ********* Request Limit Configuration **************/
val MaxRequestPartitionSizeLimitDoc = "The maximum number of partitions can be served in one request."
- /** ********* Quota Configuration ***********/
- 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 NumAlterLogDirsReplicationQuotaSamplesDoc = "The number of samples to retain in memory for alter log dirs replication quotas"
- val NumControllerQuotaSamplesDoc = "The number of samples to retain in memory for controller mutation quotas"
- val QuotaWindowSizeSecondsDoc = "The time span of each sample for client quotas"
- val ReplicationQuotaWindowSizeSecondsDoc = "The time span of each sample for replication quotas"
- val AlterLogDirsReplicationQuotaWindowSizeSecondsDoc = "The time span of each sample for alter log dirs replication quotas"
- val ControllerQuotaWindowSizeSecondsDoc = "The time span of each sample for controller mutations quotas"
-
- val ClientQuotaCallbackClassDoc = "The fully qualified name of a class that implements the ClientQuotaCallback interface, " +
- "which is used to determine quota limits applied to client requests. By default, the <user> and <client-id> " +
- "quotas that are stored in ZooKeeper are applied. For any given request, the most specific quota that matches the user principal " +
- "of the session and the client-id of the request is applied."
-
- val DeleteTopicEnableDoc = "Enables delete topic. Delete topic through the admin tool will have no effect if this config is turned off"
- val CompressionTypeDoc = "Specify the final compression type for a given topic. This configuration accepts the standard compression codecs " +
- "('gzip', 'snappy', 'lz4', 'zstd'). It additionally accepts 'uncompressed' which is equivalent to no compression; and " +
- "'producer' which means retain the original compression codec set by the producer."
-
/** ********* Kafka Metrics Configuration ***********/
val MetricSampleWindowMsDoc = CommonClientConfigs.METRICS_SAMPLE_WINDOW_MS_DOC
val MetricNumSamplesDoc = CommonClientConfigs.METRICS_NUM_SAMPLES_DOC
@@ -578,15 +550,15 @@ object KafkaConfig {
.define(ClientTelemetryMaxBytesProp, INT, Defaults.CLIENT_TELEMETRY_MAX_BYTES, atLeast(1), LOW, ClientTelemetryMaxBytesDoc)
/** ********* Quota configuration ***********/
- .define(NumQuotaSamplesProp, INT, Defaults.NUM_QUOTA_SAMPLES, atLeast(1), LOW, NumQuotaSamplesDoc)
- .define(NumReplicationQuotaSamplesProp, INT, Defaults.NUM_REPLICATION_QUOTA_SAMPLES, atLeast(1), LOW, NumReplicationQuotaSamplesDoc)
- .define(NumAlterLogDirsReplicationQuotaSamplesProp, INT, Defaults.NUM_ALTER_LOG_DIRS_REPLICATION_QUOTA_SAMPLES, atLeast(1), LOW, NumAlterLogDirsReplicationQuotaSamplesDoc)
- .define(NumControllerQuotaSamplesProp, INT, Defaults.NUM_CONTROLLER_QUOTA_SAMPLES, atLeast(1), LOW, NumControllerQuotaSamplesDoc)
- .define(QuotaWindowSizeSecondsProp, INT, Defaults.QUOTA_WINDOW_SIZE_SECONDS, atLeast(1), LOW, QuotaWindowSizeSecondsDoc)
- .define(ReplicationQuotaWindowSizeSecondsProp, INT, Defaults.REPLICATION_QUOTA_WINDOW_SIZE_SECONDS, atLeast(1), LOW, ReplicationQuotaWindowSizeSecondsDoc)
- .define(AlterLogDirsReplicationQuotaWindowSizeSecondsProp, INT, Defaults.ALTER_LOG_DIRS_REPLICATION_QUOTA_WINDOW_SIZE_SECONDS, atLeast(1), LOW, AlterLogDirsReplicationQuotaWindowSizeSecondsDoc)
- .define(ControllerQuotaWindowSizeSecondsProp, INT, Defaults.CONTROLLER_QUOTA_WINDOW_SIZE_SECONDS, atLeast(1), LOW, ControllerQuotaWindowSizeSecondsDoc)
- .define(ClientQuotaCallbackClassProp, CLASS, null, LOW, ClientQuotaCallbackClassDoc)
+ .define(QuotaConfigs.NUM_QUOTA_SAMPLES_CONFIG, INT, QuotaConfigs.NUM_QUOTA_SAMPLES_DEFAULT, atLeast(1), LOW, QuotaConfigs.NUM_QUOTA_SAMPLES_DOC)
+ .define(QuotaConfigs.NUM_REPLICATION_QUOTA_SAMPLES_CONFIG, INT, QuotaConfigs.NUM_QUOTA_SAMPLES_DEFAULT, atLeast(1), LOW, QuotaConfigs.NUM_REPLICATION_QUOTA_SAMPLES_DOC)
+ .define(QuotaConfigs.NUM_ALTER_LOG_DIRS_REPLICATION_QUOTA_SAMPLES_CONFIG, INT, QuotaConfigs.NUM_QUOTA_SAMPLES_DEFAULT, atLeast(1), LOW, QuotaConfigs.NUM_ALTER_LOG_DIRS_REPLICATION_QUOTA_SAMPLES_DOC)
+ .define(QuotaConfigs.NUM_CONTROLLER_QUOTA_SAMPLES_CONFIG, INT, QuotaConfigs.NUM_QUOTA_SAMPLES_DEFAULT, atLeast(1), LOW, QuotaConfigs.NUM_CONTROLLER_QUOTA_SAMPLES_DOC)
+ .define(QuotaConfigs.QUOTA_WINDOW_SIZE_SECONDS_CONFIG, INT, QuotaConfigs.QUOTA_WINDOW_SIZE_SECONDS_DEFAULT, atLeast(1), LOW, QuotaConfigs.QUOTA_WINDOW_SIZE_SECONDS_DOC)
+ .define(QuotaConfigs.REPLICATION_QUOTA_WINDOW_SIZE_SECONDS_CONFIG, INT, QuotaConfigs.QUOTA_WINDOW_SIZE_SECONDS_DEFAULT, atLeast(1), LOW, QuotaConfigs.REPLICATION_QUOTA_WINDOW_SIZE_SECONDS_DOC)
+ .define(QuotaConfigs.ALTER_LOG_DIRS_REPLICATION_QUOTA_WINDOW_SIZE_SECONDS_CONFIG, INT, QuotaConfigs.QUOTA_WINDOW_SIZE_SECONDS_DEFAULT, atLeast(1), LOW, QuotaConfigs.ALTER_LOG_DIRS_REPLICATION_QUOTA_WINDOW_SIZE_SECONDS_DOC)
+ .define(QuotaConfigs.CONTROLLER_QUOTA_WINDOW_SIZE_SECONDS_CONFIG, INT, QuotaConfigs.QUOTA_WINDOW_SIZE_SECONDS_DEFAULT, atLeast(1), LOW, QuotaConfigs.CONTROLLER_QUOTA_WINDOW_SIZE_SECONDS_DOC)
+ .define(QuotaConfigs.CLIENT_QUOTA_CALLBACK_CLASS_CONFIG, CLASS, null, LOW, QuotaConfigs.CLIENT_QUOTA_CALLBACK_CLASS_DOC)
/** ********* General Security Configuration ****************/
.define(KafkaSecurityConfigs.CONNECTIONS_MAX_REAUTH_MS_CONFIG, LONG, KafkaSecurityConfigs.CONNECTIONS_MAX_REAUTH_MS_DEFAULT, MEDIUM, KafkaSecurityConfigs.CONNECTIONS_MAX_REAUTH_MS_DOC)
@@ -1252,14 +1224,14 @@ class KafkaConfig private(doLog: Boolean, val props: java.util.Map[_, _], dynami
def passwordEncoderIterations = getInt(PasswordEncoderConfigs.PASSWORD_ENCODER_ITERATIONS_CONFIG)
/** ********* Quota Configuration **************/
- val numQuotaSamples = getInt(KafkaConfig.NumQuotaSamplesProp)
- val quotaWindowSizeSeconds = getInt(KafkaConfig.QuotaWindowSizeSecondsProp)
- val numReplicationQuotaSamples = getInt(KafkaConfig.NumReplicationQuotaSamplesProp)
- val replicationQuotaWindowSizeSeconds = getInt(KafkaConfig.ReplicationQuotaWindowSizeSecondsProp)
- val numAlterLogDirsReplicationQuotaSamples = getInt(KafkaConfig.NumAlterLogDirsReplicationQuotaSamplesProp)
- val alterLogDirsReplicationQuotaWindowSizeSeconds = getInt(KafkaConfig.AlterLogDirsReplicationQuotaWindowSizeSecondsProp)
- val numControllerQuotaSamples = getInt(KafkaConfig.NumControllerQuotaSamplesProp)
- val controllerQuotaWindowSizeSeconds = getInt(KafkaConfig.ControllerQuotaWindowSizeSecondsProp)
+ val numQuotaSamples = getInt(QuotaConfigs.NUM_QUOTA_SAMPLES_CONFIG)
+ val quotaWindowSizeSeconds = getInt(QuotaConfigs.QUOTA_WINDOW_SIZE_SECONDS_CONFIG)
+ val numReplicationQuotaSamples = getInt(QuotaConfigs.NUM_REPLICATION_QUOTA_SAMPLES_CONFIG)
+ val replicationQuotaWindowSizeSeconds = getInt(QuotaConfigs.REPLICATION_QUOTA_WINDOW_SIZE_SECONDS_CONFIG)
+ val numAlterLogDirsReplicationQuotaSamples = getInt(QuotaConfigs.NUM_ALTER_LOG_DIRS_REPLICATION_QUOTA_SAMPLES_CONFIG)
+ val alterLogDirsReplicationQuotaWindowSizeSeconds = getInt(QuotaConfigs.ALTER_LOG_DIRS_REPLICATION_QUOTA_WINDOW_SIZE_SECONDS_CONFIG)
+ val numControllerQuotaSamples = getInt(QuotaConfigs.NUM_CONTROLLER_QUOTA_SAMPLES_CONFIG)
+ val controllerQuotaWindowSizeSeconds = getInt(QuotaConfigs.CONTROLLER_QUOTA_WINDOW_SIZE_SECONDS_CONFIG)
/** ********* Fetch Configuration **************/
val maxIncrementalFetchSessionCacheSlots = getInt(KafkaConfig.MaxIncrementalFetchSessionCacheSlots)
diff --git a/core/src/main/scala/kafka/server/QuotaFactory.scala b/core/src/main/scala/kafka/server/QuotaFactory.scala
index 2a828f777d3..a501427ca28 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.{ClientQuotaManagerConfig, ReplicationQuotaManagerConfig}
+import org.apache.kafka.server.config.{ClientQuotaManagerConfig, ReplicationQuotaManagerConfig, QuotaConfigs}
import org.apache.kafka.server.quota.ClientQuotaType
object QuotaType {
@@ -74,7 +74,7 @@ object QuotaFactory extends Logging {
def instantiate(cfg: KafkaConfig, metrics: Metrics, time: Time, threadNamePrefix: String): QuotaManagers = {
- val clientQuotaCallback = Option(cfg.getConfiguredInstance(KafkaConfig.ClientQuotaCallbackClassProp,
+ val clientQuotaCallback = Option(cfg.getConfiguredInstance(QuotaConfigs.CLIENT_QUOTA_CALLBACK_CLASS_CONFIG,
classOf[ClientQuotaCallback]))
QuotaManagers(
new ClientQuotaManager(clientConfig(cfg), metrics, Fetch, time, threadNamePrefix, clientQuotaCallback),
diff --git a/core/src/main/scala/kafka/server/ZkAdminManager.scala b/core/src/main/scala/kafka/server/ZkAdminManager.scala
index 351fea4b81b..84f18c95a02 100644
--- a/core/src/main/scala/kafka/server/ZkAdminManager.scala
+++ b/core/src/main/scala/kafka/server/ZkAdminManager.scala
@@ -20,7 +20,6 @@ import java.util
import java.util.Properties
import kafka.common.TopicAlreadyMarkedForDeletionException
import kafka.server.ConfigAdminManager.{prepareIncrementalConfigs, toLoggableProps}
-import kafka.server.DynamicConfig.QuotaConfigs
import kafka.server.metadata.ZkConfigRepository
import kafka.utils._
import kafka.utils.Implicits._
@@ -48,7 +47,7 @@ import org.apache.kafka.common.requests.{AlterConfigsRequest, ApiError}
import org.apache.kafka.common.security.scram.internals.{ScramCredentialUtils, ScramFormatter}
import org.apache.kafka.common.utils.Sanitizer
import org.apache.kafka.server.common.AdminOperationException
-import org.apache.kafka.server.config.{ConfigType, ZooKeeperInternals}
+import org.apache.kafka.server.config.{ConfigType, QuotaConfigs, ZooKeeperInternals}
import org.apache.kafka.server.config.ServerLogConfigs.CREATE_TOPIC_POLICY_CLASS_NAME_CONFIG
import org.apache.kafka.server.config.ServerLogConfigs.ALTER_CONFIG_POLICY_CLASS_NAME_CONFIG
import org.apache.kafka.storage.internals.log.LogConfig
diff --git a/core/src/main/scala/kafka/server/metadata/ClientQuotaMetadataManager.scala b/core/src/main/scala/kafka/server/metadata/ClientQuotaMetadataManager.scala
index 919d203a4d9..045eb307326 100644
--- a/core/src/main/scala/kafka/server/metadata/ClientQuotaMetadataManager.scala
+++ b/core/src/main/scala/kafka/server/metadata/ClientQuotaMetadataManager.scala
@@ -20,14 +20,13 @@ package kafka.server.metadata
import kafka.network.ConnectionQuotas
import kafka.server.QuotaFactory.QuotaManagers
import kafka.utils.Logging
-import org.apache.kafka.common.config.internals.QuotaConfigs
import org.apache.kafka.common.metrics.Quota
import org.apache.kafka.common.quota.ClientQuotaEntity
import org.apache.kafka.common.utils.Sanitizer
import java.net.{InetAddress, UnknownHostException}
import org.apache.kafka.image.{ClientQuotaDelta, ClientQuotasDelta}
-import org.apache.kafka.server.config.ZooKeeperInternals
+import org.apache.kafka.server.config.{QuotaConfigs, ZooKeeperInternals}
import scala.compat.java8.OptionConverters._
diff --git a/core/src/test/scala/integration/kafka/api/BaseQuotaTest.scala b/core/src/test/scala/integration/kafka/api/BaseQuotaTest.scala
index a1ead48d331..c443b5afaff 100644
--- a/core/src/test/scala/integration/kafka/api/BaseQuotaTest.scala
+++ b/core/src/test/scala/integration/kafka/api/BaseQuotaTest.scala
@@ -25,7 +25,6 @@ import org.apache.kafka.clients.admin.Admin
import org.apache.kafka.clients.consumer.{Consumer, ConsumerConfig}
import org.apache.kafka.clients.producer._
import org.apache.kafka.clients.producer.internals.ErrorLoggingCallback
-import org.apache.kafka.common.config.internals.QuotaConfigs
import org.apache.kafka.common.{Metric, MetricName, TopicPartition}
import org.apache.kafka.common.metrics.{KafkaMetric, Quota}
import org.apache.kafka.common.protocol.ApiKeys
@@ -33,8 +32,8 @@ 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.coordinator.group.GroupCoordinatorConfig
+import org.apache.kafka.server.config.QuotaConfigs
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
@@ -184,7 +183,7 @@ abstract class BaseQuotaTest extends IntegrationTestHarness {
assertTrue(throttled, "Should have been throttled")
quotaTestClients.verifyConsumerClientThrottleTimeMetric(expectThrottle = true,
- Some(ClientQuotaManagerConfig.DEFAULT_QUOTA_WINDOW_SIZE_SECONDS * 1000.0))
+ Some(QuotaConfigs.QUOTA_WINDOW_SIZE_SECONDS_DEFAULT * 1000.0))
val exemptMetric = quotaTestClients.exemptRequestMetric
assertNotNull(exemptMetric, "Exempt requests not recorded")
diff --git a/core/src/test/scala/integration/kafka/api/CustomQuotaCallbackTest.scala b/core/src/test/scala/integration/kafka/api/CustomQuotaCallbackTest.scala
index 02e70ef3d47..f0f59f77c88 100644
--- a/core/src/test/scala/integration/kafka/api/CustomQuotaCallbackTest.scala
+++ b/core/src/test/scala/integration/kafka/api/CustomQuotaCallbackTest.scala
@@ -18,7 +18,6 @@ import java.{lang, util}
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.{AtomicBoolean, AtomicInteger}
import java.util.Properties
-
import kafka.api.GroupedUserPrincipalBuilder._
import kafka.api.GroupedUserQuotaCallback._
import kafka.server._
@@ -32,7 +31,7 @@ import org.apache.kafka.common.{Cluster, Reconfigurable}
import org.apache.kafka.common.config.SaslConfigs
import org.apache.kafka.common.network.ListenerName
import org.apache.kafka.common.security.auth._
-import org.apache.kafka.server.config.KafkaSecurityConfigs
+import org.apache.kafka.server.config.{KafkaSecurityConfigs, QuotaConfigs}
import org.apache.kafka.server.quota._
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, BeforeEach, Test, TestInfo}
@@ -63,7 +62,7 @@ class CustomQuotaCallbackTest extends IntegrationTestHarness with SaslSetup {
@BeforeEach
override def setUp(testInfo: TestInfo): Unit = {
startSasl(jaasSections(kafkaServerSaslMechanisms, Some("SCRAM-SHA-256"), KafkaSasl, JaasTestUtils.KafkaServerContextName))
- this.serverConfig.setProperty(KafkaConfig.ClientQuotaCallbackClassProp, classOf[GroupedUserQuotaCallback].getName)
+ this.serverConfig.setProperty(QuotaConfigs.CLIENT_QUOTA_CALLBACK_CLASS_CONFIG, classOf[GroupedUserQuotaCallback].getName)
this.serverConfig.setProperty(s"${listenerName.configPrefix}${KafkaSecurityConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG}",
classOf[GroupedUserPrincipalBuilder].getName)
this.serverConfig.setProperty(KafkaConfig.DeleteTopicEnableProp, "true")
diff --git a/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala
index 5028c948a33..f16c8198133 100644
--- a/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala
@@ -27,7 +27,7 @@ import java.util.{Collections, Optional, Properties}
import java.{time, util}
import kafka.integration.KafkaServerTestHarness
import kafka.server.metadata.KRaftMetadataCache
-import kafka.server.{DynamicConfig, KafkaConfig}
+import kafka.server.KafkaConfig
import kafka.utils.TestUtils._
import kafka.utils.{Log4jController, TestUtils}
import org.apache.kafka.clients.HostResolver
@@ -46,7 +46,7 @@ import org.apache.kafka.controller.ControllerRequestContextUtil.ANONYMOUS_CONTEX
import org.apache.kafka.coordinator.group.GroupCoordinatorConfig
import org.apache.kafka.network.SocketServerConfigs
import org.apache.kafka.security.authorizer.AclEntry
-import org.apache.kafka.server.config.{KafkaSecurityConfigs, ServerLogConfigs, ZkConfigs}
+import org.apache.kafka.server.config.{KafkaSecurityConfigs, ServerLogConfigs, QuotaConfigs, ZkConfigs}
import org.apache.kafka.storage.internals.log.{CleanerConfig, LogConfig}
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, BeforeEach, Disabled, TestInfo}
@@ -1878,7 +1878,7 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
// verify subtract operation, including from an empty property
topic1AlterConfigs = Seq(
new AlterConfigOp(new ConfigEntry(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT), AlterConfigOp.OpType.SUBTRACT),
- new AlterConfigOp(new ConfigEntry(LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, "0"), AlterConfigOp.OpType.SUBTRACT)
+ new AlterConfigOp(new ConfigEntry(QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, "0"), AlterConfigOp.OpType.SUBTRACT)
).asJava
// subtract all from this list property
@@ -1903,7 +1903,7 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
assertEquals("delete", configs.get(topic1Resource).get(TopicConfig.CLEANUP_POLICY_CONFIG).value)
assertEquals("1000", configs.get(topic1Resource).get(TopicConfig.FLUSH_MS_CONFIG).value) // verify previous change is still intact
- assertEquals("", configs.get(topic1Resource).get(LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG).value)
+ assertEquals("", configs.get(topic1Resource).get(QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG).value)
assertEquals("", configs.get(topic2Resource).get(TopicConfig.CLEANUP_POLICY_CONFIG).value )
// Alter topics with validateOnly=true
@@ -1954,12 +1954,12 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
assertNotEquals("", subtractValues)
val topicCreateConfigs = new Properties
- topicCreateConfigs.setProperty(LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, appendValues)
+ topicCreateConfigs.setProperty(QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, appendValues)
createTopic(topic, numPartitions = 1, replicationFactor = 1, topicCreateConfigs)
// Append value that is already present
val topicAppendConfigs = Seq(
- new AlterConfigOp(new ConfigEntry(LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, appendValues), AlterConfigOp.OpType.APPEND),
+ new AlterConfigOp(new ConfigEntry(QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, appendValues), AlterConfigOp.OpType.APPEND),
).asJavaCollection
val appendResult = client.incrementalAlterConfigs(Map(topicResource -> topicAppendConfigs).asJava)
@@ -1967,7 +1967,7 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
// Subtract values that are not present
val topicSubtractConfigs = Seq(
- new AlterConfigOp(new ConfigEntry(LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, subtractValues), AlterConfigOp.OpType.SUBTRACT)
+ new AlterConfigOp(new ConfigEntry(QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, subtractValues), AlterConfigOp.OpType.SUBTRACT)
).asJavaCollection
val subtractResult = client.incrementalAlterConfigs(Map(topicResource -> topicSubtractConfigs).asJava)
subtractResult.all.get
@@ -1978,7 +1978,7 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
val describeResult = client.describeConfigs(Seq(topicResource).asJava)
val configs = describeResult.all.get
- assertEquals(appendValues, configs.get(topicResource).get(LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG).value)
+ assertEquals(appendValues, configs.get(topicResource).get(QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG).value)
}
@ParameterizedTest
@@ -1987,31 +1987,31 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
client = Admin.create(createConfig)
val broker0Resource = new ConfigResource(ConfigResource.Type.BROKER, "0")
client.incrementalAlterConfigs(Map(broker0Resource ->
- Seq(new AlterConfigOp(new ConfigEntry(DynamicConfig.Broker.LeaderReplicationThrottledRateProp, "123"),
+ Seq(new AlterConfigOp(new ConfigEntry(QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG, "123"),
AlterConfigOp.OpType.SET),
- new AlterConfigOp(new ConfigEntry(DynamicConfig.Broker.FollowerReplicationThrottledRateProp, "456"),
+ new AlterConfigOp(new ConfigEntry(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG, "456"),
AlterConfigOp.OpType.SET)
).asJavaCollection).asJava).all().get()
TestUtils.waitUntilTrue(() => {
val broker0Configs = client.describeConfigs(Seq(broker0Resource).asJava).
all().get().get(broker0Resource).entries().asScala.map(entry => (entry.name, entry.value)).toMap
- ("123".equals(broker0Configs.getOrElse(DynamicConfig.Broker.LeaderReplicationThrottledRateProp, "")) &&
- "456".equals(broker0Configs.getOrElse(DynamicConfig.Broker.FollowerReplicationThrottledRateProp, "")))
+ ("123".equals(broker0Configs.getOrElse(QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG, "")) &&
+ "456".equals(broker0Configs.getOrElse(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG, "")))
}, "Expected to see the broker properties we just set", pause=25)
client.incrementalAlterConfigs(Map(broker0Resource ->
- Seq(new AlterConfigOp(new ConfigEntry(DynamicConfig.Broker.LeaderReplicationThrottledRateProp, ""),
+ Seq(new AlterConfigOp(new ConfigEntry(QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG, ""),
AlterConfigOp.OpType.DELETE),
- new AlterConfigOp(new ConfigEntry(DynamicConfig.Broker.FollowerReplicationThrottledRateProp, "654"),
+ new AlterConfigOp(new ConfigEntry(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG, "654"),
AlterConfigOp.OpType.SET),
- new AlterConfigOp(new ConfigEntry(DynamicConfig.Broker.ReplicaAlterLogDirsIoMaxBytesPerSecondProp, "987"),
+ new AlterConfigOp(new ConfigEntry(QuotaConfigs.REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_CONFIG, "987"),
AlterConfigOp.OpType.SET)
).asJavaCollection).asJava).all().get()
TestUtils.waitUntilTrue(() => {
val broker0Configs = client.describeConfigs(Seq(broker0Resource).asJava).
all().get().get(broker0Resource).entries().asScala.map(entry => (entry.name, entry.value)).toMap
- ("".equals(broker0Configs.getOrElse(DynamicConfig.Broker.LeaderReplicationThrottledRateProp, "")) &&
- "654".equals(broker0Configs.getOrElse(DynamicConfig.Broker.FollowerReplicationThrottledRateProp, "")) &&
- "987".equals(broker0Configs.getOrElse(DynamicConfig.Broker.ReplicaAlterLogDirsIoMaxBytesPerSecondProp, "")))
+ ("".equals(broker0Configs.getOrElse(QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG, "")) &&
+ "654".equals(broker0Configs.getOrElse(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG, "")) &&
+ "987".equals(broker0Configs.getOrElse(QuotaConfigs.REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_CONFIG, "")))
}, "Expected to see the broker properties we just modified", pause=25)
}
@@ -2021,34 +2021,34 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
client = Admin.create(createConfig)
val broker0Resource = new ConfigResource(ConfigResource.Type.BROKER, "0")
client.incrementalAlterConfigs(Map(broker0Resource ->
- Seq(new AlterConfigOp(new ConfigEntry(DynamicConfig.Broker.LeaderReplicationThrottledRateProp, "123"),
+ Seq(new AlterConfigOp(new ConfigEntry(QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG, "123"),
AlterConfigOp.OpType.SET),
- new AlterConfigOp(new ConfigEntry(DynamicConfig.Broker.FollowerReplicationThrottledRateProp, "456"),
+ new AlterConfigOp(new ConfigEntry(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG, "456"),
AlterConfigOp.OpType.SET),
- new AlterConfigOp(new ConfigEntry(DynamicConfig.Broker.ReplicaAlterLogDirsIoMaxBytesPerSecondProp, "789"),
+ new AlterConfigOp(new ConfigEntry(QuotaConfigs.REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_CONFIG, "789"),
AlterConfigOp.OpType.SET)
).asJavaCollection).asJava).all().get()
TestUtils.waitUntilTrue(() => {
val broker0Configs = client.describeConfigs(Seq(broker0Resource).asJava).
all().get().get(broker0Resource).entries().asScala.map(entry => (entry.name, entry.value)).toMap
- ("123".equals(broker0Configs.getOrElse(DynamicConfig.Broker.LeaderReplicationThrottledRateProp, "")) &&
- "456".equals(broker0Configs.getOrElse(DynamicConfig.Broker.FollowerReplicationThrottledRateProp, "")) &&
- "789".equals(broker0Configs.getOrElse(DynamicConfig.Broker.ReplicaAlterLogDirsIoMaxBytesPerSecondProp, "")))
+ ("123".equals(broker0Configs.getOrElse(QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG, "")) &&
+ "456".equals(broker0Configs.getOrElse(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG, "")) &&
+ "789".equals(broker0Configs.getOrElse(QuotaConfigs.REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_CONFIG, "")))
}, "Expected to see the broker properties we just set", pause=25)
client.incrementalAlterConfigs(Map(broker0Resource ->
- Seq(new AlterConfigOp(new ConfigEntry(DynamicConfig.Broker.LeaderReplicationThrottledRateProp, ""),
+ Seq(new AlterConfigOp(new ConfigEntry(QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG, ""),
AlterConfigOp.OpType.DELETE),
- new AlterConfigOp(new ConfigEntry(DynamicConfig.Broker.FollowerReplicationThrottledRateProp, ""),
+ new AlterConfigOp(new ConfigEntry(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG, ""),
AlterConfigOp.OpType.DELETE),
- new AlterConfigOp(new ConfigEntry(DynamicConfig.Broker.ReplicaAlterLogDirsIoMaxBytesPerSecondProp, ""),
+ new AlterConfigOp(new ConfigEntry(QuotaConfigs.REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_CONFIG, ""),
AlterConfigOp.OpType.DELETE)
).asJavaCollection).asJava).all().get()
TestUtils.waitUntilTrue(() => {
val broker0Configs = client.describeConfigs(Seq(broker0Resource).asJava).
all().get().get(broker0Resource).entries().asScala.map(entry => (entry.name, entry.value)).toMap
- ("".equals(broker0Configs.getOrElse(DynamicConfig.Broker.LeaderReplicationThrottledRateProp, "")) &&
- "".equals(broker0Configs.getOrElse(DynamicConfig.Broker.FollowerReplicationThrottledRateProp, "")) &&
- "".equals(broker0Configs.getOrElse(DynamicConfig.Broker.ReplicaAlterLogDirsIoMaxBytesPerSecondProp, "")))
+ ("".equals(broker0Configs.getOrElse(QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG, "")) &&
+ "".equals(broker0Configs.getOrElse(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG, "")) &&
+ "".equals(broker0Configs.getOrElse(QuotaConfigs.REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_CONFIG, "")))
}, "Expected to see the broker properties we just removed to be deleted", pause=25)
}
@@ -2495,7 +2495,7 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
@ValueSource(strings = Array("zk", "kraft"))
def testAppendConfigToExistentValue(ignored: String): Unit = {
val props = new Properties();
- props.setProperty(LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, "1:1")
+ props.setProperty(QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, "1:1")
testAppendConfig(props, "0:0", "1:1,0:0")
}
@@ -2504,7 +2504,7 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
createTopic(topic, topicConfig = props)
val topicResource = new ConfigResource(ConfigResource.Type.TOPIC, topic)
val topicAlterConfigs = Seq(
- new AlterConfigOp(new ConfigEntry(LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, append), AlterConfigOp.OpType.APPEND),
+ new AlterConfigOp(new ConfigEntry(QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, append), AlterConfigOp.OpType.APPEND),
).asJavaCollection
val alterResult = client.incrementalAlterConfigs(Map(
@@ -2513,7 +2513,7 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
alterResult.all().get(15, TimeUnit.SECONDS)
ensureConsistentKRaftMetadata()
- val config = client.describeConfigs(List(topicResource).asJava).all().get().get(topicResource).get(LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG)
+ val config = client.describeConfigs(List(topicResource).asJava).all().get().get(topicResource).get(QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG)
assertEquals(expected, config.value())
}
diff --git a/core/src/test/scala/integration/kafka/network/DynamicConnectionQuotaTest.scala b/core/src/test/scala/integration/kafka/network/DynamicConnectionQuotaTest.scala
index 7ff3ea06a8a..b88dc5c099d 100644
--- a/core/src/test/scala/integration/kafka/network/DynamicConnectionQuotaTest.scala
+++ b/core/src/test/scala/integration/kafka/network/DynamicConnectionQuotaTest.scala
@@ -22,10 +22,9 @@ import java.io.IOException
import java.net.{InetAddress, Socket}
import java.util.concurrent._
import java.util.{Collections, Properties}
-import kafka.server.{BaseRequestTest, KafkaConfig}
+import kafka.server.BaseRequestTest
import kafka.utils.TestUtils
import org.apache.kafka.clients.admin.Admin
-import org.apache.kafka.common.config.internals.QuotaConfigs
import org.apache.kafka.common.message.ProduceRequestData
import org.apache.kafka.common.network.ListenerName
import org.apache.kafka.common.protocol.Errors
@@ -35,6 +34,7 @@ import org.apache.kafka.common.requests.{ProduceRequest, ProduceResponse}
import org.apache.kafka.common.security.auth.SecurityProtocol
import org.apache.kafka.common.{KafkaException, requests}
import org.apache.kafka.network.SocketServerConfigs
+import org.apache.kafka.server.config.QuotaConfigs
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, BeforeEach, TestInfo}
import org.junit.jupiter.params.ParameterizedTest
@@ -55,7 +55,7 @@ class DynamicConnectionQuotaTest extends BaseRequestTest {
var admin: Admin = _
override def brokerPropertyOverrides(properties: Properties): Unit = {
- properties.put(KafkaConfig.NumQuotaSamplesProp, "2")
+ properties.put(QuotaConfigs.NUM_QUOTA_SAMPLES_CONFIG, "2")
properties.put("listener.name.plaintext.max.connection.creation.rate", plaintextListenerDefaultQuota.toString)
}
diff --git a/core/src/test/scala/unit/kafka/admin/ReplicationQuotaUtils.scala b/core/src/test/scala/unit/kafka/admin/ReplicationQuotaUtils.scala
index 5e6c6aca1a4..db554a573d7 100644
--- a/core/src/test/scala/unit/kafka/admin/ReplicationQuotaUtils.scala
+++ b/core/src/test/scala/unit/kafka/admin/ReplicationQuotaUtils.scala
@@ -12,11 +12,10 @@
*/
package kafka.admin
-import kafka.server.{DynamicConfig, KafkaServer}
+import kafka.server.KafkaServer
import kafka.utils.TestUtils
import kafka.zk.AdminZkClient
-import org.apache.kafka.server.config.ConfigType
-import org.apache.kafka.storage.internals.log.LogConfig
+import org.apache.kafka.server.config.{ConfigType, QuotaConfigs}
import scala.collection.Seq
@@ -26,12 +25,12 @@ object ReplicationQuotaUtils {
TestUtils.waitUntilTrue(() => {
val hasRateProp = servers.forall { server =>
val brokerConfig = adminZkClient.fetchEntityConfig(ConfigType.BROKER, server.config.brokerId.toString)
- brokerConfig.contains(DynamicConfig.Broker.LeaderReplicationThrottledRateProp) ||
- brokerConfig.contains(DynamicConfig.Broker.FollowerReplicationThrottledRateProp)
+ brokerConfig.contains(QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG) ||
+ brokerConfig.contains(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG)
}
val topicConfig = adminZkClient.fetchEntityConfig(ConfigType.TOPIC, topic)
- val hasReplicasProp = topicConfig.contains(LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG) ||
- topicConfig.contains(LogConfig.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG)
+ val hasReplicasProp = topicConfig.contains(QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG) ||
+ topicConfig.contains(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG)
!hasRateProp && !hasReplicasProp
}, "Throttle limit/replicas was not unset")
}
@@ -41,15 +40,15 @@ object ReplicationQuotaUtils {
//Check for limit in ZK
val brokerConfigAvailable = servers.forall { server =>
val configInZk = adminZkClient.fetchEntityConfig(ConfigType.BROKER, server.config.brokerId.toString)
- val zkLeaderRate = configInZk.getProperty(DynamicConfig.Broker.LeaderReplicationThrottledRateProp)
- val zkFollowerRate = configInZk.getProperty(DynamicConfig.Broker.FollowerReplicationThrottledRateProp)
+ val zkLeaderRate = configInZk.getProperty(QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG)
+ val zkFollowerRate = configInZk.getProperty(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG)
zkLeaderRate != null && expectedThrottleRate == zkLeaderRate.toLong &&
zkFollowerRate != null && expectedThrottleRate == zkFollowerRate.toLong
}
//Check replicas assigned
val topicConfig = adminZkClient.fetchEntityConfig(ConfigType.TOPIC, topic)
- val leader = topicConfig.getProperty(LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG).split(",").toSet
- val follower = topicConfig.getProperty(LogConfig.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG).split(",").toSet
+ val leader = topicConfig.getProperty(QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG).split(",").toSet
+ val follower = topicConfig.getProperty(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG).split(",").toSet
val topicConfigAvailable = leader == throttledLeaders && follower == throttledFollowers
brokerConfigAvailable && topicConfigAvailable
}, "throttle limit/replicas was not set")
diff --git a/core/src/test/scala/unit/kafka/network/ConnectionQuotasTest.scala b/core/src/test/scala/unit/kafka/network/ConnectionQuotasTest.scala
index 50143f3c807..e885d7443bc 100644
--- a/core/src/test/scala/unit/kafka/network/ConnectionQuotasTest.scala
+++ b/core/src/test/scala/unit/kafka/network/ConnectionQuotasTest.scala
@@ -27,13 +27,12 @@ import kafka.server.KafkaConfig
import kafka.utils.Implicits.MapExtensionMethods
import kafka.utils.TestUtils
import org.apache.kafka.common.config.ConfigException
-import org.apache.kafka.common.config.internals.QuotaConfigs
import org.apache.kafka.common.metrics.internals.MetricsUtils
import org.apache.kafka.common.metrics.{KafkaMetric, MetricConfig, Metrics}
import org.apache.kafka.common.network._
import org.apache.kafka.common.utils.Time
import org.apache.kafka.network.SocketServerConfigs
-import org.apache.kafka.server.config.ReplicationConfigs
+import org.apache.kafka.server.config.{ReplicationConfigs, QuotaConfigs}
import org.apache.kafka.server.metrics.KafkaMetricsGroup
import org.apache.kafka.server.util.MockTime
import org.junit.jupiter.api.Assertions._
@@ -73,8 +72,8 @@ class ConnectionQuotasTest {
// ConnectionQuotas does not limit inter-broker listener even when broker-wide connection limit is reached
props.put(ReplicationConfigs.INTER_BROKER_LISTENER_NAME_CONFIG, "REPLICATION")
props.put(SocketServerConfigs.LISTENER_SECURITY_PROTOCOL_MAP_CONFIG, "EXTERNAL:PLAINTEXT,REPLICATION:PLAINTEXT,ADMIN:PLAINTEXT")
- props.put(KafkaConfig.NumQuotaSamplesProp, numQuotaSamples.toString)
- props.put(KafkaConfig.QuotaWindowSizeSecondsProp, quotaWindowSizeSeconds.toString)
+ props.put(QuotaConfigs.NUM_QUOTA_SAMPLES_CONFIG, numQuotaSamples.toString)
+ props.put(QuotaConfigs.QUOTA_WINDOW_SIZE_SECONDS_CONFIG, quotaWindowSizeSeconds.toString)
props
}
diff --git a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
index 53bd126e169..90d7f6d114c 100644
--- a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
+++ b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
@@ -48,6 +48,7 @@ import org.apache.kafka.common.utils.{AppInfoParser, LogContext, MockTime, Time,
import org.apache.kafka.network.SocketServerConfigs
import org.apache.kafka.security.CredentialProvider
import org.apache.kafka.server.common.{Features, MetadataVersion}
+import org.apache.kafka.server.config.QuotaConfigs
import org.apache.kafka.test.{TestSslUtils, TestUtils => JTestUtils}
import org.apache.log4j.Level
import org.junit.jupiter.api.Assertions._
@@ -963,7 +964,7 @@ class SocketServerTest {
val defaultTimeoutMs = 2000
val overrideProps = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 0)
overrideProps.remove(SocketServerConfigs.MAX_CONNECTIONS_PER_IP_CONFIG)
- overrideProps.put(KafkaConfig.NumQuotaSamplesProp, String.valueOf(2))
+ overrideProps.put(QuotaConfigs.NUM_QUOTA_SAMPLES_CONFIG, String.valueOf(2))
val connectionRate = 5
val time = new MockTime()
val overrideServer = new SocketServer(KafkaConfig.fromProps(overrideProps), new Metrics(),
@@ -1014,7 +1015,7 @@ class SocketServerTest {
def testThrottledSocketsClosedOnShutdown(): Unit = {
val overrideProps = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 0)
overrideProps.remove("max.connections.per.ip")
- overrideProps.put(KafkaConfig.NumQuotaSamplesProp, String.valueOf(2))
+ overrideProps.put(QuotaConfigs.NUM_QUOTA_SAMPLES_CONFIG, String.valueOf(2))
val connectionRate = 5
val time = new MockTime()
val overrideServer = new SocketServer(KafkaConfig.fromProps(overrideProps), new Metrics(),
diff --git a/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala b/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala
index e86e613656e..34a33b6eed5 100644
--- a/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala
@@ -25,12 +25,11 @@ import kafka.test.annotation.{ClusterTest, ClusterTestDefaults, Type}
import kafka.test.junit.ClusterTestExtensions
import kafka.utils.TestUtils
import org.apache.kafka.clients.admin.{ScramCredentialInfo, ScramMechanism, UserScramCredentialUpsertion}
-import org.apache.kafka.common.config.internals.QuotaConfigs
import org.apache.kafka.common.errors.{InvalidRequestException, UnsupportedVersionException}
import org.apache.kafka.common.internals.KafkaFutureImpl
import org.apache.kafka.common.quota.{ClientQuotaAlteration, ClientQuotaEntity, ClientQuotaFilter, ClientQuotaFilterComponent}
import org.apache.kafka.common.requests.{AlterClientQuotasRequest, AlterClientQuotasResponse, DescribeClientQuotasRequest, DescribeClientQuotasResponse}
-import org.apache.kafka.server.config.ZooKeeperInternals
+import org.apache.kafka.server.config.{QuotaConfigs, ZooKeeperInternals}
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.Tag
import org.junit.jupiter.api.extension.ExtendWith
@@ -41,11 +40,6 @@ import scala.jdk.CollectionConverters._
@ExtendWith(value = Array(classOf[ClusterTestExtensions]))
@Tag("integration")
class ClientQuotasRequestTest(cluster: ClusterInstance) {
- private val ConsumerByteRateProp = QuotaConfigs.CONSUMER_BYTE_RATE_OVERRIDE_CONFIG
- private val ProducerByteRateProp = QuotaConfigs.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG
- private val RequestPercentageProp = QuotaConfigs.REQUEST_PERCENTAGE_OVERRIDE_CONFIG
- private val IpConnectionRateProp = QuotaConfigs.IP_CONNECTION_RATE_OVERRIDE_CONFIG
-
@ClusterTest
def testAlterClientQuotasRequest(): Unit = {
@@ -56,63 +50,63 @@ class ClientQuotasRequestTest(cluster: ClusterInstance) {
// Add two configuration entries.
alterEntityQuotas(entity, Map(
- (ProducerByteRateProp -> Some(10000.0)),
- (ConsumerByteRateProp -> Some(20000.0))
+ (QuotaConfigs.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG -> Some(10000.0)),
+ (QuotaConfigs.CONSUMER_BYTE_RATE_OVERRIDE_CONFIG -> Some(20000.0))
), validateOnly = false)
verifyDescribeEntityQuotas(entity, Map(
- (ProducerByteRateProp -> 10000.0),
- (ConsumerByteRateProp -> 20000.0)
+ (QuotaConfigs.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG -> 10000.0),
+ (QuotaConfigs.CONSUMER_BYTE_RATE_OVERRIDE_CONFIG -> 20000.0)
))
// Update an existing entry.
alterEntityQuotas(entity, Map(
- (ProducerByteRateProp -> Some(15000.0))
+ (QuotaConfigs.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG -> Some(15000.0))
), validateOnly = false)
verifyDescribeEntityQuotas(entity, Map(
- (ProducerByteRateProp -> 15000.0),
- (ConsumerByteRateProp -> 20000.0)
+ (QuotaConfigs.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG -> 15000.0),
+ (QuotaConfigs.CONSUMER_BYTE_RATE_OVERRIDE_CONFIG -> 20000.0)
))
// Remove an existing configuration entry.
alterEntityQuotas(entity, Map(
- (ProducerByteRateProp -> None)
+ (QuotaConfigs.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG -> None)
), validateOnly = false)
verifyDescribeEntityQuotas(entity, Map(
- (ConsumerByteRateProp -> 20000.0)
+ (QuotaConfigs.CONSUMER_BYTE_RATE_OVERRIDE_CONFIG -> 20000.0)
))
// Remove a non-existent configuration entry. This should make no changes.
alterEntityQuotas(entity, Map(
- (RequestPercentageProp -> None)
+ (QuotaConfigs.REQUEST_PERCENTAGE_OVERRIDE_CONFIG -> None)
), validateOnly = false)
verifyDescribeEntityQuotas(entity, Map(
- (ConsumerByteRateProp -> 20000.0)
+ (QuotaConfigs.CONSUMER_BYTE_RATE_OVERRIDE_CONFIG -> 20000.0)
))
// Add back a deleted configuration entry.
alterEntityQuotas(entity, Map(
- (ProducerByteRateProp -> Some(5000.0))
+ (QuotaConfigs.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG -> Some(5000.0))
), validateOnly = false)
verifyDescribeEntityQuotas(entity, Map(
- (ProducerByteRateProp -> 5000.0),
- (ConsumerByteRateProp -> 20000.0)
+ (QuotaConfigs.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG -> 5000.0),
+ (QuotaConfigs.CONSUMER_BYTE_RATE_OVERRIDE_CONFIG -> 20000.0)
))
// Perform a mixed update.
alterEntityQuotas(entity, Map(
- (ProducerByteRateProp -> Some(20000.0)),
- (ConsumerByteRateProp -> None),
- (RequestPercentageProp -> Some(12.3))
+ (QuotaConfigs.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG -> Some(20000.0)),
+ (QuotaConfigs.CONSUMER_BYTE_RATE_OVERRIDE_CONFIG -> None),
+ (QuotaConfigs.REQUEST_PERCENTAGE_OVERRIDE_CONFIG -> Some(12.3))
), validateOnly = false)
verifyDescribeEntityQuotas(entity, Map(
- (ProducerByteRateProp -> 20000.0),
- (RequestPercentageProp -> 12.3)
+ (QuotaConfigs.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG -> 20000.0),
+ (QuotaConfigs.REQUEST_PERCENTAGE_OVERRIDE_CONFIG -> 12.3)
))
}
@@ -122,55 +116,55 @@ class ClientQuotasRequestTest(cluster: ClusterInstance) {
// Set up a configuration.
alterEntityQuotas(entity, Map(
- (ProducerByteRateProp -> Some(20000.0)),
- (RequestPercentageProp -> Some(23.45))
+ (QuotaConfigs.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG -> Some(20000.0)),
+ (QuotaConfigs.REQUEST_PERCENTAGE_OVERRIDE_CONFIG -> Some(23.45))
), validateOnly = false)
verifyDescribeEntityQuotas(entity, Map(
- (ProducerByteRateProp -> 20000.0),
- (RequestPercentageProp -> 23.45)
+ (QuotaConfigs.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG -> 20000.0),
+ (QuotaConfigs.REQUEST_PERCENTAGE_OVERRIDE_CONFIG -> 23.45)
))
// Validate-only addition.
alterEntityQuotas(entity, Map(
- (ConsumerByteRateProp -> Some(50000.0))
+ (QuotaConfigs.CONSUMER_BYTE_RATE_OVERRIDE_CONFIG -> Some(50000.0))
), validateOnly = true)
verifyDescribeEntityQuotas(entity, Map(
- (ProducerByteRateProp -> 20000.0),
- (RequestPercentageProp -> 23.45)
+ (QuotaConfigs.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG -> 20000.0),
+ (QuotaConfigs.REQUEST_PERCENTAGE_OVERRIDE_CONFIG -> 23.45)
))
// Validate-only modification.
alterEntityQuotas(entity, Map(
- (ProducerByteRateProp -> Some(10000.0))
+ (QuotaConfigs.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG -> Some(10000.0))
), validateOnly = true)
verifyDescribeEntityQuotas(entity, Map(
- (ProducerByteRateProp -> 20000.0),
- (RequestPercentageProp -> 23.45)
+ (QuotaConfigs.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG -> 20000.0),
+ (QuotaConfigs.REQUEST_PERCENTAGE_OVERRIDE_CONFIG -> 23.45)
))
// Validate-only removal.
alterEntityQuotas(entity, Map(
- (RequestPercentageProp -> None)
+ (QuotaConfigs.REQUEST_PERCENTAGE_OVERRIDE_CONFIG -> None)
), validateOnly = true)
verifyDescribeEntityQuotas(entity, Map(
- (ProducerByteRateProp -> 20000.0),
- (RequestPercentageProp -> 23.45)
+ (QuotaConfigs.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG -> 20000.0),
+ (QuotaConfigs.REQUEST_PERCENTAGE_OVERRIDE_CONFIG -> 23.45)
))
// Validate-only mixed update.
alterEntityQuotas(entity, Map(
- (ProducerByteRateProp -> Some(10000.0)),
- (ConsumerByteRateProp -> Some(50000.0)),
- (RequestPercentageProp -> None)
+ (QuotaConfigs.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG -> Some(10000.0)),
+ (QuotaConfigs.CONSUMER_BYTE_RATE_OVERRIDE_CONFIG -> Some(50000.0)),
+ (QuotaConfigs.REQUEST_PERCENTAGE_OVERRIDE_CONFIG -> None)
), validateOnly = true)
verifyDescribeEntityQuotas(entity, Map(
- (ProducerByteRateProp -> 20000.0),
- (RequestPercentageProp -> 23.45)
+ (QuotaConfigs.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG -> 20000.0),
+ (QuotaConfigs.REQUEST_PERCENTAGE_OVERRIDE_CONFIG -> 23.45)
))
}
@@ -187,13 +181,13 @@ class ClientQuotasRequestTest(cluster: ClusterInstance) {
verifyDescribeEntityQuotas(entity, Map.empty)
alterEntityQuotas(entity, Map(
- (ProducerByteRateProp -> Some(10000.0)),
- (ConsumerByteRateProp -> Some(20000.0))
+ (QuotaConfigs.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG -> Some(10000.0)),
+ (QuotaConfigs.CONSUMER_BYTE_RATE_OVERRIDE_CONFIG -> Some(20000.0))
), validateOnly = false)
verifyDescribeEntityQuotas(entity, Map(
- (ProducerByteRateProp -> 10000.0),
- (ConsumerByteRateProp -> 20000.0)
+ (QuotaConfigs.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG -> 10000.0),
+ (QuotaConfigs.CONSUMER_BYTE_RATE_OVERRIDE_CONFIG -> 20000.0)
))
}
@@ -212,8 +206,8 @@ class ClientQuotasRequestTest(cluster: ClusterInstance) {
val result = describeClientQuotas(ClientQuotaFilter.containsOnly(List(entityFilter).asJava))
assertEquals(expectedMatches.keySet, result.asScala.keySet)
result.asScala.foreach { case (entity, props) =>
- assertEquals(Set(IpConnectionRateProp), props.asScala.keySet)
- assertEquals(expectedMatches(entity), props.get(IpConnectionRateProp))
+ assertEquals(Set(QuotaConfigs.IP_CONNECTION_RATE_OVERRIDE_CONFIG), props.asScala.keySet)
+ assertEquals(expectedMatches(entity), props.get(QuotaConfigs.IP_CONNECTION_RATE_OVERRIDE_CONFIG))
val entityName = entity.entries.get(ClientQuotaEntity.IP)
// ClientQuotaEntity with null name maps to default entity
val entityIp = if (entityName == null)
@@ -232,48 +226,48 @@ class ClientQuotasRequestTest(cluster: ClusterInstance) {
verifyIpQuotas(allIpEntityFilter, Map.empty)
// Add a configuration entry.
- alterEntityQuotas(entity, Map(IpConnectionRateProp -> Some(100.0)), validateOnly = false)
+ alterEntityQuotas(entity, Map(QuotaConfigs.IP_CONNECTION_RATE_OVERRIDE_CONFIG -> Some(100.0)), validateOnly = false)
verifyIpQuotas(entityFilter, Map(entity -> 100.0))
// update existing entry
- alterEntityQuotas(entity, Map(IpConnectionRateProp -> Some(150.0)), validateOnly = false)
+ alterEntityQuotas(entity, Map(QuotaConfigs.IP_CONNECTION_RATE_OVERRIDE_CONFIG -> Some(150.0)), validateOnly = false)
verifyIpQuotas(entityFilter, Map(entity -> 150.0))
// update default value
- alterEntityQuotas(defaultEntity, Map(IpConnectionRateProp -> Some(200.0)), validateOnly = false)
+ alterEntityQuotas(defaultEntity, Map(QuotaConfigs.IP_CONNECTION_RATE_OVERRIDE_CONFIG -> Some(200.0)), validateOnly = false)
verifyIpQuotas(defaultEntityFilter, Map(defaultEntity -> 200.0))
// describe all IP quotas
verifyIpQuotas(allIpEntityFilter, Map(entity -> 150.0, defaultEntity -> 200.0))
// remove entry
- alterEntityQuotas(entity, Map(IpConnectionRateProp -> None), validateOnly = false)
+ alterEntityQuotas(entity, Map(QuotaConfigs.IP_CONNECTION_RATE_OVERRIDE_CONFIG -> None), validateOnly = false)
verifyIpQuotas(entityFilter, Map.empty)
// remove default value
- alterEntityQuotas(defaultEntity, Map(IpConnectionRateProp -> None), validateOnly = false)
+ alterEntityQuotas(defaultEntity, Map(QuotaConfigs.IP_CONNECTION_RATE_OVERRIDE_CONFIG -> None), validateOnly = false)
verifyIpQuotas(allIpEntityFilter, Map.empty)
}
@ClusterTest
def testAlterClientQuotasInvalidRequests(): Unit = {
var entity = new ClientQuotaEntity(Map((ClientQuotaEntity.USER -> "")).asJava)
- assertThrows(classOf[InvalidRequestException], () => alterEntityQuotas(entity, Map((RequestPercentageProp -> Some(12.34))), validateOnly = true))
+ assertThrows(classOf[InvalidRequestException], () => alterEntityQuotas(entity, Map((QuotaConfigs.REQUEST_PERCENTAGE_OVERRIDE_CONFIG -> Some(12.34))), validateOnly = true))
entity = new ClientQuotaEntity(Map((ClientQuotaEntity.CLIENT_ID -> "")).asJava)
- assertThrows(classOf[InvalidRequestException], () => alterEntityQuotas(entity, Map((RequestPercentageProp -> Some(12.34))), validateOnly = true))
+ assertThrows(classOf[InvalidRequestException], () => alterEntityQuotas(entity, Map((QuotaConfigs.REQUEST_PERCENTAGE_OVERRIDE_CONFIG -> Some(12.34))), validateOnly = true))
entity = new ClientQuotaEntity(Map(("" -> "name")).asJava)
- assertThrows(classOf[InvalidRequestException], () => alterEntityQuotas(entity, Map((RequestPercentageProp -> Some(12.34))), validateOnly = true))
+ assertThrows(classOf[InvalidRequestException], () => alterEntityQuotas(entity, Map((QuotaConfigs.REQUEST_PERCENTAGE_OVERRIDE_CONFIG -> Some(12.34))), validateOnly = true))
entity = new ClientQuotaEntity(Map.empty.asJava)
- assertThrows(classOf[InvalidRequestException], () => alterEntityQuotas(entity, Map((ProducerByteRateProp -> Some(10000.5))), validateOnly = true))
+ assertThrows(classOf[InvalidRequestException], () => alterEntityQuotas(entity, Map((QuotaConfigs.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG -> Some(10000.5))), validateOnly = true))
entity = new ClientQuotaEntity(Map((ClientQuotaEntity.USER -> "user")).asJava)
assertThrows(classOf[InvalidRequestException], () => alterEntityQuotas(entity, Map(("bad" -> Some(1.0))), validateOnly = true))
entity = new ClientQuotaEntity(Map((ClientQuotaEntity.USER -> "user")).asJava)
- assertThrows(classOf[InvalidRequestException], () => alterEntityQuotas(entity, Map((ProducerByteRateProp -> Some(10000.5))), validateOnly = true))
+ assertThrows(classOf[InvalidRequestException], () => alterEntityQuotas(entity, Map((QuotaConfigs.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG -> Some(10000.5))), validateOnly = true))
}
private def expectInvalidRequestWithMessage(runnable: => Unit, expectedMessage: String): Unit = {
@@ -286,9 +280,9 @@ class ClientQuotasRequestTest(cluster: ClusterInstance) {
val userAndIpEntity = new ClientQuotaEntity(Map(ClientQuotaEntity.USER -> "user", ClientQuotaEntity.IP -> "1.2.3.4").asJava)
val clientAndIpEntity = new ClientQuotaEntity(Map(ClientQuotaEntity.CLIENT_ID -> "client", ClientQuotaEntity.IP -> "1.2.3.4").asJava)
val expectedExceptionMessage = "Invalid quota entity combination"
- expectInvalidRequestWithMessage(alterEntityQuotas(userAndIpEntity, Map(RequestPercentageProp -> Some(12.34)),
+ expectInvalidRequestWithMessage(alterEntityQuotas(userAndIpEntity, Map(QuotaConfigs.REQUEST_PERCENTAGE_OVERRIDE_CONFIG -> Some(12.34)),
validateOnly = true), expectedExceptionMessage)
- expectInvalidRequestWithMessage(alterEntityQuotas(clientAndIpEntity, Map(RequestPercentageProp -> Some(12.34)),
+ expectInvalidRequestWithMessage(alterEntityQuotas(clientAndIpEntity, Map(QuotaConfigs.REQUEST_PERCENTAGE_OVERRIDE_CONFIG -> Some(12.34)),
validateOnly = true), expectedExceptionMessage)
}
@@ -297,9 +291,9 @@ class ClientQuotasRequestTest(cluster: ClusterInstance) {
val invalidHostPatternEntity = new ClientQuotaEntity(Map(ClientQuotaEntity.IP -> "not a valid host because it has spaces").asJava)
val unresolvableHostEntity = new ClientQuotaEntity(Map(ClientQuotaEntity.IP -> "RFC2606.invalid").asJava)
val expectedExceptionMessage = "not a valid IP"
- expectInvalidRequestWithMessage(alterEntityQuotas(invalidHostPatternEntity, Map(IpConnectionRateProp -> Some(50.0)),
+ expectInvalidRequestWithMessage(alterEntityQuotas(invalidHostPatternEntity, Map(QuotaConfigs.IP_CONNECTION_RATE_OVERRIDE_CONFIG -> Some(50.0)),
validateOnly = true), expectedExceptionMessage)
- expectInvalidRequestWithMessage(alterEntityQuotas(unresolvableHostEntity, Map(IpConnectionRateProp -> Some(50.0)),
+ expectInvalidRequestWithMessage(alterEntityQuotas(unresolvableHostEntity, Map(QuotaConfigs.IP_CONNECTION_RATE_OVERRIDE_CONFIG -> Some(50.0)),
validateOnly = true), expectedExceptionMessage)
}
@@ -337,10 +331,10 @@ class ClientQuotasRequestTest(cluster: ClusterInstance) {
private def setupDescribeClientQuotasMatchTest() = {
val userClientQuotas = matchUserClientEntities.map { case (e, v) =>
- e -> Map((RequestPercentageProp, Some(v)))
+ e -> Map((QuotaConfigs.REQUEST_PERCENTAGE_OVERRIDE_CONFIG, Some(v)))
}.toMap
val ipQuotas = matchIpEntities.map { case (e, v) =>
- e -> Map((IpConnectionRateProp, Some(v)))
+ e -> Map((QuotaConfigs.IP_CONNECTION_RATE_OVERRIDE_CONFIG, Some(v)))
}.toMap
val result = alterClientQuotas(userClientQuotas ++ ipQuotas, validateOnly = false)
(matchUserClientEntities ++ matchIpEntities).foreach(e => result(e._1).get(10, TimeUnit.SECONDS))
@@ -366,7 +360,7 @@ class ClientQuotasRequestTest(cluster: ClusterInstance) {
val result = matchEntity(e)
assertEquals(1, result.size)
assertTrue(result.get(e) != null)
- val value = result.get(e).get(RequestPercentageProp)
+ val value = result.get(e).get(QuotaConfigs.REQUEST_PERCENTAGE_OVERRIDE_CONFIG)
assertNotNull(value)
assertEquals(value, v, 1e-6)
}
@@ -408,7 +402,7 @@ class ClientQuotasRequestTest(cluster: ClusterInstance) {
if (expectedMatchesMap.contains(entity)) {
val config = result.get(entity)
assertNotNull(config)
- val value = config.get(RequestPercentageProp)
+ val value = config.get(QuotaConfigs.REQUEST_PERCENTAGE_OVERRIDE_CONFIG)
assertNotNull(value)
assertEquals(expectedValue, value, 1e-6)
} else {
@@ -419,7 +413,7 @@ class ClientQuotasRequestTest(cluster: ClusterInstance) {
if (expectedMatchesMap.contains(entity)) {
val config = result.get(entity)
assertNotNull(config)
- val value = config.get(IpConnectionRateProp)
+ val value = config.get(QuotaConfigs.IP_CONNECTION_RATE_OVERRIDE_CONFIG)
assertNotNull(value)
assertEquals(expectedValue, value, 1e-6)
} else {
@@ -514,11 +508,11 @@ class ClientQuotasRequestTest(cluster: ClusterInstance) {
val entity = new ClientQuotaEntity(Map((ClientQuotaEntity.USER -> "user with spaces")).asJava)
alterEntityQuotas(entity, Map(
- (ProducerByteRateProp -> Some(20000.0)),
+ (QuotaConfigs.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG -> Some(20000.0)),
), validateOnly = false)
verifyDescribeEntityQuotas(entity, Map(
- (ProducerByteRateProp -> 20000.0),
+ (QuotaConfigs.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG -> 20000.0),
))
}
@@ -527,8 +521,8 @@ class ClientQuotasRequestTest(cluster: ClusterInstance) {
// An entity using the name associated with the default entity name. The entity's name should be sanitized so
// that it does not conflict with the default entity name.
val entity = new ClientQuotaEntity(Map((ClientQuotaEntity.CLIENT_ID -> ZooKeeperInternals.DEFAULT_STRING)).asJava)
- alterEntityQuotas(entity, Map((ProducerByteRateProp -> Some(20000.0))), validateOnly = false)
- verifyDescribeEntityQuotas(entity, Map((ProducerByteRateProp -> 20000.0)))
+ alterEntityQuotas(entity, Map((QuotaConfigs.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG -> Some(20000.0))), validateOnly = false)
+ verifyDescribeEntityQuotas(entity, Map((QuotaConfigs.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG -> 20000.0)))
// This should not match.
val result = describeClientQuotas(
diff --git a/core/src/test/scala/unit/kafka/server/ControllerMutationQuotaTest.scala b/core/src/test/scala/unit/kafka/server/ControllerMutationQuotaTest.scala
index 6b329a7af0a..80bade70c1a 100644
--- a/core/src/test/scala/unit/kafka/server/ControllerMutationQuotaTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ControllerMutationQuotaTest.scala
@@ -18,7 +18,6 @@ import java.util.concurrent.ExecutionException
import java.util.concurrent.TimeUnit
import kafka.server.ClientQuotaManager.DefaultTags
import kafka.utils.TestUtils
-import org.apache.kafka.common.config.internals.QuotaConfigs
import org.apache.kafka.common.internals.KafkaFutureImpl
import org.apache.kafka.common.message.CreatePartitionsRequestData
import org.apache.kafka.common.message.CreatePartitionsRequestData.CreatePartitionsTopic
@@ -41,7 +40,7 @@ import org.apache.kafka.common.requests.DeleteTopicsResponse
import org.apache.kafka.common.security.auth.AuthenticationContext
import org.apache.kafka.common.security.auth.KafkaPrincipal
import org.apache.kafka.common.security.authenticator.DefaultKafkaPrincipalBuilder
-import org.apache.kafka.server.config.KafkaSecurityConfigs
+import org.apache.kafka.server.config.{KafkaSecurityConfigs, QuotaConfigs}
import org.apache.kafka.coordinator.group.GroupCoordinatorConfig
import org.apache.kafka.test.{TestUtils => JTestUtils}
import org.junit.jupiter.api.Assertions.assertEquals
@@ -102,8 +101,8 @@ class ControllerMutationQuotaTest extends BaseRequestTest {
properties.put(KafkaSecurityConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG,
classOf[ControllerMutationQuotaTest.TestPrincipalBuilder].getName)
// Specify number of samples and window size.
- properties.put(KafkaConfig.NumControllerQuotaSamplesProp, ControllerQuotaSamples.toString)
- properties.put(KafkaConfig.ControllerQuotaWindowSizeSecondsProp, ControllerQuotaWindowSizeSeconds.toString)
+ properties.put(QuotaConfigs.NUM_CONTROLLER_QUOTA_SAMPLES_CONFIG, ControllerQuotaSamples.toString)
+ properties.put(QuotaConfigs.CONTROLLER_QUOTA_WINDOW_SIZE_SECONDS_CONFIG, ControllerQuotaWindowSizeSeconds.toString)
}
@BeforeEach
diff --git a/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala b/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala
index 921ead3de81..2c55402fc3d 100644
--- a/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala
+++ b/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala
@@ -35,7 +35,6 @@ import org.apache.kafka.clients.admin.AlterConfigOp.OpType.SET
import org.apache.kafka.clients.admin.{Admin, AlterConfigOp, Config, ConfigEntry}
import org.apache.kafka.common.{TopicPartition, Uuid}
import org.apache.kafka.common.config.{ConfigResource, TopicConfig}
-import org.apache.kafka.common.config.internals.QuotaConfigs
import org.apache.kafka.common.errors.{InvalidRequestException, UnknownTopicOrPartitionException}
import org.apache.kafka.common.metrics.Quota
import org.apache.kafka.common.quota.ClientQuotaAlteration.Op
@@ -44,8 +43,7 @@ import org.apache.kafka.common.quota.{ClientQuotaAlteration, ClientQuotaEntity}
import org.apache.kafka.common.record.{CompressionType, RecordVersion}
import org.apache.kafka.common.security.auth.KafkaPrincipal
import org.apache.kafka.server.common.MetadataVersion.IBP_3_0_IV1
-import org.apache.kafka.server.config.{ConfigType, ServerLogConfigs, ZooKeeperInternals}
-import org.apache.kafka.storage.internals.log.LogConfig
+import org.apache.kafka.server.config.{ConfigType, ServerLogConfigs, QuotaConfigs, ZooKeeperInternals}
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{Test, Timeout}
import org.junit.jupiter.params.ParameterizedTest
@@ -494,11 +492,11 @@ class DynamicConfigChangeUnitTest {
val props: Properties = new Properties()
//Given
- props.put(LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, "0:101,0:102,1:101,1:102")
+ props.put(QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, "0:101,0:102,1:101,1:102")
//When/Then
- assertEquals(Seq(0,1), configHandler.parseThrottledPartitions(props, 102, LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG))
- assertEquals(Seq(), configHandler.parseThrottledPartitions(props, 103, LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG))
+ assertEquals(Seq(0,1), configHandler.parseThrottledPartitions(props, 102, QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG))
+ assertEquals(Seq(), configHandler.parseThrottledPartitions(props, 103, QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG))
}
@Test
@@ -507,10 +505,10 @@ class DynamicConfigChangeUnitTest {
val props: Properties = new Properties()
//Given
- props.put(LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, "*")
+ props.put(QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, "*")
//When
- val result = configHandler.parseThrottledPartitions(props, 102, LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG)
+ val result = configHandler.parseThrottledPartitions(props, 102, QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG)
//Then
assertEquals(AllReplicas, result)
@@ -520,8 +518,8 @@ class DynamicConfigChangeUnitTest {
def shouldParseRegardlessOfWhitespaceAroundValues(): Unit = {
def parse(configHandler: TopicConfigHandler, value: String): Seq[Int] = {
configHandler.parseThrottledPartitions(
- CoreUtils.propsWith(LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, value),
- 102, LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG)
+ CoreUtils.propsWith(QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, value),
+ 102, QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG)
}
val configHandler: TopicConfigHandler = new TopicConfigHandler(null, null, null, null)
assertEquals(AllReplicas, parse(configHandler, "* "))
@@ -537,10 +535,10 @@ class DynamicConfigChangeUnitTest {
val props: Properties = new Properties()
//Given
- props.put(LogConfig.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG, "")
+ props.put(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG, "")
//When
- val result = configHandler.parseThrottledPartitions(props, 102, LogConfig.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG)
+ val result = configHandler.parseThrottledPartitions(props, 102, QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG)
//Then
assertEquals(Seq(), result)
diff --git a/core/src/test/scala/unit/kafka/server/DynamicConfigTest.scala b/core/src/test/scala/unit/kafka/server/DynamicConfigTest.scala
index ddcb6377357..47f68358994 100644
--- a/core/src/test/scala/unit/kafka/server/DynamicConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/server/DynamicConfigTest.scala
@@ -17,10 +17,9 @@
package kafka.server
import kafka.utils.CoreUtils._
-import kafka.server.QuorumTestHarness
import org.apache.kafka.common.config._
-import org.apache.kafka.common.config.internals.QuotaConfigs
import org.apache.kafka.server.common.AdminOperationException
+import org.apache.kafka.server.config.QuotaConfigs
import org.junit.jupiter.api.Assertions.assertThrows
import org.junit.jupiter.api.Test
@@ -43,13 +42,13 @@ class DynamicConfigTest extends QuorumTestHarness {
@Test
def shouldFailLeaderConfigsWithInvalidValues(): Unit = {
assertThrows(classOf[ConfigException], () => adminZkClient.changeBrokerConfig(Seq(0),
- propsWith(DynamicConfig.Broker.LeaderReplicationThrottledRateProp, "-100")))
+ propsWith(QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG, "-100")))
}
@Test
def shouldFailFollowerConfigsWithInvalidValues(): Unit = {
assertThrows(classOf[ConfigException], () => adminZkClient.changeBrokerConfig(Seq(0),
- propsWith(DynamicConfig.Broker.FollowerReplicationThrottledRateProp, "-100")))
+ propsWith(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG, "-100")))
}
@Test
diff --git a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
index bf1817acfb6..11fad5eaa1d 100755
--- a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
@@ -43,9 +43,9 @@ import org.apache.kafka.security.PasswordEncoderConfigs
import org.apache.kafka.network.SocketServerConfigs
import org.apache.kafka.server.common.MetadataVersion
import org.apache.kafka.server.common.MetadataVersion.{IBP_0_8_2, IBP_3_0_IV1}
-import org.apache.kafka.server.config.{KafkaSecurityConfigs, ReplicationConfigs, ServerLogConfigs, ServerTopicConfigSynonyms, ZkConfigs}
+import org.apache.kafka.server.config.{KafkaSecurityConfigs, ReplicationConfigs, ServerLogConfigs, QuotaConfigs, ServerTopicConfigSynonyms, ZkConfigs}
import org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig
-import org.apache.kafka.storage.internals.log.{CleanerConfig, LogConfig}
+import org.apache.kafka.storage.internals.log.CleanerConfig
import org.junit.jupiter.api.function.Executable
import scala.annotation.nowarn
@@ -912,7 +912,7 @@ class KafkaConfigTest {
case TransactionLogConfigs.TRANSACTIONS_TOPIC_PARTITIONS_CONFIG => assertPropertyInvalid(baseProperties, name, "not_a_number", "0", "-2")
case TransactionLogConfigs.TRANSACTIONS_TOPIC_SEGMENT_BYTES_CONFIG => assertPropertyInvalid(baseProperties, name, "not_a_number", "0", "-2")
case TransactionLogConfigs.TRANSACTIONS_TOPIC_REPLICATION_FACTOR_CONFIG => assertPropertyInvalid(baseProperties, name, "not_a_number", "0", "-2")
- case KafkaConfig.QuotaWindowSizeSecondsProp => assertPropertyInvalid(baseProperties, name, "not_a_number", "0")
+ case QuotaConfigs.QUOTA_WINDOW_SIZE_SECONDS_CONFIG => assertPropertyInvalid(baseProperties, name, "not_a_number", "0")
case KafkaConfig.DeleteTopicEnableProp => assertPropertyInvalid(baseProperties, name, "not_a_boolean", "0")
case KafkaConfig.MetricNumSamplesProp => assertPropertyInvalid(baseProperties, name, "not_a_number", "-1", "0")
@@ -1132,9 +1132,9 @@ class KafkaConfigTest {
assertDynamic(kafkaConfigProp, 10016L, () => config.logLocalRetentionBytes)
case TopicConfig.MESSAGE_FORMAT_VERSION_CONFIG =>
// not dynamically updatable
- case LogConfig.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG =>
+ case QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG =>
// topic only config
- case LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG =>
+ case QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG =>
// topic only config
case prop =>
fail(prop + " must be explicitly checked for dynamic updatability. Note that LogConfig(s) require that KafkaConfig value lookups are dynamic and not static values.")
diff --git a/core/src/test/scala/unit/kafka/server/KafkaMetricReporterExceptionHandlingTest.scala b/core/src/test/scala/unit/kafka/server/KafkaMetricReporterExceptionHandlingTest.scala
index 5b7509a2a3e..bbcfa5171d7 100644
--- a/core/src/test/scala/unit/kafka/server/KafkaMetricReporterExceptionHandlingTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaMetricReporterExceptionHandlingTest.scala
@@ -15,13 +15,13 @@
package kafka.server
import kafka.utils.TestUtils
-import org.apache.kafka.common.config.internals.QuotaConfigs
import org.apache.kafka.common.message.ListGroupsRequestData
import org.apache.kafka.common.metrics.{KafkaMetric, MetricsReporter}
import org.apache.kafka.common.network.ListenerName
import org.apache.kafka.common.protocol.Errors
import org.apache.kafka.common.requests.{ListGroupsRequest, ListGroupsResponse}
import org.apache.kafka.common.security.auth.SecurityProtocol
+import org.apache.kafka.server.config.QuotaConfigs
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, BeforeEach, TestInfo}
import org.junit.jupiter.params.ParameterizedTest
diff --git a/core/src/test/scala/unit/kafka/server/ReplicationQuotasTest.scala b/core/src/test/scala/unit/kafka/server/ReplicationQuotasTest.scala
index d9db91dda1f..ec59053027a 100644
--- a/core/src/test/scala/unit/kafka/server/ReplicationQuotasTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ReplicationQuotasTest.scala
@@ -20,7 +20,6 @@ package kafka.server
import java.util.AbstractMap.SimpleImmutableEntry
import java.util.{Collections, Properties}
import java.util.Map.Entry
-import kafka.server.DynamicConfig.Broker
import kafka.server.KafkaConfig.fromProps
import kafka.server.QuotaType._
import kafka.utils.TestUtils._
@@ -38,7 +37,7 @@ import org.apache.kafka.common.network.ListenerName
import org.apache.kafka.common.security.auth.SecurityProtocol.PLAINTEXT
import org.apache.kafka.controller.ControllerRequestContextUtil
import org.apache.kafka.server.common.MetadataVersion
-import org.apache.kafka.storage.internals.log.LogConfig
+import org.apache.kafka.server.config.QuotaConfigs
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.AfterEach
import org.junit.jupiter.params.ParameterizedTest
@@ -127,15 +126,15 @@ class ReplicationQuotasTest extends QuorumTestHarness {
controllerServer.controller.incrementalAlterConfigs(
ControllerRequestContextUtil.ANONYMOUS_CONTEXT,
Map(new ConfigResource(BROKER, String.valueOf(brokerId)) -> Map(
- Broker.LeaderReplicationThrottledRateProp -> entry,
- Broker.FollowerReplicationThrottledRateProp -> entry).asJava).asJava,
+ QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG -> entry,
+ QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG -> entry).asJava).asJava,
false
).get()
} else {
adminZkClient.changeBrokerConfig(Seq(brokerId),
propsWith(
- (DynamicConfig.Broker.LeaderReplicationThrottledRateProp, throttle.toString),
- (DynamicConfig.Broker.FollowerReplicationThrottledRateProp, throttle.toString)
+ (QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG, throttle.toString),
+ (QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG, throttle.toString)
))
}
}
@@ -143,13 +142,13 @@ class ReplicationQuotasTest extends QuorumTestHarness {
if (leaderThrottle) {
admin.incrementalAlterConfigs(
Map(new ConfigResource(TOPIC, topic) -> Seq(new AlterConfigOp(new ConfigEntry(
- LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, "0:100,1:101,2:102,3:103,4:104,5:105"),
+ QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, "0:100,1:101,2:102,3:103,4:104,5:105"),
SET)).asJavaCollection).asJava
).all().get()
} else {
admin.incrementalAlterConfigs(
Map(new ConfigResource(TOPIC, topic) -> Seq(new AlterConfigOp(new ConfigEntry(
- LogConfig.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG, "0:106,1:106,2:106,3:107,4:107,5:107"),
+ QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG, "0:106,1:106,2:106,3:107,4:107,5:107"),
SET)).asJavaCollection).asJava
).all().get()
}
@@ -239,9 +238,9 @@ class ReplicationQuotasTest extends QuorumTestHarness {
//Set the throttle to only limit leader
val configs = Map(
new ConfigResource(BROKER, "100") ->
- Seq(new AlterConfigOp(new ConfigEntry(Broker.LeaderReplicationThrottledRateProp, throttle.toString), SET)).asJavaCollection,
+ Seq(new AlterConfigOp(new ConfigEntry(QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG, throttle.toString), SET)).asJavaCollection,
new ConfigResource(TOPIC, topic) ->
- Seq(new AlterConfigOp(new ConfigEntry(LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, "0:100"), SET)).asJavaCollection
+ Seq(new AlterConfigOp(new ConfigEntry(QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, "0:100"), SET)).asJavaCollection
).asJava
admin.incrementalAlterConfigs(configs).all().get()
}
diff --git a/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala b/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala
index 02807cb7c2d..0baf9ac67b5 100644
--- a/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala
+++ b/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala
@@ -19,7 +19,6 @@ import kafka.security.authorizer.AclAuthorizer
import kafka.utils.TestUtils
import org.apache.kafka.common._
import org.apache.kafka.common.acl._
-import org.apache.kafka.common.config.internals.QuotaConfigs
import org.apache.kafka.common.config.{ConfigResource, TopicConfig}
import org.apache.kafka.common.internals.Topic
import org.apache.kafka.common.message.CreatePartitionsRequestData.CreatePartitionsTopic
@@ -45,7 +44,7 @@ import org.apache.kafka.coordinator.group.GroupCoordinatorConfig
import org.apache.kafka.metadata.authorizer.StandardAuthorizer
import org.apache.kafka.network.Session
import org.apache.kafka.server.authorizer.{Action, AuthorizableRequestContext, AuthorizationResult}
-import org.apache.kafka.server.config.KafkaSecurityConfigs
+import org.apache.kafka.server.config.{KafkaSecurityConfigs, QuotaConfigs}
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, BeforeEach, TestInfo}
import org.junit.jupiter.params.ParameterizedTest
diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
index e4d3af9253b..34728d26d5a 100755
--- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala
+++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
@@ -76,9 +76,8 @@ import org.apache.kafka.metadata.properties.MetaProperties
import org.apache.kafka.network.SocketServerConfigs
import org.apache.kafka.server.{ClientMetricsManager, ControllerRequestCompletionHandler}
import org.apache.kafka.server.authorizer.{AuthorizableRequestContext, Authorizer => JAuthorizer}
-import org.apache.kafka.server.config.{ReplicationConfigs, ServerLogConfigs}
+import org.apache.kafka.server.config.{ReplicationConfigs, ServerLogConfigs, QuotaConfigs, ZkConfigs}
import org.apache.kafka.server.common.{ApiMessageAndVersion, MetadataVersion}
-import org.apache.kafka.server.config.ZkConfigs
import org.apache.kafka.server.metrics.KafkaYammerMetrics
import org.apache.kafka.server.util.MockTime
import org.apache.kafka.server.util.timer.SystemTimer
@@ -2328,8 +2327,8 @@ object TestUtils extends Logging {
*/
def throttleAllBrokersReplication(adminClient: Admin, brokerIds: Seq[Int], throttleBytes: Int): Unit = {
val throttleConfigs = Seq(
- new AlterConfigOp(new ConfigEntry(DynamicConfig.Broker.LeaderReplicationThrottledRateProp, throttleBytes.toString), AlterConfigOp.OpType.SET),
- new AlterConfigOp(new ConfigEntry(DynamicConfig.Broker.FollowerReplicationThrottledRateProp, throttleBytes.toString), AlterConfigOp.OpType.SET)
+ new AlterConfigOp(new ConfigEntry(QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG, throttleBytes.toString), AlterConfigOp.OpType.SET),
+ new AlterConfigOp(new ConfigEntry(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG, throttleBytes.toString), AlterConfigOp.OpType.SET)
).asJavaCollection
adminClient.incrementalAlterConfigs(
@@ -2346,8 +2345,8 @@ object TestUtils extends Logging {
val throttles = allReplicasByPartition.groupBy(_._1.topic()).map {
case (topic, replicasByPartition) =>
new ConfigResource(TOPIC, topic) -> Seq(
- new AlterConfigOp(new ConfigEntry(LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, formatReplicaThrottles(replicasByPartition)), AlterConfigOp.OpType.SET),
- new AlterConfigOp(new ConfigEntry(LogConfig.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG, formatReplicaThrottles(replicasByPartition)), AlterConfigOp.OpType.SET)
+ new AlterConfigOp(new ConfigEntry(QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, formatReplicaThrottles(replicasByPartition)), AlterConfigOp.OpType.SET),
+ new AlterConfigOp(new ConfigEntry(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG, formatReplicaThrottles(replicasByPartition)), AlterConfigOp.OpType.SET)
).asJavaCollection
}
adminClient.incrementalAlterConfigs(throttles.asJava).all().get()
@@ -2357,8 +2356,8 @@ object TestUtils extends Logging {
val throttles = partitions.map {
tp =>
new ConfigResource(TOPIC, tp.topic()) -> Seq(
- new AlterConfigOp(new ConfigEntry(LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, ""), AlterConfigOp.OpType.DELETE),
- new AlterConfigOp(new ConfigEntry(LogConfig.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG, ""), AlterConfigOp.OpType.DELETE)
+ new AlterConfigOp(new ConfigEntry(QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, ""), AlterConfigOp.OpType.DELETE),
+ new AlterConfigOp(new ConfigEntry(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG, ""), AlterConfigOp.OpType.DELETE)
).asJavaCollection
}.toMap
adminClient.incrementalAlterConfigs(throttles.asJava).all().get()
diff --git a/core/src/test/scala/unit/kafka/zk/AdminZkClientTest.scala b/core/src/test/scala/unit/kafka/zk/AdminZkClientTest.scala
index 18147683a30..7454a55bfcd 100644
--- a/core/src/test/scala/unit/kafka/zk/AdminZkClientTest.scala
+++ b/core/src/test/scala/unit/kafka/zk/AdminZkClientTest.scala
@@ -19,7 +19,6 @@ package kafka.admin
import java.util
import java.util.{Optional, Properties}
import kafka.controller.ReplicaAssignment
-import kafka.server.DynamicConfig.Broker._
import kafka.server.KafkaConfig._
import kafka.server.{KafkaConfig, KafkaServer, QuorumTestHarness}
import kafka.utils.CoreUtils._
@@ -28,11 +27,10 @@ import kafka.utils.{Logging, TestUtils}
import kafka.zk.{AdminZkClient, ConfigEntityTypeZNode, KafkaZkClient}
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.config.TopicConfig
-import org.apache.kafka.common.config.internals.QuotaConfigs
import org.apache.kafka.common.errors.{InvalidReplicaAssignmentException, InvalidTopicException, TopicExistsException}
import org.apache.kafka.common.metrics.Quota
import org.apache.kafka.server.common.AdminOperationException
-import org.apache.kafka.server.config.ConfigType
+import org.apache.kafka.server.config.{ConfigType, QuotaConfigs}
import org.apache.kafka.storage.internals.log.LogConfig
import org.apache.kafka.test.{TestUtils => JTestUtils}
import org.junit.jupiter.api.Assertions._
@@ -203,8 +201,8 @@ class AdminZkClientTest extends QuorumTestHarness with Logging with RackAwareTes
val props = new Properties()
props.setProperty(TopicConfig.MAX_MESSAGE_BYTES_CONFIG, messageSize.toString)
props.setProperty(TopicConfig.RETENTION_MS_CONFIG, retentionMs.toString)
- props.setProperty(LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, throttledLeaders)
- props.setProperty(LogConfig.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG, throttledFollowers)
+ props.setProperty(QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, throttledLeaders)
+ props.setProperty(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG, throttledFollowers)
props
}
@@ -261,7 +259,7 @@ class AdminZkClientTest extends QuorumTestHarness with Logging with RackAwareTes
checkConfig(maxMessageSize, retentionMs, "0:0,1:0,2:0", "0:1,1:1,2:1", quotaManagerIsThrottled = true)
//Now ensure updating to "" removes the throttled replica list also
- adminZkClient.changeTopicConfig(topic, propsWith((LogConfig.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG, ""), (LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, "")))
+ adminZkClient.changeTopicConfig(topic, propsWith((QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG, ""), (QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, "")))
checkConfig(LogConfig.DEFAULT_MAX_MESSAGE_BYTES, LogConfig.DEFAULT_RETENTION_MS, "", "", quotaManagerIsThrottled = false)
}
@@ -283,27 +281,27 @@ class AdminZkClientTest extends QuorumTestHarness with Logging with RackAwareTes
// Set the limit & check it is applied to the log
adminZkClient.changeBrokerConfig(brokerIds, propsWith(
- (LeaderReplicationThrottledRateProp, limit.toString),
- (FollowerReplicationThrottledRateProp, limit.toString)))
+ (QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG, limit.toString),
+ (QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG, limit.toString)))
checkConfig(limit)
// Now double the config values for the topic and check that it is applied
val newLimit = 2 * limit
adminZkClient.changeBrokerConfig(brokerIds, propsWith(
- (LeaderReplicationThrottledRateProp, newLimit.toString),
- (FollowerReplicationThrottledRateProp, newLimit.toString)))
+ (QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG, newLimit.toString),
+ (QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG, newLimit.toString)))
checkConfig(newLimit)
// Verify that the same config can be read from ZK
for (brokerId <- brokerIds) {
val configInZk = adminZkClient.fetchEntityConfig(ConfigType.BROKER, brokerId.toString)
- assertEquals(newLimit, configInZk.getProperty(LeaderReplicationThrottledRateProp).toInt)
- assertEquals(newLimit, configInZk.getProperty(FollowerReplicationThrottledRateProp).toInt)
+ assertEquals(newLimit, configInZk.getProperty(QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG).toInt)
+ assertEquals(newLimit, configInZk.getProperty(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG).toInt)
}
//Now delete the config
adminZkClient.changeBrokerConfig(brokerIds, new Properties)
- checkConfig(DefaultReplicationThrottledRate)
+ checkConfig(QuotaConfigs.QUOTA_BYTES_PER_SECOND_DEFAULT)
}
/**
diff --git a/core/src/test/scala/unit/kafka/zk/migration/ZkConfigMigrationClientTest.scala b/core/src/test/scala/unit/kafka/zk/migration/ZkConfigMigrationClientTest.scala
index 636895cba10..6b7babdba1d 100644
--- a/core/src/test/scala/unit/kafka/zk/migration/ZkConfigMigrationClientTest.scala
+++ b/core/src/test/scala/unit/kafka/zk/migration/ZkConfigMigrationClientTest.scala
@@ -20,7 +20,6 @@ import kafka.utils.CoreUtils
import kafka.server.ZkAdminManager
import kafka.zk.{AdminZkClient, ZkMigrationClient}
import org.apache.kafka.clients.admin.ScramMechanism
-import org.apache.kafka.common.config.internals.QuotaConfigs
import org.apache.kafka.common.config.types.Password
import org.apache.kafka.common.config.{ConfigResource, TopicConfig}
import org.apache.kafka.common.metadata.ClientQuotaRecord
@@ -38,8 +37,7 @@ import org.apache.kafka.metadata.RecordTestUtils
import org.apache.kafka.metadata.migration.KRaftMigrationZkWriter
import org.apache.kafka.metadata.migration.ZkMigrationLeadershipState
import org.apache.kafka.server.common.ApiMessageAndVersion
-import org.apache.kafka.server.config.{ConfigType, KafkaSecurityConfigs}
-import org.apache.kafka.server.config.ReplicationConfigs
+import org.apache.kafka.server.config.{ConfigType, KafkaSecurityConfigs, ReplicationConfigs, QuotaConfigs}
import org.apache.kafka.server.util.MockRandom
import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue, fail}
import org.junit.jupiter.api.Test
diff --git a/metadata/src/main/java/org/apache/kafka/controller/ClientQuotaControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/ClientQuotaControlManager.java
index cb1ddee8c98..bb518a06805 100644
--- a/metadata/src/main/java/org/apache/kafka/controller/ClientQuotaControlManager.java
+++ b/metadata/src/main/java/org/apache/kafka/controller/ClientQuotaControlManager.java
@@ -18,7 +18,6 @@
package org.apache.kafka.controller;
import org.apache.kafka.common.config.ConfigDef;
-import org.apache.kafka.common.config.internals.QuotaConfigs;
import org.apache.kafka.common.errors.InvalidRequestException;
import org.apache.kafka.common.metadata.ClientQuotaRecord;
import org.apache.kafka.common.metadata.ClientQuotaRecord.EntityData;
@@ -28,6 +27,7 @@ import org.apache.kafka.common.quota.ClientQuotaEntity;
import org.apache.kafka.common.requests.ApiError;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.config.QuotaConfigs;
import org.apache.kafka.server.mutable.BoundedList;
import org.apache.kafka.timeline.SnapshotRegistry;
import org.apache.kafka.timeline.TimelineHashMap;
diff --git a/metadata/src/test/java/org/apache/kafka/controller/ClientQuotaControlManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/ClientQuotaControlManagerTest.java
index 321bd3b1ea2..a9d4cbfc6f6 100644
--- a/metadata/src/test/java/org/apache/kafka/controller/ClientQuotaControlManagerTest.java
+++ b/metadata/src/test/java/org/apache/kafka/controller/ClientQuotaControlManagerTest.java
@@ -18,7 +18,6 @@
package org.apache.kafka.controller;
import org.apache.kafka.common.config.ConfigDef;
-import org.apache.kafka.common.config.internals.QuotaConfigs;
import org.apache.kafka.common.metadata.ClientQuotaRecord;
import org.apache.kafka.common.metadata.ClientQuotaRecord.EntityData;
import org.apache.kafka.common.protocol.Errors;
@@ -27,6 +26,8 @@ import org.apache.kafka.common.quota.ClientQuotaEntity;
import org.apache.kafka.common.requests.ApiError;
import org.apache.kafka.metadata.RecordTestUtils;
import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.config.QuotaConfigs;
+
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
diff --git a/metadata/src/test/java/org/apache/kafka/image/ClientQuotasImageTest.java b/metadata/src/test/java/org/apache/kafka/image/ClientQuotasImageTest.java
index 9b6a8fda9d0..3b9fd83910b 100644
--- a/metadata/src/test/java/org/apache/kafka/image/ClientQuotasImageTest.java
+++ b/metadata/src/test/java/org/apache/kafka/image/ClientQuotasImageTest.java
@@ -17,7 +17,6 @@
package org.apache.kafka.image;
-import org.apache.kafka.common.config.internals.QuotaConfigs;
import org.apache.kafka.common.metadata.ClientQuotaRecord;
import org.apache.kafka.common.metadata.ClientQuotaRecord.EntityData;
import org.apache.kafka.common.quota.ClientQuotaEntity;
@@ -25,6 +24,8 @@ import org.apache.kafka.image.writer.ImageWriterOptions;
import org.apache.kafka.image.writer.RecordListWriter;
import org.apache.kafka.metadata.RecordTestUtils;
import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.config.QuotaConfigs;
+
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
diff --git a/server-common/src/main/java/org/apache/kafka/server/config/QuotaConfigs.java b/server-common/src/main/java/org/apache/kafka/server/config/QuotaConfigs.java
new file mode 100644
index 00000000000..88b49b82827
--- /dev/null
+++ b/server-common/src/main/java/org/apache/kafka/server/config/QuotaConfigs.java
@@ -0,0 +1,150 @@
+/*
+ * 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;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.security.scram.internals.ScramMechanism;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+public class QuotaConfigs {
+ public static final String NUM_QUOTA_SAMPLES_CONFIG = "quota.window.num";
+ public static final String NUM_QUOTA_SAMPLES_DOC = "The number of samples to retain in memory for client quotas";
+ public static final String NUM_CONTROLLER_QUOTA_SAMPLES_CONFIG = "controller.quota.window.num";
+ public static final String NUM_CONTROLLER_QUOTA_SAMPLES_DOC = "The number of samples to retain in memory for controller mutation quotas";
+ public static final String NUM_REPLICATION_QUOTA_SAMPLES_CONFIG = "replication.quota.window.num";
+ public static final String NUM_REPLICATION_QUOTA_SAMPLES_DOC = "The number of samples to retain in memory for replication quotas";
+ public static final String NUM_ALTER_LOG_DIRS_REPLICATION_QUOTA_SAMPLES_CONFIG = "alter.log.dirs.replication.quota.window.num";
+ public static final String NUM_ALTER_LOG_DIRS_REPLICATION_QUOTA_SAMPLES_DOC = "The number of samples to retain in memory for alter log dirs replication quotas";
+
+ // Always have 10 whole windows + 1 current window
+ public static final int NUM_QUOTA_SAMPLES_DEFAULT = 11;
+
+ public static final String QUOTA_WINDOW_SIZE_SECONDS_CONFIG = "quota.window.size.seconds";
+ public static final String QUOTA_WINDOW_SIZE_SECONDS_DOC = "The time span of each sample for client quotas";
+ public static final String CONTROLLER_QUOTA_WINDOW_SIZE_SECONDS_CONFIG = "controller.quota.window.size.seconds";
+ public static final String CONTROLLER_QUOTA_WINDOW_SIZE_SECONDS_DOC = "The time span of each sample for controller mutations quotas";
+ public static final String REPLICATION_QUOTA_WINDOW_SIZE_SECONDS_CONFIG = "replication.quota.window.size.seconds";
+ public static final String REPLICATION_QUOTA_WINDOW_SIZE_SECONDS_DOC = "The time span of each sample for replication quotas";
+ public static final String ALTER_LOG_DIRS_REPLICATION_QUOTA_WINDOW_SIZE_SECONDS_CONFIG = "alter.log.dirs.replication.quota.window.size.seconds";
+ public static final String ALTER_LOG_DIRS_REPLICATION_QUOTA_WINDOW_SIZE_SECONDS_DOC = "The time span of each sample for alter log dirs replication quotas";
+ public static final int QUOTA_WINDOW_SIZE_SECONDS_DEFAULT = 1;
+
+ public static final String CLIENT_QUOTA_CALLBACK_CLASS_CONFIG = "client.quota.callback.class";
+ public static final String CLIENT_QUOTA_CALLBACK_CLASS_DOC = "The fully qualified name of a class that implements the ClientQuotaCallback interface, " +
+ "which is used to determine quota limits applied to client requests. By default, the <user> and <client-id> " +
+ "quotas that are stored in ZooKeeper are applied. For any given request, the most specific quota that matches the user principal " +
+ "of the session and the client-id of the request is applied.";
+
+ public static final String LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG = "leader.replication.throttled.replicas";
+ public static final String LEADER_REPLICATION_THROTTLED_REPLICAS_DOC = "A list of replicas for which log replication should be throttled on " +
+ "the leader side. The list should describe a set of replicas in the form " +
+ "[PartitionId]:[BrokerId],[PartitionId]:[BrokerId]:... or alternatively the wildcard '*' can be used to throttle " +
+ "all replicas for this topic.";
+ public static final List<String> LEADER_REPLICATION_THROTTLED_REPLICAS_DEFAULT = Collections.emptyList();
+
+ public static final String FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG = "follower.replication.throttled.replicas";
+ public static final String FOLLOWER_REPLICATION_THROTTLED_REPLICAS_DOC = "A list of replicas for which log replication should be throttled on " +
+ "the follower side. The list should describe a set of " + "replicas in the form " +
+ "[PartitionId]:[BrokerId],[PartitionId]:[BrokerId]:... or alternatively the wildcard '*' can be used to throttle " +
+ "all replicas for this topic.";
+ public static final List<String> FOLLOWER_REPLICATION_THROTTLED_REPLICAS_DEFAULT = Collections.emptyList();
+
+
+ public static final String LEADER_REPLICATION_THROTTLED_RATE_CONFIG = "leader.replication.throttled.rate";
+ public static final String LEADER_REPLICATION_THROTTLED_RATE_DOC = "A long representing the upper bound (bytes/sec) on replication traffic for leaders enumerated in the " +
+ String.format("property %s (for each topic). This property can be only set dynamically. It is suggested that the ", LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG) +
+ "limit be kept above 1MB/s for accurate behaviour.";
+
+ public static final String FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG = "follower.replication.throttled.rate";
+ public static final String FOLLOWER_REPLICATION_THROTTLED_RATE_DOC = "A long representing the upper bound (bytes/sec) on replication traffic for followers enumerated in the " +
+ String.format("property %s (for each topic). This property can be only set dynamically. It is suggested that the ", FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG) +
+ "limit be kept above 1MB/s for accurate behaviour.";
+ public static final String REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_CONFIG = "replica.alter.log.dirs.io.max.bytes.per.second";
+ public static final String REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_DOC = "A long representing the upper bound (bytes/sec) on disk IO used for moving replica between log directories on the same broker. " +
+ "This property can be only set dynamically. It is suggested that the limit be kept above 1MB/s for accurate behaviour.";
+ public static final long QUOTA_BYTES_PER_SECOND_DEFAULT = Long.MAX_VALUE;
+
+ public static final String PRODUCER_BYTE_RATE_OVERRIDE_CONFIG = "producer_byte_rate";
+ public static final String CONSUMER_BYTE_RATE_OVERRIDE_CONFIG = "consumer_byte_rate";
+ public static final String REQUEST_PERCENTAGE_OVERRIDE_CONFIG = "request_percentage";
+ public static final String CONTROLLER_MUTATION_RATE_OVERRIDE_CONFIG = "controller_mutation_rate";
+ public static final String IP_CONNECTION_RATE_OVERRIDE_CONFIG = "connection_creation_rate";
+ public static final String PRODUCER_BYTE_RATE_DOC = "A rate representing the upper bound (bytes/sec) for producer traffic.";
+ public static final String CONSUMER_BYTE_RATE_DOC = "A rate representing the upper bound (bytes/sec) for consumer traffic.";
+ public static final String REQUEST_PERCENTAGE_DOC = "A percentage representing the upper bound of time spent for processing requests.";
+ public static final String CONTROLLER_MUTATION_RATE_DOC = "The rate at which mutations are accepted for the create " +
+ "topics request, the create partitions request and the delete topics request. The rate is accumulated by " +
+ "the number of partitions created or deleted.";
+ public static final String IP_CONNECTION_RATE_DOC = "An int representing the upper bound of connections accepted " +
+ "for the specified IP.";
+
+ public static final int IP_CONNECTION_RATE_DEFAULT = Integer.MAX_VALUE;
+
+ private final static Set<String> USER_AND_CLIENT_QUOTA_NAMES = Collections.unmodifiableSet(new HashSet<>(Arrays.asList(
+ PRODUCER_BYTE_RATE_OVERRIDE_CONFIG,
+ CONSUMER_BYTE_RATE_OVERRIDE_CONFIG,
+ REQUEST_PERCENTAGE_OVERRIDE_CONFIG,
+ CONTROLLER_MUTATION_RATE_OVERRIDE_CONFIG
+ )));
+
+ private static void buildUserClientQuotaConfigDef(ConfigDef configDef) {
+ configDef.define(PRODUCER_BYTE_RATE_OVERRIDE_CONFIG, ConfigDef.Type.LONG, Long.MAX_VALUE,
+ ConfigDef.Importance.MEDIUM, PRODUCER_BYTE_RATE_DOC);
+
+ configDef.define(CONSUMER_BYTE_RATE_OVERRIDE_CONFIG, ConfigDef.Type.LONG, Long.MAX_VALUE,
+ ConfigDef.Importance.MEDIUM, CONSUMER_BYTE_RATE_DOC);
+
+ configDef.define(REQUEST_PERCENTAGE_OVERRIDE_CONFIG, ConfigDef.Type.DOUBLE,
+ Integer.valueOf(Integer.MAX_VALUE).doubleValue(),
+ ConfigDef.Importance.MEDIUM, REQUEST_PERCENTAGE_DOC);
+
+ configDef.define(CONTROLLER_MUTATION_RATE_OVERRIDE_CONFIG, ConfigDef.Type.DOUBLE,
+ Integer.valueOf(Integer.MAX_VALUE).doubleValue(),
+ ConfigDef.Importance.MEDIUM, CONTROLLER_MUTATION_RATE_DOC);
+ }
+
+ public static ConfigDef userAndClientQuotaConfigs() {
+ ConfigDef configDef = new ConfigDef();
+ buildUserClientQuotaConfigDef(configDef);
+ return configDef;
+ }
+
+ public static ConfigDef scramMechanismsPlusUserAndClientQuotaConfigs() {
+ ConfigDef configDef = new ConfigDef();
+ ScramMechanism.mechanismNames().forEach(mechanismName -> {
+ configDef.define(mechanismName, ConfigDef.Type.STRING, null, ConfigDef.Importance.MEDIUM,
+ "User credentials for SCRAM mechanism " + mechanismName);
+ });
+ buildUserClientQuotaConfigDef(configDef);
+ return configDef;
+ }
+
+ public static ConfigDef ipConfigs() {
+ ConfigDef configDef = new ConfigDef();
+ configDef.define(IP_CONNECTION_RATE_OVERRIDE_CONFIG, ConfigDef.Type.INT, Integer.MAX_VALUE,
+ ConfigDef.Range.atLeast(0), ConfigDef.Importance.MEDIUM, IP_CONNECTION_RATE_DOC);
+ return configDef;
+ }
+
+ public static Boolean isClientOrUserQuotaConfig(String name) {
+ return USER_AND_CLIENT_QUOTA_NAMES.contains(name);
+ }
+}
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
index 3ac9b5bd119..a620ac878df 100644
--- a/server/src/main/java/org/apache/kafka/server/config/ClientQuotaManagerConfig.java
+++ b/server/src/main/java/org/apache/kafka/server/config/ClientQuotaManagerConfig.java
@@ -15,12 +15,7 @@
* 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;
@@ -36,10 +31,10 @@ public class ClientQuotaManagerConfig {
}
public ClientQuotaManagerConfig() {
- this(DEFAULT_NUM_QUOTA_SAMPLES, DEFAULT_QUOTA_WINDOW_SIZE_SECONDS);
+ this(QuotaConfigs.NUM_QUOTA_SAMPLES_DEFAULT, QuotaConfigs.QUOTA_WINDOW_SIZE_SECONDS_DEFAULT);
}
public ClientQuotaManagerConfig(int numQuotaSamples) {
- this(numQuotaSamples, DEFAULT_QUOTA_WINDOW_SIZE_SECONDS);
+ this(numQuotaSamples, QuotaConfigs.QUOTA_WINDOW_SIZE_SECONDS_DEFAULT);
}
}
diff --git a/server/src/main/java/org/apache/kafka/server/config/Defaults.java b/server/src/main/java/org/apache/kafka/server/config/Defaults.java
index 7d156b2ef83..50663e53371 100644
--- a/server/src/main/java/org/apache/kafka/server/config/Defaults.java
+++ b/server/src/main/java/org/apache/kafka/server/config/Defaults.java
@@ -65,16 +65,6 @@ public class Defaults {
/** ********* Request Limit Configuration ***********/
public static final int MAX_REQUEST_PARTITION_SIZE_LIMIT = 2000;
- /** ********* Quota Configuration *********/
- public static final int NUM_QUOTA_SAMPLES = ClientQuotaManagerConfig.DEFAULT_NUM_QUOTA_SAMPLES;
- public static final int QUOTA_WINDOW_SIZE_SECONDS = ClientQuotaManagerConfig.DEFAULT_QUOTA_WINDOW_SIZE_SECONDS;
- public static final int NUM_REPLICATION_QUOTA_SAMPLES = ReplicationQuotaManagerConfig.DEFAULT_NUM_QUOTA_SAMPLES;
- public static final int REPLICATION_QUOTA_WINDOW_SIZE_SECONDS = ReplicationQuotaManagerConfig.DEFAULT_QUOTA_WINDOW_SIZE_SECONDS;
- public static final int NUM_ALTER_LOG_DIRS_REPLICATION_QUOTA_SAMPLES = ReplicationQuotaManagerConfig.DEFAULT_NUM_QUOTA_SAMPLES;
- public static final int ALTER_LOG_DIRS_REPLICATION_QUOTA_WINDOW_SIZE_SECONDS = ReplicationQuotaManagerConfig.DEFAULT_QUOTA_WINDOW_SIZE_SECONDS;
- public static final int NUM_CONTROLLER_QUOTA_SAMPLES = ClientQuotaManagerConfig.DEFAULT_NUM_QUOTA_SAMPLES;
- public static final int CONTROLLER_QUOTA_WINDOW_SIZE_SECONDS = ClientQuotaManagerConfig.DEFAULT_QUOTA_WINDOW_SIZE_SECONDS;
-
/** ********* Kafka Metrics Configuration *********/
public static final int METRIC_NUM_SAMPLES = 2;
public static final int METRIC_SAMPLE_WINDOW_MS = 30000;
diff --git a/server/src/main/java/org/apache/kafka/server/config/ReplicationQuotaManagerConfig.java b/server/src/main/java/org/apache/kafka/server/config/ReplicationQuotaManagerConfig.java
index a8bb7b4c6af..a3c109ecb34 100644
--- a/server/src/main/java/org/apache/kafka/server/config/ReplicationQuotaManagerConfig.java
+++ b/server/src/main/java/org/apache/kafka/server/config/ReplicationQuotaManagerConfig.java
@@ -17,10 +17,6 @@
package org.apache.kafka.server.config;
public class ReplicationQuotaManagerConfig {
- public static final long DEFAULT_QUOTA_BYTES_PER_SECOND = Long.MAX_VALUE;
- // 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;
// Purge sensors after 1 hour of inactivity
public static final int INACTIVE_SENSOR_EXPIRATION_TIME_SECONDS = 3600;
@@ -42,10 +38,10 @@ public class ReplicationQuotaManagerConfig {
}
public ReplicationQuotaManagerConfig(int numQuotaSamples, int quotaWindowSizeSeconds) {
- this(DEFAULT_QUOTA_BYTES_PER_SECOND, numQuotaSamples, quotaWindowSizeSeconds);
+ this(QuotaConfigs.QUOTA_BYTES_PER_SECOND_DEFAULT, numQuotaSamples, quotaWindowSizeSeconds);
}
public ReplicationQuotaManagerConfig() {
- this(DEFAULT_QUOTA_BYTES_PER_SECOND, DEFAULT_NUM_QUOTA_SAMPLES, DEFAULT_QUOTA_WINDOW_SIZE_SECONDS);
+ this(QuotaConfigs.QUOTA_BYTES_PER_SECOND_DEFAULT, QuotaConfigs.NUM_QUOTA_SAMPLES_DEFAULT, QuotaConfigs.QUOTA_WINDOW_SIZE_SECONDS_DEFAULT);
}
-}
\ No newline at end of file
+}
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java
index 174987e6bcc..72211f37cec 100644
--- a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java
@@ -56,6 +56,7 @@ import org.apache.kafka.common.utils.ConfigUtils;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.server.common.MetadataVersion;
import org.apache.kafka.server.common.MetadataVersionValidator;
+import org.apache.kafka.server.config.QuotaConfigs;
import org.apache.kafka.server.config.ServerLogConfigs;
import org.apache.kafka.server.config.ServerTopicConfigSynonyms;
import org.apache.kafka.server.record.BrokerCompressionType;
@@ -173,18 +174,12 @@ public class LogConfig extends AbstractConfig {
public static final boolean DEFAULT_REMOTE_STORAGE_ENABLE = false;
public static final long DEFAULT_LOCAL_RETENTION_BYTES = -2; // It indicates the value to be derived from RetentionBytes
public static final long DEFAULT_LOCAL_RETENTION_MS = -2; // It indicates the value to be derived from RetentionMs
- public static final List<String> DEFAULT_LEADER_REPLICATION_THROTTLED_REPLICAS = Collections.emptyList();
- public static final List<String> DEFAULT_FOLLOWER_REPLICATION_THROTTLED_REPLICAS = Collections.emptyList();
/* See `TopicConfig.MESSAGE_FORMAT_VERSION_CONFIG` for details
* Keep DEFAULT_MESSAGE_FORMAT_VERSION as a way to handlee the deprecated value */
@Deprecated
public static final String DEFAULT_MESSAGE_FORMAT_VERSION = ServerLogConfigs.LOG_MESSAGE_FORMAT_VERSION_DEFAULT;
- // Leave these out of TopicConfig for now as they are replication quota configs
- public static final String LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG = "leader.replication.throttled.replicas";
- public static final String FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG = "follower.replication.throttled.replicas";
-
/* See `TopicConfig.MESSAGE_FORMAT_VERSION_CONFIG` for details */
@SuppressWarnings("deprecation")
private static final String MESSAGE_FORMAT_VERSION_CONFIG = TopicConfig.MESSAGE_FORMAT_VERSION_CONFIG;
@@ -198,19 +193,10 @@ public class LogConfig extends AbstractConfig {
// Visible for testing
public static final Set<String> CONFIGS_WITH_NO_SERVER_DEFAULTS = Collections.unmodifiableSet(Utils.mkSet(
TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG,
- LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG,
- FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG
+ QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG,
+ QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG
));
- public static final String LEADER_REPLICATION_THROTTLED_REPLICAS_DOC = "A list of replicas for which log replication should be throttled on " +
- "the leader side. The list should describe a set of replicas in the form " +
- "[PartitionId]:[BrokerId],[PartitionId]:[BrokerId]:... or alternatively the wildcard '*' can be used to throttle " +
- "all replicas for this topic.";
- public static final String FOLLOWER_REPLICATION_THROTTLED_REPLICAS_DOC = "A list of replicas for which log replication should be throttled on " +
- "the follower side. The list should describe a set of " + "replicas in the form " +
- "[PartitionId]:[BrokerId],[PartitionId]:[BrokerId]:... or alternatively the wildcard '*' can be used to throttle " +
- "all replicas for this topic.";
-
@SuppressWarnings("deprecation")
private static final String MESSAGE_FORMAT_VERSION_DOC = TopicConfig.MESSAGE_FORMAT_VERSION_DOC;
@@ -266,10 +252,10 @@ public class LogConfig extends AbstractConfig {
atLeast(0), MEDIUM, TopicConfig.MESSAGE_TIMESTAMP_BEFORE_MAX_MS_DOC)
.define(TopicConfig.MESSAGE_TIMESTAMP_AFTER_MAX_MS_CONFIG, LONG, ServerLogConfigs.LOG_MESSAGE_TIMESTAMP_AFTER_MAX_MS_DEFAULT,
atLeast(0), MEDIUM, TopicConfig.MESSAGE_TIMESTAMP_AFTER_MAX_MS_DOC)
- .define(LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, LIST, DEFAULT_LEADER_REPLICATION_THROTTLED_REPLICAS,
- ThrottledReplicaListValidator.INSTANCE, MEDIUM, LEADER_REPLICATION_THROTTLED_REPLICAS_DOC)
- .define(FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG, LIST, DEFAULT_FOLLOWER_REPLICATION_THROTTLED_REPLICAS,
- ThrottledReplicaListValidator.INSTANCE, MEDIUM, FOLLOWER_REPLICATION_THROTTLED_REPLICAS_DOC)
+ .define(QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, LIST, QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_DEFAULT,
+ ThrottledReplicaListValidator.INSTANCE, MEDIUM, QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_DOC)
+ .define(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG, LIST, QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_DEFAULT,
+ ThrottledReplicaListValidator.INSTANCE, MEDIUM, QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_DOC)
.define(TopicConfig.MESSAGE_DOWNCONVERSION_ENABLE_CONFIG, BOOLEAN, ServerLogConfigs.LOG_MESSAGE_DOWNCONVERSION_ENABLE_DEFAULT, LOW,
TopicConfig.MESSAGE_DOWNCONVERSION_ENABLE_DOC)
.define(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, BOOLEAN, DEFAULT_REMOTE_STORAGE_ENABLE, null,
@@ -368,8 +354,8 @@ public class LogConfig extends AbstractConfig {
this.messageTimestampDifferenceMaxMs = getLong(TopicConfig.MESSAGE_TIMESTAMP_DIFFERENCE_MAX_MS_CONFIG);
this.messageTimestampBeforeMaxMs = getMessageTimestampBeforeMaxMs();
this.messageTimestampAfterMaxMs = getMessageTimestampAfterMaxMs();
- this.leaderReplicationThrottledReplicas = Collections.unmodifiableList(getList(LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG));
- this.followerReplicationThrottledReplicas = Collections.unmodifiableList(getList(LogConfig.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG));
+ this.leaderReplicationThrottledReplicas = Collections.unmodifiableList(getList(QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG));
+ this.followerReplicationThrottledReplicas = Collections.unmodifiableList(getList(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG));
this.messageDownConversionEnable = getBoolean(TopicConfig.MESSAGE_DOWNCONVERSION_ENABLE_CONFIG);
remoteLogConfig = new RemoteLogConfig(this);
diff --git a/tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java b/tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java
index 1f63a23b290..96fd33360db 100644
--- a/tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java
+++ b/tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java
@@ -42,6 +42,7 @@ import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.server.common.AdminCommandFailedException;
import org.apache.kafka.server.common.AdminOperationException;
+import org.apache.kafka.server.config.QuotaConfigs;
import org.apache.kafka.server.util.CommandLineUtils;
import org.apache.kafka.server.util.Json;
import org.apache.kafka.server.util.json.DecodeJson;
@@ -97,27 +98,15 @@ public class ReassignPartitionsCommand {
*/
static final int EARLIEST_TOPICS_JSON_VERSION = 1;
- // Throttles that are set at the level of an individual broker.
- //DynamicConfig.Broker.LeaderReplicationThrottledRateProp
- static final String BROKER_LEVEL_LEADER_THROTTLE = "leader.replication.throttled.rate";
- //DynamicConfig.Broker.FollowerReplicationThrottledRateProp
- static final String BROKER_LEVEL_FOLLOWER_THROTTLE = "follower.replication.throttled.rate";
- //DynamicConfig.Broker.ReplicaAlterLogDirsIoMaxBytesPerSecondProp
- static final String BROKER_LEVEL_LOG_DIR_THROTTLE = "replica.alter.log.dirs.io.max.bytes.per.second";
static final List<String> BROKER_LEVEL_THROTTLES = Arrays.asList(
- BROKER_LEVEL_LEADER_THROTTLE,
- BROKER_LEVEL_FOLLOWER_THROTTLE,
- BROKER_LEVEL_LOG_DIR_THROTTLE
+ QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG,
+ QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG,
+ QuotaConfigs.REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_CONFIG
);
- // Throttles that are set at the level of an individual topic.
- //LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG
- static final String TOPIC_LEVEL_LEADER_THROTTLE = "leader.replication.throttled.replicas";
- //LogConfig.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG
- static final String TOPIC_LEVEL_FOLLOWER_THROTTLE = "follower.replication.throttled.replicas";
private static final List<String> TOPIC_LEVEL_THROTTLES = Arrays.asList(
- TOPIC_LEVEL_LEADER_THROTTLE,
- TOPIC_LEVEL_FOLLOWER_THROTTLE
+ QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG,
+ QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG
);
private static final String CANNOT_EXECUTE_BECAUSE_OF_EXISTING_MESSAGE = "Cannot execute because " +
@@ -1105,10 +1094,10 @@ public class ReassignPartitionsCommand {
topicNames.forEach(topicName -> {
List<AlterConfigOp> ops = new ArrayList<>();
if (leaderThrottles.containsKey(topicName)) {
- ops.add(new AlterConfigOp(new ConfigEntry(TOPIC_LEVEL_LEADER_THROTTLE, leaderThrottles.get(topicName)), AlterConfigOp.OpType.SET));
+ ops.add(new AlterConfigOp(new ConfigEntry(QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, leaderThrottles.get(topicName)), AlterConfigOp.OpType.SET));
}
if (followerThrottles.containsKey(topicName)) {
- ops.add(new AlterConfigOp(new ConfigEntry(TOPIC_LEVEL_FOLLOWER_THROTTLE, followerThrottles.get(topicName)), AlterConfigOp.OpType.SET));
+ ops.add(new AlterConfigOp(new ConfigEntry(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG, followerThrottles.get(topicName)), AlterConfigOp.OpType.SET));
}
if (!ops.isEmpty()) {
configs.put(new ConfigResource(ConfigResource.Type.TOPIC, topicName), ops);
@@ -1144,9 +1133,9 @@ public class ReassignPartitionsCommand {
Map<ConfigResource, Collection<AlterConfigOp>> configs = new HashMap<>();
reassigningBrokers.forEach(brokerId -> {
List<AlterConfigOp> ops = new ArrayList<>();
- ops.add(new AlterConfigOp(new ConfigEntry(BROKER_LEVEL_LEADER_THROTTLE,
+ ops.add(new AlterConfigOp(new ConfigEntry(QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG,
Long.toString(interBrokerThrottle)), AlterConfigOp.OpType.SET));
- ops.add(new AlterConfigOp(new ConfigEntry(BROKER_LEVEL_FOLLOWER_THROTTLE,
+ ops.add(new AlterConfigOp(new ConfigEntry(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG,
Long.toString(interBrokerThrottle)), AlterConfigOp.OpType.SET));
configs.put(new ConfigResource(ConfigResource.Type.BROKER, Long.toString(brokerId)), ops);
});
@@ -1169,7 +1158,7 @@ public class ReassignPartitionsCommand {
Map<ConfigResource, Collection<AlterConfigOp>> configs = new HashMap<>();
movingBrokers.forEach(brokerId -> {
List<AlterConfigOp> ops = new ArrayList<>();
- ops.add(new AlterConfigOp(new ConfigEntry(BROKER_LEVEL_LOG_DIR_THROTTLE, Long.toString(logDirThrottle)), AlterConfigOp.OpType.SET));
+ ops.add(new AlterConfigOp(new ConfigEntry(QuotaConfigs.REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_CONFIG, Long.toString(logDirThrottle)), AlterConfigOp.OpType.SET));
configs.put(new ConfigResource(ConfigResource.Type.BROKER, Long.toString(brokerId)), ops);
});
admin.incrementalAlterConfigs(configs).all().get();
diff --git a/tools/src/test/java/org/apache/kafka/tools/ToolsTestUtils.java b/tools/src/test/java/org/apache/kafka/tools/ToolsTestUtils.java
index 2e90fe23366..8e6f8be9315 100644
--- a/tools/src/test/java/org/apache/kafka/tools/ToolsTestUtils.java
+++ b/tools/src/test/java/org/apache/kafka/tools/ToolsTestUtils.java
@@ -16,7 +16,6 @@
*/
package org.apache.kafka.tools;
-import kafka.server.DynamicConfig;
import kafka.utils.TestInfoUtils;
import kafka.utils.TestUtils;
import org.apache.kafka.clients.admin.Admin;
@@ -25,7 +24,7 @@ import org.apache.kafka.clients.admin.ConfigEntry;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.config.ConfigResource;
import org.apache.kafka.common.utils.Exit;
-import org.apache.kafka.storage.internals.log.LogConfig;
+import org.apache.kafka.server.config.QuotaConfigs;
import java.io.ByteArrayOutputStream;
import java.io.File;
@@ -132,9 +131,9 @@ public class ToolsTestUtils {
*/
public static void throttleAllBrokersReplication(Admin adminClient, List<Integer> brokerIds, int throttleBytes) throws ExecutionException, InterruptedException {
List<AlterConfigOp> throttleConfigs = new ArrayList<>();
- throttleConfigs.add(new AlterConfigOp(new ConfigEntry(DynamicConfig.Broker$.MODULE$.LeaderReplicationThrottledRateProp(),
+ throttleConfigs.add(new AlterConfigOp(new ConfigEntry(QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG,
Integer.toString(throttleBytes)), AlterConfigOp.OpType.SET));
- throttleConfigs.add(new AlterConfigOp(new ConfigEntry(DynamicConfig.Broker$.MODULE$.FollowerReplicationThrottledRateProp(),
+ throttleConfigs.add(new AlterConfigOp(new ConfigEntry(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG,
Integer.toString(throttleBytes)), AlterConfigOp.OpType.SET));
Map<ConfigResource, Collection<AlterConfigOp>> configs = new HashMap<>();
@@ -168,10 +167,10 @@ public class ToolsTestUtils {
Map<TopicPartition, List<Integer>> replicaThrottle =
entry.getValue().stream().collect(Collectors.toMap(Entry::getKey, Entry::getValue));
alterConfigOps.add(new AlterConfigOp(
- new ConfigEntry(LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, formatReplicaThrottles(replicaThrottle)),
+ new ConfigEntry(QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, formatReplicaThrottles(replicaThrottle)),
AlterConfigOp.OpType.SET));
alterConfigOps.add(new AlterConfigOp(
- new ConfigEntry(LogConfig.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG, formatReplicaThrottles(replicaThrottle)),
+ new ConfigEntry(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG, formatReplicaThrottles(replicaThrottle)),
AlterConfigOp.OpType.SET));
return alterConfigOps;
}
@@ -187,9 +186,9 @@ public class ToolsTestUtils {
Map<ConfigResource, Collection<AlterConfigOp>> throttles = partitions.stream().collect(Collectors.toMap(
tp -> new ConfigResource(ConfigResource.Type.TOPIC, tp.topic()),
tp -> Arrays.asList(
- new AlterConfigOp(new ConfigEntry(LogConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, ""),
+ new AlterConfigOp(new ConfigEntry(QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, ""),
AlterConfigOp.OpType.DELETE),
- new AlterConfigOp(new ConfigEntry(LogConfig.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG, ""),
+ new AlterConfigOp(new ConfigEntry(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG, ""),
AlterConfigOp.OpType.DELETE))
));
diff --git a/tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsIntegrationTest.java b/tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsIntegrationTest.java
index 4e5bf059dea..59001333cc9 100644
--- a/tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsIntegrationTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsIntegrationTest.java
@@ -42,6 +42,7 @@ import org.apache.kafka.common.security.auth.SecurityProtocol;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.server.config.QuotaConfigs;
import org.apache.kafka.tools.TerseException;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Timeout;
@@ -79,9 +80,6 @@ import static org.apache.kafka.server.config.ReplicationConfigs.REPLICA_FETCH_BA
import static org.apache.kafka.server.config.ReplicationConfigs.REPLICA_LAG_TIME_MAX_MS_CONFIG;
import static org.apache.kafka.server.common.MetadataVersion.IBP_2_7_IV1;
import static org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS;
-import static org.apache.kafka.tools.reassign.ReassignPartitionsCommand.BROKER_LEVEL_FOLLOWER_THROTTLE;
-import static org.apache.kafka.tools.reassign.ReassignPartitionsCommand.BROKER_LEVEL_LEADER_THROTTLE;
-import static org.apache.kafka.tools.reassign.ReassignPartitionsCommand.BROKER_LEVEL_LOG_DIR_THROTTLE;
import static org.apache.kafka.tools.reassign.ReassignPartitionsCommand.BROKER_LEVEL_THROTTLES;
import static org.apache.kafka.tools.reassign.ReassignPartitionsCommand.cancelAssignment;
import static org.apache.kafka.tools.reassign.ReassignPartitionsCommand.executeAssignment;
@@ -474,17 +472,17 @@ public class ReassignPartitionsIntegrationTest extends QuorumTestHarness {
private void waitForLogDirThrottle(Set<Integer> throttledBrokers, Long logDirThrottle) {
Map<String, Long> throttledConfigMap = new HashMap<>();
- throttledConfigMap.put(BROKER_LEVEL_LEADER_THROTTLE, -1L);
- throttledConfigMap.put(BROKER_LEVEL_FOLLOWER_THROTTLE, -1L);
- throttledConfigMap.put(BROKER_LEVEL_LOG_DIR_THROTTLE, logDirThrottle);
+ throttledConfigMap.put(QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG, -1L);
+ throttledConfigMap.put(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG, -1L);
+ throttledConfigMap.put(QuotaConfigs.REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_CONFIG, logDirThrottle);
waitForBrokerThrottles(throttledBrokers, throttledConfigMap);
}
private void waitForInterBrokerThrottle(List<Integer> throttledBrokers, Long interBrokerThrottle) {
Map<String, Long> throttledConfigMap = new HashMap<>();
- throttledConfigMap.put(BROKER_LEVEL_LEADER_THROTTLE, interBrokerThrottle);
- throttledConfigMap.put(BROKER_LEVEL_FOLLOWER_THROTTLE, interBrokerThrottle);
- throttledConfigMap.put(BROKER_LEVEL_LOG_DIR_THROTTLE, -1L);
+ throttledConfigMap.put(QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG, interBrokerThrottle);
+ throttledConfigMap.put(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG, interBrokerThrottle);
+ throttledConfigMap.put(QuotaConfigs.REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_CONFIG, -1L);
waitForBrokerThrottles(throttledBrokers, throttledConfigMap);
}
@@ -572,7 +570,7 @@ public class ReassignPartitionsIntegrationTest extends QuorumTestHarness {
cluster.adminClient.incrementalAlterConfigs(Collections.singletonMap(
new ConfigResource(ConfigResource.Type.BROKER, "0"),
Collections.singletonList(new AlterConfigOp(
- new ConfigEntry(BROKER_LEVEL_LOG_DIR_THROTTLE, ""), AlterConfigOp.OpType.DELETE))))
+ new ConfigEntry(QuotaConfigs.REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_CONFIG, ""), AlterConfigOp.OpType.DELETE))))
.all().get();
waitForBrokerLevelThrottles(unthrottledBrokerConfigs);
diff --git a/tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsUnitTest.java b/tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsUnitTest.java
index 9d248ee2072..609b68c4cc9 100644
--- a/tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsUnitTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsUnitTest.java
@@ -31,6 +31,8 @@ import org.apache.kafka.common.utils.Exit;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.server.common.AdminCommandFailedException;
import org.apache.kafka.server.common.AdminOperationException;
+import org.apache.kafka.server.config.QuotaConfigs;
+
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
@@ -50,11 +52,6 @@ import java.util.stream.Collectors;
import java.util.stream.Stream;
import static java.util.Arrays.asList;
-import static org.apache.kafka.tools.reassign.ReassignPartitionsCommand.BROKER_LEVEL_FOLLOWER_THROTTLE;
-import static org.apache.kafka.tools.reassign.ReassignPartitionsCommand.BROKER_LEVEL_LEADER_THROTTLE;
-import static org.apache.kafka.tools.reassign.ReassignPartitionsCommand.BROKER_LEVEL_LOG_DIR_THROTTLE;
-import static org.apache.kafka.tools.reassign.ReassignPartitionsCommand.TOPIC_LEVEL_FOLLOWER_THROTTLE;
-import static org.apache.kafka.tools.reassign.ReassignPartitionsCommand.TOPIC_LEVEL_LEADER_THROTTLE;
import static org.apache.kafka.tools.reassign.ReassignPartitionsCommand.alterPartitionReassignments;
import static org.apache.kafka.tools.reassign.ReassignPartitionsCommand.alterReplicaLogDirs;
import static org.apache.kafka.tools.reassign.ReassignPartitionsCommand.calculateFollowerThrottles;
@@ -692,13 +689,13 @@ public class ReassignPartitionsUnitTest {
config.entries().forEach(entry -> configs.put(entry.name(), entry.value()));
if (expectedInterBrokerThrottle >= 0) {
assertEquals(Long.toString(expectedInterBrokerThrottle),
- configs.getOrDefault(BROKER_LEVEL_LEADER_THROTTLE, ""));
+ configs.getOrDefault(QuotaConfigs.LEADER_REPLICATION_THROTTLED_RATE_CONFIG, ""));
assertEquals(Long.toString(expectedInterBrokerThrottle),
- configs.getOrDefault(BROKER_LEVEL_FOLLOWER_THROTTLE, ""));
+ configs.getOrDefault(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG, ""));
}
if (expectedReplicaAlterLogDirsThrottle >= 0) {
assertEquals(Long.toString(expectedReplicaAlterLogDirsThrottle),
- configs.getOrDefault(BROKER_LEVEL_LOG_DIR_THROTTLE, ""));
+ configs.getOrDefault(QuotaConfigs.REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_CONFIG, ""));
}
}
@@ -729,9 +726,9 @@ public class ReassignPartitionsUnitTest {
Map<String, String> configs = new HashMap<>();
config.entries().forEach(entry -> configs.put(entry.name(), entry.value()));
assertEquals(expectedLeaderThrottle,
- configs.getOrDefault(TOPIC_LEVEL_LEADER_THROTTLE, ""));
+ configs.getOrDefault(QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, ""));
assertEquals(expectedFollowerThrottle,
- configs.getOrDefault(TOPIC_LEVEL_FOLLOWER_THROTTLE, ""));
+ configs.getOrDefault(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG, ""));
}
@Test