MINOR: Remove unused QuotaConfgHandler (#18617)

Reviewers: Chia-Ping Tsai <[email protected]>

Author: frankvicky

core/src/main/scala/kafka/server/ConfigHandler.scala

@@ -17,16 +17,12 @@
 
 package kafka.server
 
-import java.net.{InetAddress, UnknownHostException}
 import java.util.{Collections, Properties}
 import kafka.log.UnifiedLog
-import kafka.network.ConnectionQuotas
 import kafka.server.QuotaFactory.QuotaManagers
 import kafka.utils.Logging
 import org.apache.kafka.server.config.{QuotaConfig, ZooKeeperInternals}
-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.coordinator.group.GroupCoordinator
 import org.apache.kafka.server.ClientMetricsManager
 import org.apache.kafka.server.common.StopPartition
@@ -142,60 +138,6 @@ class TopicConfigHandler(private val replicaManager: ReplicaManager,
   }
 }
 
-
-/**
- * Handles <client-id>, <user> or <user, client-id> quota config updates in ZK.
- * This implementation reports the overrides to the respective ClientQuotaManager objects
- */
-class QuotaConfigHandler(private val quotaManagers: QuotaManagers) {
-
-  def updateQuotaConfig(sanitizedUser: Option[String], sanitizedClientId: Option[String], config: Properties): Unit = {
-    val clientId = sanitizedClientId.map(Sanitizer.desanitize)
-    val producerQuota =
-      if (config.containsKey(QuotaConfig.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG))
-        Some(new Quota(config.getProperty(QuotaConfig.PRODUCER_BYTE_RATE_OVERRIDE_CONFIG).toLong.toDouble, true))
-      else
-        None
-    quotaManagers.produce.updateQuota(sanitizedUser, clientId, sanitizedClientId, producerQuota)
-    val consumerQuota =
-      if (config.containsKey(QuotaConfig.CONSUMER_BYTE_RATE_OVERRIDE_CONFIG))
-        Some(new Quota(config.getProperty(QuotaConfig.CONSUMER_BYTE_RATE_OVERRIDE_CONFIG).toLong.toDouble, true))
-      else
-        None
-    quotaManagers.fetch.updateQuota(sanitizedUser, clientId, sanitizedClientId, consumerQuota)
-    val requestQuota =
-      if (config.containsKey(QuotaConfig.REQUEST_PERCENTAGE_OVERRIDE_CONFIG))
-        Some(new Quota(config.getProperty(QuotaConfig.REQUEST_PERCENTAGE_OVERRIDE_CONFIG).toDouble, true))
-      else
-        None
-    quotaManagers.request.updateQuota(sanitizedUser, clientId, sanitizedClientId, requestQuota)
-    val controllerMutationQuota =
-      if (config.containsKey(QuotaConfig.CONTROLLER_MUTATION_RATE_OVERRIDE_CONFIG))
-        Some(new Quota(config.getProperty(QuotaConfig.CONTROLLER_MUTATION_RATE_OVERRIDE_CONFIG).toDouble, true))
-      else
-        None
-    quotaManagers.controllerMutation.updateQuota(sanitizedUser, clientId, sanitizedClientId, controllerMutationQuota)
-  }
-}
-
-class IpConfigHandler(private val connectionQuotas: ConnectionQuotas) extends ConfigHandler with Logging {
-
-  def processConfigChanges(ip: String, config: Properties): Unit = {
-    val ipConnectionRateQuota = Option(config.getProperty(QuotaConfig.IP_CONNECTION_RATE_OVERRIDE_CONFIG)).map(_.toInt)
-    val updatedIp = {
-      if (ip != ZooKeeperInternals.DEFAULT_STRING) {
-        try {
-          Some(InetAddress.getByName(ip))
-        } catch {
-          case _: UnknownHostException => throw new IllegalArgumentException(s"Unable to resolve address $ip")
-        }
-      } else
-        None
-    }
-    connectionQuotas.updateIpConnectionRateQuota(updatedIp, ipConnectionRateQuota)
-  }
-}
-
 /**
   * The BrokerConfigHandler will process individual broker config changes in ZK.
   * The callback provides the brokerId and the full properties set read from ZK.

core/src/main/scala/kafka/server/metadata/ClientQuotaMetadataManager.scala

@@ -105,7 +105,7 @@ class ClientQuotaMetadataManager(private[metadata] val quotaManagers: QuotaManag
     }
   }
 
-  private def handleIpQuota(ipEntity: QuotaEntity, quotaDelta: ClientQuotaDelta): Unit = {
+  private[metadata] def handleIpQuota(ipEntity: QuotaEntity, quotaDelta: ClientQuotaDelta): Unit = {
     val inetAddress = ipEntity match {
       case IpEntity(ip) =>
         try {

core/src/test/scala/kafka/server/metadata/ClientQuotaMetadataManagerTest.scala

@@ -0,0 +1,36 @@
+/*
+ * 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 kafka.server.metadata
+
+import org.apache.kafka.image.ClientQuotaDelta
+import org.junit.jupiter.api.Assertions.{assertDoesNotThrow, assertThrows}
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.function.Executable
+
+
+class ClientQuotaMetadataManagerTest {
+
+  @Test
+  def testHandleIpQuota(): Unit = {
+    val manager = new ClientQuotaMetadataManager(null, null)
+    assertThrows(classOf[IllegalArgumentException], () => manager.handleIpQuota(IpEntity("a"), new ClientQuotaDelta(null)))
+    assertThrows(classOf[IllegalStateException], () => manager.handleIpQuota(UserEntity("a"), new ClientQuotaDelta(null)))
+    assertDoesNotThrow { new Executable { def execute(): Unit = manager.handleIpQuota(DefaultIpEntity, new ClientQuotaDelta(null)) } }
+    assertDoesNotThrow { new Executable { def execute(): Unit = manager.handleIpQuota(IpEntity("192.168.1.1"), new ClientQuotaDelta(null)) } }
+    assertDoesNotThrow { new Executable { def execute(): Unit = manager.handleIpQuota(IpEntity("2001:db8::1"), new ClientQuotaDelta(null)) } }
+  }
+}

core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala

@@ -486,14 +486,6 @@ class DynamicConfigChangeTest extends KafkaServerTestHarness {
 }
 
 class DynamicConfigChangeUnitTest {
-  @Test
-  def testIpHandlerUnresolvableAddress(): Unit = {
-    val configHandler = new IpConfigHandler(null)
-    val props: Properties = new Properties()
-    props.put(QuotaConfig.IP_CONNECTION_RATE_OVERRIDE_CONFIG, "1")
-
-    assertThrows(classOf[IllegalArgumentException], () => configHandler.processConfigChanges("illegal-hostname", props))
-  }
 
   @Test
   def shouldParseReplicationQuotaProperties(): Unit = {