From 296330b879b3458fd10c780cf10f3ce3a3d0e932 Mon Sep 17 00:00:00 2001
From: David Mao <dmao@confluent.io>
Date: Mon, 5 Oct 2020 15:42:41 -0700
Subject: [PATCH 1/9] Alter and describe IP entity with client quota API

---
 .../kafka/common/quota/ClientQuotaEntity.java |   1 +
 .../scala/kafka/server/AdminManager.scala     | 127 ++++++++++-----
 .../kafka/server/DynamicConfigManager.scala   |   2 +-
 .../network/DynamicConnectionQuotaTest.scala  |  15 +-
 .../server/ClientQuotasRequestTest.scala      | 146 +++++++++++++++---
 .../scala/unit/kafka/utils/TestUtils.scala    |  11 ++
 6 files changed, 239 insertions(+), 63 deletions(-)

diff --git a/clients/src/main/java/org/apache/kafka/common/quota/ClientQuotaEntity.java b/clients/src/main/java/org/apache/kafka/common/quota/ClientQuotaEntity.java
index 0fee3d3bed106..94fc858293751 100644
--- a/clients/src/main/java/org/apache/kafka/common/quota/ClientQuotaEntity.java
+++ b/clients/src/main/java/org/apache/kafka/common/quota/ClientQuotaEntity.java
@@ -32,6 +32,7 @@ public class ClientQuotaEntity {
      */
     public static final String USER = "user";
     public static final String CLIENT_ID = "client-id";
+    public static final String IP = "IP";
 
     /**
      * Constructs a quota entity for the given types and names. If a name is null,
diff --git a/core/src/main/scala/kafka/server/AdminManager.scala b/core/src/main/scala/kafka/server/AdminManager.scala
index f3ff6697b7bf4..4e6598be10984 100644
--- a/core/src/main/scala/kafka/server/AdminManager.scala
+++ b/core/src/main/scala/kafka/server/AdminManager.scala
@@ -727,7 +727,7 @@ class AdminManager(val config: KafkaConfig,
       case _ => DescribeConfigsResponse.ConfigType.UNKNOWN
     }
   }
-  
+
   private def configSynonyms(name: String, synonyms: List[String], isSensitive: Boolean): List[DescribeConfigsResponseData.DescribeConfigsSynonym] = {
     val dynamicConfig = config.dynamicConfig
     val allSynonyms = mutable.Buffer[DescribeConfigsResponseData.DescribeConfigsSynonym]()
@@ -807,23 +807,25 @@ class AdminManager(val config: KafkaConfig,
       case name => Sanitizer.desanitize(name)
     }
 
-  private def entityToSanitizedUserClientId(entity: ClientQuotaEntity): (Option[String], Option[String]) = {
+  private def entityToSanitizedUserClientId(entity: ClientQuotaEntity): (Option[String], Option[String], Option[String]) = {
     if (entity.entries.isEmpty)
       throw new InvalidRequestException("Invalid empty client quota entity")
 
     var user: Option[String] = None
     var clientId: Option[String] = None
+    var ip: Option[String] = None
     entity.entries.forEach { (entityType, entityName) =>
       val sanitizedEntityName = Some(sanitizeEntityName(entityName))
       entityType match {
         case ClientQuotaEntity.USER => user = sanitizedEntityName
         case ClientQuotaEntity.CLIENT_ID => clientId = sanitizedEntityName
+        case ClientQuotaEntity.IP => ip = sanitizedEntityName
         case _ => throw new InvalidRequestException(s"Unhandled client quota entity type: ${entityType}")
       }
       if (entityName != null && entityName.isEmpty)
         throw new InvalidRequestException(s"Empty ${entityType} not supported")
     }
-    (user, clientId)
+    (user, clientId, ip)
   }
 
   private def userClientIdToEntity(user: Option[String], clientId: Option[String]): ClientQuotaEntity = {
@@ -833,16 +835,21 @@ class AdminManager(val config: KafkaConfig,
   def describeClientQuotas(filter: ClientQuotaFilter): Map[ClientQuotaEntity, Map[String, Double]] = {
     var userComponent: Option[ClientQuotaFilterComponent] = None
     var clientIdComponent: Option[ClientQuotaFilterComponent] = None
+    var ipComponent: Option[ClientQuotaFilterComponent] = None
     filter.components.forEach { component =>
       component.entityType match {
         case ClientQuotaEntity.USER =>
           if (userComponent.isDefined)
-            throw new InvalidRequestException(s"Duplicate user filter component entity type");
+            throw new InvalidRequestException(s"Duplicate user filter component entity type")
           userComponent = Some(component)
         case ClientQuotaEntity.CLIENT_ID =>
           if (clientIdComponent.isDefined)
-            throw new InvalidRequestException(s"Duplicate client filter component entity type");
+            throw new InvalidRequestException(s"Duplicate client filter component entity type")
           clientIdComponent = Some(component)
+        case ClientQuotaEntity.IP =>
+          if (ipComponent.isDefined)
+            throw new InvalidRequestException(s"Duplicate ip filter component entity type")
+          ipComponent = Some(component)
         case "" =>
           throw new InvalidRequestException(s"Unexpected empty filter component entity type")
         case et =>
@@ -850,28 +857,54 @@ class AdminManager(val config: KafkaConfig,
           throw new UnsupportedVersionException(s"Custom entity type '${et}' not supported")
       }
     }
-    handleDescribeClientQuotas(userComponent, clientIdComponent, filter.strict)
+    if ((userComponent.isDefined || clientIdComponent.isDefined) && ipComponent.isDefined)
+      throw new InvalidRequestException(s"Invalid entity filter component combination")
+
+    val userClientQuotas = if (ipComponent.isEmpty)
+      handleDescribeClientQuotas(userComponent, clientIdComponent, filter.strict)
+    else
+      Map.empty
+
+    val ipQuotas = if (userComponent.isEmpty && clientIdComponent.isEmpty)
+      handleDescribeIpQuotas(ipComponent, filter.strict)
+    else
+      Map.empty
+
+    (userClientQuotas ++ ipQuotas).toMap
+  }
+
+  private def wantExact(component: Option[ClientQuotaFilterComponent]): Boolean = component.exists(_.`match` != null)
+
+  private def toOption(opt: java.util.Optional[String]): Option[String] = {
+    if (opt == null)
+      None
+    else if (opt.isPresent)
+      Some(opt.get)
+    else
+      Some(null)
+  }
+
+  private def sanitized(name: Option[String]): String = name.map(n => sanitizeEntityName(n)).getOrElse("")
+
+  private def fromProps(props: Map[String, String]): Map[String, Double] = {
+    props.map { case (key, value) =>
+      val doubleValue = try value.toDouble catch {
+        case _: NumberFormatException =>
+          throw new IllegalStateException(s"Unexpected client quota configuration value: $key -> $value")
+      }
+      key -> doubleValue
+    }
   }
 
   def handleDescribeClientQuotas(userComponent: Option[ClientQuotaFilterComponent],
     clientIdComponent: Option[ClientQuotaFilterComponent], strict: Boolean): Map[ClientQuotaEntity, Map[String, Double]] = {
 
-    def toOption(opt: java.util.Optional[String]): Option[String] =
-      if (opt == null)
-        None
-      else if (opt.isPresent)
-        Some(opt.get)
-      else
-        Some(null)
-
     val user = userComponent.flatMap(c => toOption(c.`match`))
     val clientId = clientIdComponent.flatMap(c => toOption(c.`match`))
 
-    def sanitized(name: Option[String]): String = name.map(n => sanitizeEntityName(n)).getOrElse("")
     val sanitizedUser = sanitized(user)
     val sanitizedClientId = sanitized(clientId)
 
-    def wantExact(component: Option[ClientQuotaFilterComponent]): Boolean = component.exists(_.`match` != null)
     val exactUser = wantExact(userComponent)
     val exactClientId = wantExact(clientIdComponent)
 
@@ -920,32 +953,49 @@ class AdminManager(val config: KafkaConfig,
         !name.isDefined || !strict
     }
 
-    def fromProps(props: Map[String, String]): Map[String, Double] = {
-      props.map { case (key, value) =>
-        val doubleValue = try value.toDouble catch {
-          case _: NumberFormatException =>
-            throw new IllegalStateException(s"Unexpected client quota configuration value: $key -> $value")
-        }
-        key -> doubleValue
-      }
-    }
-
-    (userEntries ++ clientIdEntries ++ bothEntries).map { case ((u, c), p) =>
+    (userEntries ++ clientIdEntries ++ bothEntries).flatMap { case ((u, c), p) =>
       val quotaProps = p.asScala.filter { case (key, _) => QuotaConfigs.isQuotaConfig(key) }
       if (quotaProps.nonEmpty && matches(userComponent, u) && matches(clientIdComponent, c))
         Some(userClientIdToEntity(u, c) -> fromProps(quotaProps))
       else
         None
-    }.flatten.toMap
+    }.toMap
+  }
+
+  def handleDescribeIpQuotas(ipComponent: Option[ClientQuotaFilterComponent], strict: Boolean): Map[ClientQuotaEntity, Map[String, Double]] = {
+    val ip = ipComponent.flatMap(c => toOption(c.`match`))
+    val exactIp = wantExact(ipComponent)
+    val allIps = ipComponent.exists(_.`match` == null) || (ipComponent.isEmpty && !strict)
+    val ipEntries = if (exactIp)
+      Map(Some(ip.get) -> adminZkClient.fetchEntityConfig(ConfigType.Ip, sanitized(ip)))
+    else if (allIps)
+      adminZkClient.fetchAllEntityConfigs(ConfigType.Ip).map { case (name, props) =>
+        Some(desanitizeEntityName(name)) -> props
+      }
+    else
+      Map.empty
+
+    def ipToQuotaEntity(ip: Option[String]): ClientQuotaEntity = {
+      new ClientQuotaEntity(ip.map(ipName => ClientQuotaEntity.IP -> ipName).toMap.asJava)
+    }
+
+    ipEntries.flatMap { case (ip, props) =>
+      val ipQuotaProps = props.asScala.filter { case (key, _) => DynamicConfig.Ip.names.contains(key) }
+      if (ipQuotaProps.nonEmpty)
+        Some(ipToQuotaEntity(ip) -> fromProps(ipQuotaProps))
+      else
+        None
+    }
   }
 
   def alterClientQuotas(entries: Seq[ClientQuotaAlteration], validateOnly: Boolean): Map[ClientQuotaEntity, ApiError] = {
     def alterEntityQuotas(entity: ClientQuotaEntity, ops: Iterable[ClientQuotaAlteration.Op]): Unit = {
       val (path, configType, configKeys) = entityToSanitizedUserClientId(entity) match {
-        case (Some(user), Some(clientId)) => (user + "/clients/" + clientId, ConfigType.User, DynamicConfig.User.configKeys)
-        case (Some(user), None) => (user, ConfigType.User, DynamicConfig.User.configKeys)
-        case (None, Some(clientId)) => (clientId, ConfigType.Client, DynamicConfig.Client.configKeys)
-        case _ => throw new InvalidRequestException("Invalid empty client quota entity")
+        case (Some(user), Some(clientId), None) => (user + "/clients/" + clientId, ConfigType.User, DynamicConfig.User.configKeys)
+        case (Some(user), None, None) => (user, ConfigType.User, DynamicConfig.User.configKeys)
+        case (None, Some(clientId), None) => (clientId, ConfigType.Client, DynamicConfig.Client.configKeys)
+        case (None, None, Some(ip)) => (ip, ConfigType.Ip, DynamicConfig.Ip.configKeys)
+        case _ => throw new InvalidRequestException("Invalid client quota entity")
       }
 
       val props = adminZkClient.fetchEntityConfig(configType, path)
@@ -959,12 +1009,15 @@ class AdminManager(val config: KafkaConfig,
             case key => key.`type` match {
               case ConfigDef.Type.DOUBLE =>
                 props.setProperty(op.key, value.toString)
-              case ConfigDef.Type.LONG =>
+              case ConfigDef.Type.LONG | ConfigDef.Type.INT =>
                 val epsilon = 1e-6
-                val longValue = (value + epsilon).toLong
-                if ((longValue.toDouble - value).abs > epsilon)
-                  throw new InvalidRequestException(s"Configuration ${op.key} must be a Long value")
-                props.setProperty(op.key, longValue.toString)
+                val intValue = if (key.`type` == ConfigDef.Type.LONG)
+                  (value + epsilon).toLong
+                else
+                  (value + epsilon).toInt
+                if ((intValue.toDouble - value).abs > epsilon)
+                  throw new InvalidRequestException(s"Configuration ${op.key} must be a ${key.`type`} value")
+                props.setProperty(op.key, intValue.toString)
               case _ =>
                 throw new IllegalStateException(s"Unexpected config type ${key.`type`}")
             }
diff --git a/core/src/main/scala/kafka/server/DynamicConfigManager.scala b/core/src/main/scala/kafka/server/DynamicConfigManager.scala
index 3eed3827782d7..67c8182bafb72 100644
--- a/core/src/main/scala/kafka/server/DynamicConfigManager.scala
+++ b/core/src/main/scala/kafka/server/DynamicConfigManager.scala
@@ -38,7 +38,7 @@ object ConfigType {
   val Client = "clients"
   val User = "users"
   val Broker = "brokers"
-  val Ip = "ips"
+  val Ip = "IPs"
   val all = Seq(Topic, Client, User, Broker, Ip)
 }
 
diff --git a/core/src/test/scala/integration/kafka/network/DynamicConnectionQuotaTest.scala b/core/src/test/scala/integration/kafka/network/DynamicConnectionQuotaTest.scala
index b5cd7f6c7d056..df1e603901519 100644
--- a/core/src/test/scala/integration/kafka/network/DynamicConnectionQuotaTest.scala
+++ b/core/src/test/scala/integration/kafka/network/DynamicConnectionQuotaTest.scala
@@ -23,12 +23,13 @@ import java.net.{InetAddress, Socket}
 import java.util.concurrent._
 import java.util.{Collections, Properties}
 
-import kafka.server.{BaseRequestTest, ConfigEntityName, DynamicConfig, KafkaConfig}
-import kafka.utils.{CoreUtils, TestUtils}
+import kafka.server.{BaseRequestTest, DynamicConfig, KafkaConfig}
+import kafka.utils.TestUtils
 import org.apache.kafka.clients.admin.{Admin, AdminClientConfig}
 import org.apache.kafka.common.message.ProduceRequestData
 import org.apache.kafka.common.network.ListenerName
 import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.quota.ClientQuotaEntity
 import org.apache.kafka.common.record.{CompressionType, MemoryRecords, SimpleRecord}
 import org.apache.kafka.common.requests.{ProduceRequest, ProduceResponse}
 import org.apache.kafka.common.security.auth.SecurityProtocol
@@ -259,13 +260,19 @@ class DynamicConnectionQuotaTest extends BaseRequestTest {
   }
 
   private def updateIpConnectionRate(ip: Option[String], updatedRate: Int): Unit = {
-    adminZkClient.changeIpConfig(ip.getOrElse(ConfigEntityName.Default),
-      CoreUtils.propsWith(DynamicConfig.Ip.IpConnectionRateOverrideProp, updatedRate.toString))
+    val initialConnectionCount = connectionCount
+    val adminClient = createAdminClient()
+    val entity = new ClientQuotaEntity(Map(ClientQuotaEntity.IP -> ip.getOrElse(null)).asJava)
+    val request = Map(entity -> Map(DynamicConfig.Ip.IpConnectionRateOverrideProp -> Some(updatedRate.toDouble)))
+    TestUtils.alterClientQuotas(adminClient, request).all.get()
     // use a random throwaway address if ip isn't specified to get the default value
     TestUtils.waitUntilTrue(() => servers.head.socketServer.connectionQuotas.
       connectionRateForIp(InetAddress.getByName(ip.getOrElse("255.255.3.4"))) == updatedRate,
       s"Timed out waiting for connection rate update to propagate"
     )
+    adminClient.close()
+    TestUtils.waitUntilTrue(() => initialConnectionCount == connectionCount,
+      s"Admin client connection not closed (initial = $initialConnectionCount, current = $connectionCount)")
   }
 
   private def waitForListener(listenerName: String): Unit = {
diff --git a/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala b/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala
index f012e49dd1c87..2b9a75e9df3a4 100644
--- a/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala
@@ -17,6 +17,8 @@
 
 package kafka.server
 
+import java.net.InetAddress
+
 import org.apache.kafka.clients.admin.{ScramCredentialInfo, ScramMechanism, UserScramCredentialUpsertion}
 import org.apache.kafka.common.errors.{InvalidRequestException, UnsupportedVersionException}
 import org.apache.kafka.common.internals.KafkaFutureImpl
@@ -33,6 +35,7 @@ class ClientQuotasRequestTest extends BaseRequestTest {
   private val ConsumerByteRateProp = DynamicConfig.Client.ConsumerByteRateOverrideProp
   private val ProducerByteRateProp = DynamicConfig.Client.ProducerByteRateOverrideProp
   private val RequestPercentageProp = DynamicConfig.Client.RequestPercentageOverrideProp
+  private val IpConnectionRateProp = DynamicConfig.Ip.IpConnectionRateOverrideProp
 
   override val brokerCount = 1
 
@@ -187,6 +190,59 @@ class ClientQuotasRequestTest extends BaseRequestTest {
     ))
   }
 
+  @Test
+  def testAlterIpQuotasRequest(): Unit = {
+    val knownHost = "1.2.3.4"
+    val unknownHost = "2.3.4.5"
+    val entity = toIpEntity(Some(knownHost))
+    val defaultEntity = toIpEntity(Some(null))
+    val entityFilter = ClientQuotaFilterComponent.ofEntity(ClientQuotaEntity.IP, knownHost)
+    val defaultEntityFilter = ClientQuotaFilterComponent.ofDefaultEntity(ClientQuotaEntity.IP)
+    val allIpEntityFilter = ClientQuotaFilterComponent.ofEntityType(ClientQuotaEntity.IP)
+
+    def verifyIpQuotas(entityFilter: ClientQuotaFilterComponent, expectedMatches: Map[ClientQuotaEntity, Double]): Unit = {
+      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))
+        val entityName = entity.entries.get(ClientQuotaEntity.IP)
+        // ClientQuotaEntity with null name maps to default entity
+        val entityIp = if (entityName == null)
+          InetAddress.getByName(unknownHost)
+        else
+          InetAddress.getByName(entityName)
+        assertEquals(expectedMatches(entity), servers.head.socketServer.connectionQuotas.connectionRateForIp(entityIp), 0.01)
+      }
+    }
+
+    // Expect an empty configuration.
+    verifyIpQuotas(allIpEntityFilter, Map.empty)
+
+    // Add a configuration entry.
+    alterEntityQuotas(entity, Map(IpConnectionRateProp -> Some(100.0)), validateOnly = false)
+    verifyIpQuotas(entityFilter, Map(entity -> 100.0))
+
+    // update existing entry
+    alterEntityQuotas(entity, Map(IpConnectionRateProp -> Some(150.0)), validateOnly = false)
+    verifyIpQuotas(entityFilter, Map(entity -> 150.0))
+
+    // update default value
+    alterEntityQuotas(defaultEntity, Map(IpConnectionRateProp -> 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)
+    verifyIpQuotas(entityFilter, Map.empty)
+
+    // remove default value
+    alterEntityQuotas(defaultEntity, Map(IpConnectionRateProp -> None), validateOnly = false)
+    verifyIpQuotas(allIpEntityFilter, Map.empty)
+  }
+
   @Test(expected = classOf[InvalidRequestException])
   def testAlterClientQuotasBadUser(): Unit = {
     val entity = new ClientQuotaEntity(Map((ClientQuotaEntity.USER -> "")).asJava)
@@ -223,8 +279,26 @@ class ClientQuotasRequestTest extends BaseRequestTest {
     alterEntityQuotas(entity, Map((ProducerByteRateProp -> Some(10000.5))), validateOnly = true)
   }
 
+  @Test
+  def testAlterClientQuotasInvalidEntityCombination(): Unit = {
+    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)
+    assertThrows(classOf[InvalidRequestException], () => alterEntityQuotas(userAndIpEntity, Map(RequestPercentageProp -> Some(12.34)), validateOnly = true))
+    assertThrows(classOf[InvalidRequestException], () => alterEntityQuotas(clientAndIpEntity, Map(RequestPercentageProp -> Some(12.34)), validateOnly = true))
+  }
+
+  @Test
+  def testDescribeClientQuotasInvalidFilterCombination(): Unit = {
+    val ipFilterComponent = ClientQuotaFilterComponent.ofEntityType(ClientQuotaEntity.IP)
+    val userFilterComponent = ClientQuotaFilterComponent.ofEntityType(ClientQuotaEntity.USER)
+    val clientIdFilterComponent = ClientQuotaFilterComponent.ofEntityType(ClientQuotaEntity.CLIENT_ID)
+
+    assertThrows(classOf[InvalidRequestException], () => describeClientQuotas(ClientQuotaFilter.contains(List(ipFilterComponent, userFilterComponent).asJava)))
+    assertThrows(classOf[InvalidRequestException], () => describeClientQuotas(ClientQuotaFilter.contains(List(ipFilterComponent, clientIdFilterComponent).asJava)))
+  }
+
   // Entities to be matched against.
-  private val matchEntities = List(
+  private val matchUserClientEntities = List(
     (Some("user-1"), Some("client-id-1"), 50.50),
     (Some("user-2"), Some("client-id-1"), 51.51),
     (Some("user-3"), Some("client-id-2"), 52.52),
@@ -236,13 +310,22 @@ class ClientQuotasRequestTest extends BaseRequestTest {
     (Some("user-3"), None, 58.58),
     (Some(null), None, 59.59),
     (None, Some("client-id-2"), 60.60)
-  ).map { case (u, c, v) => (toEntity(u, c), v) }
+  ).map { case (u, c, v) => (toClientEntity(u, c), v) }
+
+  private val matchIpEntities = List(
+    (Some("1.2.3.4"), 10.0),
+    (Some("2.3.4.5"), 20.0)
+  ).map { case (ip, quota) => (toIpEntity(ip), quota)}
 
   private def setupDescribeClientQuotasMatchTest() = {
-    val result = alterClientQuotas(matchEntities.map { case (e, v) =>
-      (e -> Map((RequestPercentageProp, Some(v))))
-    }.toMap, validateOnly = false)
-    matchEntities.foreach(e => result.get(e._1).get.get(10, TimeUnit.SECONDS))
+    val userClientQuotas = matchUserClientEntities.map { case (e, v) =>
+      e -> Map((RequestPercentageProp, Some(v)))
+    }.toMap
+    val ipQuotas = matchIpEntities.map { case (e, v) =>
+      e -> Map((IpConnectionRateProp, Some(v)))
+    }.toMap
+    val result = alterClientQuotas(userClientQuotas ++ ipQuotas, validateOnly = false)
+    (matchUserClientEntities ++ matchIpEntities).foreach(e => result(e._1).get(10, TimeUnit.SECONDS))
 
     // Allow time for watch callbacks to be triggered.
     Thread.sleep(500)
@@ -263,7 +346,7 @@ class ClientQuotasRequestTest extends BaseRequestTest {
     }
 
     // Test exact matches.
-    matchEntities.foreach { case (e, v) =>
+    matchUserClientEntities.foreach { case (e, v) =>
       val result = matchEntity(e)
       assertEquals(1, result.size)
       assertTrue(result.get(e) != null)
@@ -299,19 +382,30 @@ class ClientQuotasRequestTest extends BaseRequestTest {
 
     def testMatchEntities(filter: ClientQuotaFilter, expectedMatchSize: Int, partition: ClientQuotaEntity => Boolean): Unit = {
       val result = describeClientQuotas(filter)
-      val (expectedMatches, expectedNonMatches) = matchEntities.partition(e => partition(e._1))
+      val (expectedMatches, _) = (matchUserClientEntities ++ matchIpEntities).partition(e => partition(e._1))
       assertEquals(expectedMatchSize, expectedMatches.size)  // for test verification
       assertEquals(expectedMatchSize, result.size)
       val expectedMatchesMap = expectedMatches.toMap
-      matchEntities.foreach { case (entity, expectedValue) =>
+      matchUserClientEntities.foreach { case (entity, expectedValue) =>
         if (expectedMatchesMap.contains(entity)) {
           val config = result.get(entity)
-          assertTrue(config != null)
+          assertNotNull(config)
           val value = config.get(RequestPercentageProp)
-          assertTrue(value != null)
+          assertNotNull(value)
+          assertEquals(expectedValue, value, 1e-6)
+        } else {
+          assertNull(result.get(entity))
+        }
+      }
+      matchIpEntities.foreach { case (entity, expectedValue) =>
+        if (expectedMatchesMap.contains(entity)) {
+          val config = result.get(entity)
+          assertNotNull(config)
+          val value = config.get(IpConnectionRateProp)
+          assertNotNull(value)
           assertEquals(expectedValue, value, 1e-6)
         } else {
-          assertTrue(result.get(entity) == null)
+          assertNull(result.get(entity))
         }
       }
     }
@@ -376,8 +470,14 @@ class ClientQuotasRequestTest extends BaseRequestTest {
       entity => entity.entries.containsKey(ClientQuotaEntity.CLIENT_ID)
     )
 
+    // Match against all entities with IP type in an open-ended match.
+    testMatchEntities(
+      ClientQuotaFilter.contains(List(ClientQuotaFilterComponent.ofEntityType(ClientQuotaEntity.IP)).asJava), 2,
+      entity => entity.entries.containsKey(ClientQuotaEntity.IP)
+    )
+
     // Match open-ended empty filter list. This should match all entities.
-    testMatchEntities(ClientQuotaFilter.contains(List.empty.asJava), 11, entity => true)
+    testMatchEntities(ClientQuotaFilter.contains(List.empty.asJava), 13, entity => true)
 
     // Match close-ended empty filter list. This should match no entities.
     testMatchEntities(ClientQuotaFilter.containsOnly(List.empty.asJava), 0, entity => false)
@@ -386,11 +486,7 @@ class ClientQuotasRequestTest extends BaseRequestTest {
   @Test
   def testClientQuotasUnsupportedEntityTypes(): Unit = {
     val entity = new ClientQuotaEntity(Map(("other" -> "name")).asJava)
-    try {
-      verifyDescribeEntityQuotas(entity, Map())
-    } catch {
-      case e: ExecutionException => assertTrue(e.getCause.isInstanceOf[UnsupportedVersionException])
-    }
+    assertThrows(classOf[UnsupportedVersionException], () => verifyDescribeEntityQuotas(entity, Map.empty))
   }
 
   @Test
@@ -422,7 +518,11 @@ class ClientQuotasRequestTest extends BaseRequestTest {
   }
 
   private def verifyDescribeEntityQuotas(entity: ClientQuotaEntity, quotas: Map[String, Double]) = {
-    val components = entity.entries.asScala.map(e => ClientQuotaFilterComponent.ofEntity(e._1, e._2))
+    val components = entity.entries.asScala.map { case (entityType, entityName) =>
+      Option(entityName).map{ name => ClientQuotaFilterComponent.ofEntity(entityType, name)}
+        .getOrElse(ClientQuotaFilterComponent.ofDefaultEntity(entityType)
+      )
+    }
     val describe = describeClientQuotas(ClientQuotaFilter.containsOnly(components.toList.asJava))
     if (quotas.isEmpty) {
       assertEquals(0, describe.size)
@@ -439,13 +539,17 @@ class ClientQuotasRequestTest extends BaseRequestTest {
     }
   }
 
-  private def toEntity(user: Option[String], clientId: Option[String]) =
+  private def toClientEntity(user: Option[String], clientId: Option[String]) =
     new ClientQuotaEntity((user.map((ClientQuotaEntity.USER -> _)) ++ clientId.map((ClientQuotaEntity.CLIENT_ID -> _))).toMap.asJava)
 
+  private def toIpEntity(ip: Option[String]) = new ClientQuotaEntity(ip.map(ClientQuotaEntity.IP -> _).toMap.asJava)
+
   private def describeClientQuotas(filter: ClientQuotaFilter) = {
     val result = new KafkaFutureImpl[java.util.Map[ClientQuotaEntity, java.util.Map[String, java.lang.Double]]]
     sendDescribeClientQuotasRequest(filter).complete(result)
-    result.get
+    try result.get catch {
+      case e: ExecutionException => throw e.getCause
+    }
   }
 
   private def sendDescribeClientQuotasRequest(filter: ClientQuotaFilter): DescribeClientQuotasResponse = {
diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
index 9ff32227e081f..418bcacc83cf6 100755
--- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala
+++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
@@ -49,6 +49,7 @@ import org.apache.kafka.common.errors.{KafkaStorageException, UnknownTopicOrPart
 import org.apache.kafka.common.header.Header
 import org.apache.kafka.common.internals.Topic
 import org.apache.kafka.common.network.{ListenerName, Mode}
+import org.apache.kafka.common.quota.{ClientQuotaAlteration, ClientQuotaEntity}
 import org.apache.kafka.common.record._
 import org.apache.kafka.common.resource.ResourcePattern
 import org.apache.kafka.common.security.auth.SecurityProtocol
@@ -1529,6 +1530,16 @@ object TestUtils extends Logging {
     adminClient.incrementalAlterConfigs(configs)
   }
 
+  def alterClientQuotas(adminClient: Admin, request: Map[ClientQuotaEntity, Map[String, Option[Double]]]): AlterClientQuotasResult = {
+    val entries = request.map { case (entity, alter) =>
+      val ops = alter.map { case (key, value) =>
+        new ClientQuotaAlteration.Op(key, value.map(Double.box).getOrElse(null))
+      }.asJavaCollection
+        new ClientQuotaAlteration(entity, ops)
+    }.asJavaCollection
+    adminClient.alterClientQuotas(entries)
+  }
+
   def assertLeader(client: Admin, topicPartition: TopicPartition, expectedLeader: Int): Unit = {
     waitForLeaderToBecome(client, topicPartition, Some(expectedLeader))
   }

From cf888a9ac80f85ecb924cd41372c6f67a4d328d3 Mon Sep 17 00:00:00 2001
From: David Mao <dmao@confluent.io>
Date: Tue, 10 Nov 2020 19:11:09 -0800
Subject: [PATCH 2/9] Adds IP entity to config command

---
 .../scala/kafka/admin/ConfigCommand.scala     |  79 +++++---
 .../unit/kafka/admin/ConfigCommandTest.scala  | 190 +++++++++++++++++-
 2 files changed, 234 insertions(+), 35 deletions(-)

diff --git a/core/src/main/scala/kafka/admin/ConfigCommand.scala b/core/src/main/scala/kafka/admin/ConfigCommand.scala
index eb420dd4faa50..2073e6cbf245e 100644
--- a/core/src/main/scala/kafka/admin/ConfigCommand.scala
+++ b/core/src/main/scala/kafka/admin/ConfigCommand.scala
@@ -46,7 +46,7 @@ import scala.jdk.CollectionConverters._
 import scala.collection._
 
 /**
- * This script can be used to change configs for topics/clients/users/brokers dynamically
+ * This script can be used to change configs for topics/clients/users/brokers/IPs dynamically
  * An entity described or altered by the command may be one of:
  * <ul>
  *     <li> topic: --topic <topic> OR --entity-type topics --entity-name <topic>
@@ -56,11 +56,12 @@ import scala.collection._
  *                          --entity-type users --entity-name <user-principal> --entity-type clients --entity-name <client-id>
  *     <li> broker: --broker <broker-id> OR --entity-type brokers --entity-name <broker-id>
  *     <li> broker-logger: --broker-logger <broker-id> OR --entity-type broker-loggers --entity-name <broker-id>
+ *     <li> ip: --IP <ip> OR --entity-type IPs --entity-name <ip>
  * </ul>
- * --entity-type <users|clients|brokers> --entity-default may be specified in place of --entity-type <users|clients|brokers> --entity-name <entityName>
- * when describing or altering default configuration for users, clients, or brokers, respectively.
- * Alternatively, --user-defaults, --client-defaults, or --broker-defaults may be specified in place of
- * --entity-type <users|clients|brokers> --entity-default, respectively.
+ * --entity-type <users|clients|brokers|IPs> --entity-default may be specified in place of --entity-type <users|clients|brokers|IPs> --entity-name <entityName>
+ * when describing or altering default configuration for users, clients, brokers, or IPs, respectively.
+ * Alternatively, --user-defaults, --client-defaults, --broker-defaults, or --IP-defaults may be specified in place of
+ * --entity-type <users|clients|brokers|IPs> --entity-default, respectively.
  */
 object ConfigCommand extends Config {
 
@@ -84,7 +85,7 @@ object ConfigCommand extends Config {
     try {
       val opts = new ConfigCommandOptions(args)
 
-      CommandLineUtils.printHelpAndExitIfNeeded(opts, "This tool helps to manipulate and describe entity config for a topic, client, user or broker")
+      CommandLineUtils.printHelpAndExitIfNeeded(opts, "This tool helps to manipulate and describe entity config for a topic, client, user, broker or IP")
 
       opts.checkArgs()
 
@@ -401,7 +402,11 @@ object ConfigCommand extends Config {
             throw new IllegalStateException(s"Altering user SCRAM credentials should never occur for more zero or multiple users: $entityNames")
           alterUserScramCredentialConfigs(adminClient, entityNames.head, scramConfigsToAddMap, scramConfigsToDelete)
         }
-
+      case ConfigType.Ip =>
+        val unknownConfigs = (configsToBeAdded.keys ++ configsToBeDeleted).filterNot(key => DynamicConfig.Ip.names.contains(key))
+        if (unknownConfigs.nonEmpty)
+          throw new IllegalArgumentException(s"Only connection quota configs can be added for '${ConfigType.Ip}' using --bootstrap-server. Unexpected config names: ${unknownConfigs.mkString(",")}")
+        alterQuotaConfigs(adminClient, entityTypes, entityNames, configsToBeAddedMap, configsToBeDeleted)
       case _ => throw new IllegalArgumentException(s"Unsupported entity type: $entityTypeHead")
     }
 
@@ -443,12 +448,11 @@ object ConfigCommand extends Config {
     if (invalidConfigs.nonEmpty)
       throw new InvalidConfigurationException(s"Invalid config(s): ${invalidConfigs.mkString(",")}")
 
-    val alterEntityTypes = entityTypes.map { entType =>
-      entType match {
-        case ConfigType.User => ClientQuotaEntity.USER
-        case ConfigType.Client => ClientQuotaEntity.CLIENT_ID
-        case _ => throw new IllegalArgumentException(s"Unexpected entity type: ${entType}")
-      }
+    val alterEntityTypes = entityTypes.map {
+      case ConfigType.User => ClientQuotaEntity.USER
+      case ConfigType.Client => ClientQuotaEntity.CLIENT_ID
+      case ConfigType.Ip => ClientQuotaEntity.IP
+      case entType => throw new IllegalArgumentException(s"Unexpected entity type: $entType")
     }
     val alterEntityNames = entityNames.map(en => if (en.nonEmpty) en else null)
 
@@ -461,7 +465,7 @@ object ConfigCommand extends Config {
     val alterOps = (configsToBeAddedMap.map { case (key, value) =>
       val doubleValue = try value.toDouble catch {
         case _: NumberFormatException =>
-          throw new IllegalArgumentException(s"Cannot parse quota configuration value for ${key}: ${value}")
+          throw new IllegalArgumentException(s"Cannot parse quota configuration value for $key: $value")
       }
       new ClientQuotaAlteration.Op(key, doubleValue)
     } ++ configsToBeDeleted.map(key => new ClientQuotaAlteration.Op(key, null))).asJavaCollection
@@ -480,6 +484,8 @@ object ConfigCommand extends Config {
         describeResourceConfig(adminClient, entityTypes.head, entityNames.headOption, describeAll)
       case ConfigType.User | ConfigType.Client =>
         describeClientQuotaAndUserScramCredentialConfigs(adminClient, entityTypes, entityNames)
+      case ConfigType.Ip =>
+        describeQuotaConfigs(adminClient, entityTypes, entityNames)
       case entityType => throw new IllegalArgumentException(s"Invalid entity type: $entityType")
     }
   }
@@ -551,7 +557,7 @@ object ConfigCommand extends Config {
       }).toSeq
   }
 
-  private def describeClientQuotaAndUserScramCredentialConfigs(adminClient: Admin, entityTypes: List[String], entityNames: List[String]) = {
+  private def describeQuotaConfigs(adminClient: Admin, entityTypes: List[String], entityNames: List[String]) = {
     val quotaConfigs = getAllClientQuotasConfigs(adminClient, entityTypes, entityNames)
     quotaConfigs.forKeyValue { (entity, entries) =>
       val entityEntries = entity.entries.asScala
@@ -561,15 +567,22 @@ object ConfigCommand extends Config {
           val typeStr = entityType match {
             case ClientQuotaEntity.USER => "user-principal"
             case ClientQuotaEntity.CLIENT_ID => "client-id"
+            case ClientQuotaEntity.IP => "IP"
           }
-          if (name != null) s"${typeStr} '${name}'"
-          else s"the default ${typeStr}"
+          if (name != null) s"$typeStr '$name'"
+          else s"the default $typeStr"
         }
 
-      val entityStr = (entitySubstr(ClientQuotaEntity.USER) ++ entitySubstr(ClientQuotaEntity.CLIENT_ID)).mkString(", ")
+      val entityStr = (entitySubstr(ClientQuotaEntity.USER) ++
+                       entitySubstr(ClientQuotaEntity.CLIENT_ID) ++
+                       entitySubstr(ClientQuotaEntity.IP)).mkString(", ")
       val entriesStr = entries.asScala.map(e => s"${e._1}=${e._2}").mkString(", ")
-      println(s"Quota configs for ${entityStr} are ${entriesStr}")
+      println(s"Quota configs for $entityStr are $entriesStr")
     }
+  }
+
+  private def describeClientQuotaAndUserScramCredentialConfigs(adminClient: Admin, entityTypes: List[String], entityNames: List[String]) = {
+    describeQuotaConfigs(adminClient, entityTypes, entityNames)
     // we describe user SCRAM credentials only when we are not describing client information
     // and we are not given either --entity-default or --user-defaults
     if (!entityTypes.contains(ConfigType.Client) && !entityNames.contains("")) {
@@ -597,6 +610,7 @@ object ConfigCommand extends Config {
       val entityType = entityTypeOpt match {
         case Some(ConfigType.User) => ClientQuotaEntity.USER
         case Some(ConfigType.Client) => ClientQuotaEntity.CLIENT_ID
+        case Some(ConfigType.Ip) => ClientQuotaEntity.IP
         case Some(_) => throw new IllegalArgumentException(s"Unexpected entity type ${entityTypeOpt.get}")
         case None => throw new IllegalArgumentException("More entity names specified than entity types")
       }
@@ -735,10 +749,10 @@ object ConfigCommand extends Config {
     val describeOpt = parser.accepts("describe", "List configs for the given entity.")
     val allOpt = parser.accepts("all", "List all configs for the given topic, broker, or broker-logger entity (includes static configuration when the entity type is brokers)")
 
-    val entityType = parser.accepts("entity-type", "Type of entity (topics/clients/users/brokers/broker-loggers)")
+    val entityType = parser.accepts("entity-type", "Type of entity (topics/clients/users/brokers/broker-loggers/IPs)")
             .withRequiredArg
             .ofType(classOf[String])
-    val entityName = parser.accepts("entity-name", "Name of entity (topic name/client id/user principal name/broker id)")
+    val entityName = parser.accepts("entity-name", "Name of entity (topic name/client id/user principal name/broker id/IP)")
             .withRequiredArg
             .ofType(classOf[String])
     val entityDefault = parser.accepts("entity-default", "Default entity name for clients/users/brokers (applies to corresponding entity type in command line)")
@@ -749,6 +763,7 @@ object ConfigCommand extends Config {
             "For entity-type '" + ConfigType.Broker + "': " + DynamicConfig.Broker.names.asScala.toSeq.sorted.map("\t" + _).mkString(nl, nl, nl) +
             "For entity-type '" + ConfigType.User + "': " + DynamicConfig.User.names.asScala.toSeq.sorted.map("\t" + _).mkString(nl, nl, nl) +
             "For entity-type '" + ConfigType.Client + "': " + DynamicConfig.Client.names.asScala.toSeq.sorted.map("\t" + _).mkString(nl, nl, nl) +
+            "For entity-type '" + ConfigType.Ip + "': " + DynamicConfig.Ip.names.asScala.toSeq.sorted.map("\t" + _).mkString(nl, nl, nl) +
             s"Entity types '${ConfigType.User}' and '${ConfigType.Client}' may be specified together to update config for clients of a specific user.")
             .withRequiredArg
             .ofType(classOf[String])
@@ -778,6 +793,10 @@ object ConfigCommand extends Config {
     val brokerLogger = parser.accepts("broker-logger", "The broker's ID for its logger config.")
       .withRequiredArg
       .ofType(classOf[String])
+    val ipDefaults = parser.accepts("IP-defaults", "The config defaults for all IPs.")
+    val ip = parser.accepts("IP", "The IP address.")
+      .withRequiredArg
+      .ofType(classOf[String])
     val zkTlsConfigFile = parser.accepts("zk-tls-config-file",
       "Identifies the file where ZooKeeper client TLS connectivity properties are defined.  Any properties other than " +
         KafkaConfig.ZkSslConfigToSystemPropertyMap.keys.toList.sorted.mkString(", ") + " are ignored.")
@@ -788,11 +807,13 @@ object ConfigCommand extends Config {
       (client, ConfigType.Client),
       (user, ConfigType.User),
       (broker, ConfigType.Broker),
-      (brokerLogger, BrokerLoggerConfigType))
+      (brokerLogger, BrokerLoggerConfigType),
+      (ip, ConfigType.Ip))
 
     private val entityDefaultsFlags = List((clientDefaults, ConfigType.Client),
       (userDefaults, ConfigType.User),
-      (brokerDefaults, ConfigType.Broker))
+      (brokerDefaults, ConfigType.Broker),
+      (ipDefaults, ConfigType.Ip))
 
     private[admin] def entityTypes: List[String] = {
       options.valuesOf(entityType).asScala.toList ++
@@ -864,11 +885,21 @@ object ConfigCommand extends Config {
         }
       }
 
+      if (hasEntityName && entityTypeVals.contains(ConfigType.Ip)) {
+        Seq(entityName, ip).filter(options.has(_)).map(options.valueOf(_)).foreach { ipAddress =>
+          if (!Utils.validHostPattern(ipAddress))
+            throw new IllegalArgumentException(s"The entity name for ${entityTypeVals.head} must be a valid IP, but it is: $ipAddress")
+        }
+      }
+
       if (options.has(describeOpt) && entityTypeVals.contains(BrokerLoggerConfigType) && !hasEntityName)
         throw new IllegalArgumentException(s"an entity name must be specified with --describe of ${entityTypeVals.mkString(",")}")
 
       if (options.has(alterOpt)) {
-        if (entityTypeVals.contains(ConfigType.User) || entityTypeVals.contains(ConfigType.Client) || entityTypeVals.contains(ConfigType.Broker)) {
+        if (entityTypeVals.contains(ConfigType.User) ||
+            entityTypeVals.contains(ConfigType.Client) ||
+            entityTypeVals.contains(ConfigType.Broker) ||
+            entityTypeVals.contains(ConfigType.Ip)) {
           if (!hasEntityName && !hasEntityDefault)
             throw new IllegalArgumentException("an entity-name or default entity must be specified with --alter of users, clients or brokers")
         } else if (!hasEntityName)
diff --git a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
index 212e0eb8979c3..eeb234ce49496 100644
--- a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
+++ b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
@@ -133,6 +133,16 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
     testArgumentParse("broker-loggers", zkConfig = false)
   }
 
+  @Test
+  def shouldParseArgumentsForIpEntityType(): Unit = {
+    testArgumentParse("IPs", zkConfig = false)
+  }
+
+  @Test
+  def shouldParseArgumentsForIpEntityTypeUsingZookeeper(): Unit = {
+    testArgumentParse("IPs", zkConfig = true)
+  }
+
   def testArgumentParse(entityType: String, zkConfig: Boolean): Unit = {
     val shortFlag: String = s"--${entityType.dropRight(1)}"
 
@@ -296,6 +306,7 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
 
     testExpectedEntityTypeNames(List(ConfigType.Topic), List("A"), "--entity-type", "topics", "--entity-name", "A")
     testExpectedEntityTypeNames(List(ConfigType.Broker), List("0"), "--entity-name", "0", "--entity-type", "brokers")
+    testExpectedEntityTypeNames(List(ConfigType.Ip), List("1.2.3.4"), "--entity-name", "1.2.3.4", "--entity-type", "IPs")
     testExpectedEntityTypeNames(List(ConfigType.User, ConfigType.Client), List("A", ""),
       "--entity-type", "users", "--entity-type", "clients", "--entity-name", "A", "--entity-default")
     testExpectedEntityTypeNames(List(ConfigType.User, ConfigType.Client), List("", "B"),
@@ -303,6 +314,7 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
 
     testExpectedEntityTypeNames(List(ConfigType.Topic), List("A"), "--topic", "A")
     testExpectedEntityTypeNames(List(ConfigType.Broker), List("0"), "--broker", "0")
+    testExpectedEntityTypeNames(List(ConfigType.Ip), List("1.2.3.4"), "--IP", "1.2.3.4")
     testExpectedEntityTypeNames(List(ConfigType.Client, ConfigType.User), List("B", "A"), "--client", "B", "--user", "A")
     testExpectedEntityTypeNames(List(ConfigType.Client, ConfigType.User), List("B", ""), "--client", "B", "--user-defaults")
     testExpectedEntityTypeNames(List(ConfigType.Client, ConfigType.User), List("A"),
@@ -311,6 +323,7 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
     testExpectedEntityTypeNames(List(ConfigType.Topic), List.empty, "--entity-type", "topics")
     testExpectedEntityTypeNames(List(ConfigType.User), List.empty, "--entity-type", "users")
     testExpectedEntityTypeNames(List(ConfigType.Broker), List.empty, "--entity-type", "brokers")
+    testExpectedEntityTypeNames(List(ConfigType.Ip), List.empty, "--entity-type", "IPs")
   }
 
   @Test
@@ -379,6 +392,13 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
     createOpts.checkArgs()
   }
 
+  @Test(expected = classOf[IllegalArgumentException])
+  def shouldFailIfInvalidHost(): Unit = {
+    val createOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092",
+      "--entity-name", "A,B", "--entity-type", "IPs", "--describe"))
+    createOpts.checkArgs()
+  }
+
   @Test
   def shouldAddClientConfigUsingZookeeper(): Unit = {
     val createOpts = new ConfigCommandOptions(Array("--zookeeper", zkConnect,
@@ -398,19 +418,39 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
     ConfigCommand.alterConfigWithZk(null, createOpts, new TestAdminZkClient(zkClient))
   }
 
-  def testShouldAddClientConfig(user: Option[String], clientId: Option[String]): Unit = {
-    def toValues(entityName: Option[String], entityType: String, command: String):
-        (Array[String], Option[String], Option[ClientQuotaFilterComponent]) = {
-      entityName match {
-        case Some(null) =>
-          (Array("--entity-type", command, "--entity-default"), Some(null),
-            Some(ClientQuotaFilterComponent.ofDefaultEntity(entityType)))
-        case Some(name) =>
-          (Array("--entity-type", command, "--entity-name", name), Some(name),
-            Some(ClientQuotaFilterComponent.ofEntity(entityType, name)))
-        case None => (Array.empty, None, None)
+  @Test
+  def shouldAddIpConfigsUsingZookeeper(): Unit = {
+    val createOpts = new ConfigCommandOptions(Array("--zookeeper", zkConnect,
+      "--entity-name", "1.2.3.4",
+      "--entity-type", "IPs",
+      "--alter",
+      "--add-config", "a=b,c=d"))
+
+    class TestAdminZkClient(zkClient: KafkaZkClient) extends AdminZkClient(zkClient) {
+      override def changeIpConfig(ip: String, configChange: Properties): Unit = {
+        assertEquals("1.2.3.4", ip)
+        assertEquals("b", configChange.get("a"))
+        assertEquals("d", configChange.get("c"))
       }
     }
+
+    ConfigCommand.alterConfigWithZk(null, createOpts, new TestAdminZkClient(zkClient))
+  }
+
+  private def toValues(entityName: Option[String], entityType: String, command: String):
+  (Array[String], Option[String], Option[ClientQuotaFilterComponent]) = {
+    entityName match {
+      case Some(null) =>
+        (Array("--entity-type", command, "--entity-default"), Some(null),
+          Some(ClientQuotaFilterComponent.ofDefaultEntity(entityType)))
+      case Some(name) =>
+        (Array("--entity-type", command, "--entity-name", name), Some(name),
+          Some(ClientQuotaFilterComponent.ofEntity(entityType, name)))
+      case None => (Array.empty, None, None)
+    }
+  }
+
+  def testShouldAddClientConfig(user: Option[String], clientId: Option[String]): Unit = {
     val (userArgs, userEntity, userComponent) = toValues(user, ClientQuotaEntity.USER, "users")
     val (clientIdArgs, clientIdEntity, clientIdComponent) = toValues(clientId, ClientQuotaEntity.CLIENT_ID, "clients")
 
@@ -472,6 +512,134 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
     EasyMock.reset(alterResult, describeResult)
   }
 
+  @Test
+  def shouldNotAlterNonQuotaIpConfigsUsingBootstrapServer(): Unit = {
+    // when using --bootstrap-server, it should be illegal to alter anything that is not a connection quota
+    // for IP entities
+    val node = new Node(1, "localhost", 9092)
+    val mockAdminClient = new MockAdminClient(util.Collections.singletonList(node), node)
+
+    def verifyCommand(entityType: String, alterOpts: String*): Unit = {
+      val opts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092",
+        "--entity-type", entityType, "--entity-name", "admin",
+        "--alter") ++ alterOpts)
+      val e = intercept[IllegalArgumentException] {
+        ConfigCommand.alterConfig(mockAdminClient, opts)
+      }
+      assertTrue(s"Unexpected exception: $e", e.getMessage.contains("some_config"))
+    }
+
+    verifyCommand("IPs", "--add-config", "connection_creation_rate=10000,some_config=10")
+    verifyCommand("IPs", "--add-config", "some_config=10")
+    verifyCommand("IPs", "--delete-config", "connection_creation_rate=10000,some_config=10")
+    verifyCommand("IPs", "--delete-config", "some_config=10")
+  }
+
+  @Test
+  def testDescribeIpConfigs(): Unit = {
+    def testShouldDescribeIpConfig(ip: Option[String]): Unit = {
+      val entityType = ClientQuotaEntity.IP
+      val (ipArgs, filterComponent) = ip match {
+        case Some(null) => (Array("--entity-default"), ClientQuotaFilterComponent.ofDefaultEntity(entityType))
+        case Some(addr) => (Array("--entity-name", addr), ClientQuotaFilterComponent.ofEntity(entityType, addr))
+        case None => (Array.empty[String], ClientQuotaFilterComponent.ofEntityType(entityType))
+      }
+
+      val describeOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092",
+        "--describe", "--entity-type", "IPs") ++ ipArgs)
+
+      val expectedFilter = ClientQuotaFilter.containsOnly(List(filterComponent).asJava)
+
+      var describedConfigs = false
+      val describeFuture = new KafkaFutureImpl[util.Map[ClientQuotaEntity, util.Map[String, java.lang.Double]]]
+      describeFuture.complete(Map.empty[ClientQuotaEntity, util.Map[String, java.lang.Double]].asJava)
+      val describeResult: DescribeClientQuotasResult = EasyMock.createNiceMock(classOf[DescribeClientQuotasResult])
+      EasyMock.expect(describeResult.entities()).andReturn(describeFuture)
+
+      val node = new Node(1, "localhost", 9092)
+      val mockAdminClient = new MockAdminClient(util.Collections.singletonList(node), node) {
+        override def describeClientQuotas(filter: ClientQuotaFilter, options: DescribeClientQuotasOptions): DescribeClientQuotasResult = {
+          assertTrue(filter.strict)
+          assertEquals(expectedFilter, filter)
+          describedConfigs = true
+          describeResult
+        }
+      }
+      EasyMock.replay(describeResult)
+      ConfigCommand.describeConfig(mockAdminClient, describeOpts)
+      assertTrue(describedConfigs)
+      EasyMock.reset(describeResult)
+    }
+
+    testShouldDescribeIpConfig(Some("1.2.3.4"))
+    testShouldDescribeIpConfig(Some(null))
+    testShouldDescribeIpConfig(None)
+  }
+
+  @Test
+  def testAlterIpConfig(): Unit = {
+    def testShouldAlterIpConfig(ip: Option[String], remove: Boolean): Unit = {
+      val (ipArgs, ipEntity, ipComponent) = toValues(ip, ClientQuotaEntity.IP, "IPs")
+      val alterOpts = if (remove)
+        Array("--delete-config", "connection_creation_rate")
+      else
+        Array("--add-config", "connection_creation_rate=100")
+      val createOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092",
+        "--alter") ++ ipArgs ++ alterOpts)
+
+      // Explicitly populate a HashMap to ensure nulls are recorded properly.
+      val entityMap = new java.util.HashMap[String, String]
+      ipEntity.foreach(u => entityMap.put(ClientQuotaEntity.IP, u))
+      val entity = new ClientQuotaEntity(entityMap)
+
+      var describedConfigs = false
+      val describeFuture = new KafkaFutureImpl[util.Map[ClientQuotaEntity, util.Map[String, java.lang.Double]]]
+      describeFuture.complete(Map(entity -> Map("connection_creation_rate" -> Double.box(50.0)).asJava).asJava)
+      val describeResult: DescribeClientQuotasResult = EasyMock.createNiceMock(classOf[DescribeClientQuotasResult])
+      EasyMock.expect(describeResult.entities()).andReturn(describeFuture)
+
+      var alteredConfigs = false
+      val alterFuture = new KafkaFutureImpl[Void]
+      alterFuture.complete(null)
+      val alterResult: AlterClientQuotasResult = EasyMock.createNiceMock(classOf[AlterClientQuotasResult])
+      EasyMock.expect(alterResult.all()).andReturn(alterFuture)
+
+      val node = new Node(1, "localhost", 9092)
+      val mockAdminClient = new MockAdminClient(util.Collections.singletonList(node), node) {
+        override def describeClientQuotas(filter: ClientQuotaFilter, options: DescribeClientQuotasOptions): DescribeClientQuotasResult = {
+          assertTrue(filter.strict)
+          assertEquals(ClientQuotaFilter.containsOnly(List(ipComponent.get).asJava), filter)
+          describedConfigs = true
+          describeResult
+        }
+
+        override def alterClientQuotas(entries: util.Collection[ClientQuotaAlteration], options: AlterClientQuotasOptions): AlterClientQuotasResult = {
+          assertFalse(options.validateOnly)
+          assertEquals(1, entries.size)
+          val alteration = entries.asScala.head
+          assertEquals(entity, alteration.entity)
+          val ops = alteration.ops.asScala
+          val expectedOps = if (remove)
+            Set(new ClientQuotaAlteration.Op("connection_creation_rate", null))
+          else
+            Set(new ClientQuotaAlteration.Op("connection_creation_rate", Double.box(100)))
+          assertEquals(expectedOps, ops.toSet)
+          alteredConfigs = true
+          alterResult
+        }
+      }
+      EasyMock.replay(alterResult, describeResult)
+      ConfigCommand.alterConfig(mockAdminClient, createOpts)
+      assertTrue(describedConfigs)
+      assertTrue(alteredConfigs)
+      EasyMock.reset(alterResult, describeResult)
+    }
+
+    testShouldAlterIpConfig(Some("1.2.3.4"), remove = false)
+    testShouldAlterIpConfig(Some("1.2.3.4"), remove = true)
+    testShouldAlterIpConfig(Some(null), remove = false)
+    testShouldAlterIpConfig(Some(null), remove = true)
+  }
 
   @Test
   def shouldAddClientConfig(): Unit = {

From cdd2364e2c538de77ab28d8da54a5f333a128f2c Mon Sep 17 00:00:00 2001
From: David Mao <dmao@confluent.io>
Date: Tue, 1 Dec 2020 12:18:33 -0600
Subject: [PATCH 3/9] Address feedback, fix test

---
 core/src/main/scala/kafka/server/AdminManager.scala  |  7 ++++---
 .../kafka/network/DynamicConnectionQuotaTest.scala   | 12 ++++++++----
 core/src/test/scala/unit/kafka/utils/TestUtils.scala |  2 +-
 3 files changed, 13 insertions(+), 8 deletions(-)

diff --git a/core/src/main/scala/kafka/server/AdminManager.scala b/core/src/main/scala/kafka/server/AdminManager.scala
index 4e6598be10984..94ba28416cf0c 100644
--- a/core/src/main/scala/kafka/server/AdminManager.scala
+++ b/core/src/main/scala/kafka/server/AdminManager.scala
@@ -807,7 +807,7 @@ class AdminManager(val config: KafkaConfig,
       case name => Sanitizer.desanitize(name)
     }
 
-  private def entityToSanitizedUserClientId(entity: ClientQuotaEntity): (Option[String], Option[String], Option[String]) = {
+  private def parseAndSanitizeQuotaEntity(entity: ClientQuotaEntity): (Option[String], Option[String], Option[String]) = {
     if (entity.entries.isEmpty)
       throw new InvalidRequestException("Invalid empty client quota entity")
 
@@ -858,7 +858,8 @@ class AdminManager(val config: KafkaConfig,
       }
     }
     if ((userComponent.isDefined || clientIdComponent.isDefined) && ipComponent.isDefined)
-      throw new InvalidRequestException(s"Invalid entity filter component combination")
+      throw new InvalidRequestException(s"Invalid entity filter component combination, IP filter component should not be used with " +
+        s"user or clientId filter component.")
 
     val userClientQuotas = if (ipComponent.isEmpty)
       handleDescribeClientQuotas(userComponent, clientIdComponent, filter.strict)
@@ -990,7 +991,7 @@ class AdminManager(val config: KafkaConfig,
 
   def alterClientQuotas(entries: Seq[ClientQuotaAlteration], validateOnly: Boolean): Map[ClientQuotaEntity, ApiError] = {
     def alterEntityQuotas(entity: ClientQuotaEntity, ops: Iterable[ClientQuotaAlteration.Op]): Unit = {
-      val (path, configType, configKeys) = entityToSanitizedUserClientId(entity) match {
+      val (path, configType, configKeys) = parseAndSanitizeQuotaEntity(entity) match {
         case (Some(user), Some(clientId), None) => (user + "/clients/" + clientId, ConfigType.User, DynamicConfig.User.configKeys)
         case (Some(user), None, None) => (user, ConfigType.User, DynamicConfig.User.configKeys)
         case (None, Some(clientId), None) => (clientId, ConfigType.Client, DynamicConfig.Client.configKeys)
diff --git a/core/src/test/scala/integration/kafka/network/DynamicConnectionQuotaTest.scala b/core/src/test/scala/integration/kafka/network/DynamicConnectionQuotaTest.scala
index df1e603901519..06dbaaf57b3d9 100644
--- a/core/src/test/scala/integration/kafka/network/DynamicConnectionQuotaTest.scala
+++ b/core/src/test/scala/integration/kafka/network/DynamicConnectionQuotaTest.scala
@@ -48,6 +48,7 @@ class DynamicConnectionQuotaTest extends BaseRequestTest {
   val topic = "test"
   val listener = ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT)
   val localAddress = InetAddress.getByName("127.0.0.1")
+  val unknownHost = "255.255.0.1"
   val plaintextListenerDefaultQuota = 30
   var executor: ExecutorService = _
 
@@ -237,16 +238,19 @@ class DynamicConnectionQuotaTest extends BaseRequestTest {
   @Test
   def testDynamicIpConnectionRateQuota(): Unit = {
     val connRateLimit = 10
+    val initialConnectionCount = connectionCount
     // before setting connection rate to 10, verify we can do at least double that by default (no limit)
-    verifyConnectionRate(2 * connRateLimit, Int.MaxValue, "PLAINTEXT", ignoreIOExceptions = false)
+    verifyConnectionRate(2 * connRateLimit, plaintextListenerDefaultQuota, "PLAINTEXT", ignoreIOExceptions = false)
+    waitForConnectionCount(initialConnectionCount)
     // set default IP connection rate quota, verify that we don't exceed the limit
     updateIpConnectionRate(None, connRateLimit)
     verifyConnectionRate(8, connRateLimit, "PLAINTEXT", ignoreIOExceptions = true)
-
+    waitForConnectionCount(initialConnectionCount)
     // set a higher IP connection rate quota override, verify that the higher limit is now enforced
     val newRateLimit = 18
     updateIpConnectionRate(Some(localAddress.getHostAddress), newRateLimit)
     verifyConnectionRate(14, newRateLimit, "PLAINTEXT", ignoreIOExceptions = true)
+    waitForConnectionCount(initialConnectionCount)
   }
 
   private def reconfigureServers(newProps: Properties, perBrokerConfig: Boolean, aPropToVerify: (String, String)): Unit = {
@@ -262,12 +266,12 @@ class DynamicConnectionQuotaTest extends BaseRequestTest {
   private def updateIpConnectionRate(ip: Option[String], updatedRate: Int): Unit = {
     val initialConnectionCount = connectionCount
     val adminClient = createAdminClient()
-    val entity = new ClientQuotaEntity(Map(ClientQuotaEntity.IP -> ip.getOrElse(null)).asJava)
+    val entity = new ClientQuotaEntity(Map(ClientQuotaEntity.IP -> ip.orNull).asJava)
     val request = Map(entity -> Map(DynamicConfig.Ip.IpConnectionRateOverrideProp -> Some(updatedRate.toDouble)))
     TestUtils.alterClientQuotas(adminClient, request).all.get()
     // use a random throwaway address if ip isn't specified to get the default value
     TestUtils.waitUntilTrue(() => servers.head.socketServer.connectionQuotas.
-      connectionRateForIp(InetAddress.getByName(ip.getOrElse("255.255.3.4"))) == updatedRate,
+      connectionRateForIp(InetAddress.getByName(ip.getOrElse(unknownHost))) == updatedRate,
       s"Timed out waiting for connection rate update to propagate"
     )
     adminClient.close()
diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
index 418bcacc83cf6..094b0fab68547 100755
--- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala
+++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
@@ -1535,7 +1535,7 @@ object TestUtils extends Logging {
       val ops = alter.map { case (key, value) =>
         new ClientQuotaAlteration.Op(key, value.map(Double.box).getOrElse(null))
       }.asJavaCollection
-        new ClientQuotaAlteration(entity, ops)
+      new ClientQuotaAlteration(entity, ops)
     }.asJavaCollection
     adminClient.alterClientQuotas(entries)
   }

From 1d5d69aa7f18ca931feb7d50bb30881b7b7a6f57 Mon Sep 17 00:00:00 2001
From: David Mao <dmao@confluent.io>
Date: Wed, 2 Dec 2020 18:10:38 -0600
Subject: [PATCH 4/9] more verbose error message

---
 core/src/main/scala/kafka/server/AdminManager.scala | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/core/src/main/scala/kafka/server/AdminManager.scala b/core/src/main/scala/kafka/server/AdminManager.scala
index 94ba28416cf0c..33a2435534572 100644
--- a/core/src/main/scala/kafka/server/AdminManager.scala
+++ b/core/src/main/scala/kafka/server/AdminManager.scala
@@ -996,6 +996,8 @@ class AdminManager(val config: KafkaConfig,
         case (Some(user), None, None) => (user, ConfigType.User, DynamicConfig.User.configKeys)
         case (None, Some(clientId), None) => (clientId, ConfigType.Client, DynamicConfig.Client.configKeys)
         case (None, None, Some(ip)) => (ip, ConfigType.Ip, DynamicConfig.Ip.configKeys)
+        case (_, _, Some(_)) => throw new InvalidRequestException(s"Invalid quota entity combination, " +
+          s"IP entity should not be used with user/client ID entity.")
         case _ => throw new InvalidRequestException("Invalid client quota entity")
       }
 

From f9e14cc25e5f8003e108202d09c90cf20d8ab179 Mon Sep 17 00:00:00 2001
From: David Mao <dmao@confluent.io>
Date: Thu, 3 Dec 2020 15:25:39 -0600
Subject: [PATCH 5/9] IP => ip

---
 .../kafka/common/quota/ClientQuotaEntity.java |  2 +-
 .../scala/kafka/admin/ConfigCommand.scala     | 24 ++++++++--------
 .../kafka/server/DynamicConfigManager.scala   |  2 +-
 .../unit/kafka/admin/ConfigCommandTest.scala  | 28 +++++++++----------
 4 files changed, 28 insertions(+), 28 deletions(-)

diff --git a/clients/src/main/java/org/apache/kafka/common/quota/ClientQuotaEntity.java b/clients/src/main/java/org/apache/kafka/common/quota/ClientQuotaEntity.java
index 94fc858293751..dfad50917a926 100644
--- a/clients/src/main/java/org/apache/kafka/common/quota/ClientQuotaEntity.java
+++ b/clients/src/main/java/org/apache/kafka/common/quota/ClientQuotaEntity.java
@@ -32,7 +32,7 @@ public class ClientQuotaEntity {
      */
     public static final String USER = "user";
     public static final String CLIENT_ID = "client-id";
-    public static final String IP = "IP";
+    public static final String IP = "ip";
 
     /**
      * Constructs a quota entity for the given types and names. If a name is null,
diff --git a/core/src/main/scala/kafka/admin/ConfigCommand.scala b/core/src/main/scala/kafka/admin/ConfigCommand.scala
index 2073e6cbf245e..f9ff58e4a4f32 100644
--- a/core/src/main/scala/kafka/admin/ConfigCommand.scala
+++ b/core/src/main/scala/kafka/admin/ConfigCommand.scala
@@ -46,7 +46,7 @@ import scala.jdk.CollectionConverters._
 import scala.collection._
 
 /**
- * This script can be used to change configs for topics/clients/users/brokers/IPs dynamically
+ * This script can be used to change configs for topics/clients/users/brokers/ips dynamically
  * An entity described or altered by the command may be one of:
  * <ul>
  *     <li> topic: --topic <topic> OR --entity-type topics --entity-name <topic>
@@ -56,12 +56,12 @@ import scala.collection._
  *                          --entity-type users --entity-name <user-principal> --entity-type clients --entity-name <client-id>
  *     <li> broker: --broker <broker-id> OR --entity-type brokers --entity-name <broker-id>
  *     <li> broker-logger: --broker-logger <broker-id> OR --entity-type broker-loggers --entity-name <broker-id>
- *     <li> ip: --IP <ip> OR --entity-type IPs --entity-name <ip>
+ *     <li> ip: --ip <ip> OR --entity-type ips --entity-name <ip>
  * </ul>
- * --entity-type <users|clients|brokers|IPs> --entity-default may be specified in place of --entity-type <users|clients|brokers|IPs> --entity-name <entityName>
- * when describing or altering default configuration for users, clients, brokers, or IPs, respectively.
- * Alternatively, --user-defaults, --client-defaults, --broker-defaults, or --IP-defaults may be specified in place of
- * --entity-type <users|clients|brokers|IPs> --entity-default, respectively.
+ * --entity-type <users|clients|brokers|ips> --entity-default may be specified in place of --entity-type <users|clients|brokers|ips> --entity-name <entityName>
+ * when describing or altering default configuration for users, clients, brokers, or ips, respectively.
+ * Alternatively, --user-defaults, --client-defaults, --broker-defaults, or --ip-defaults may be specified in place of
+ * --entity-type <users|clients|brokers|ips> --entity-default, respectively.
  */
 object ConfigCommand extends Config {
 
@@ -85,7 +85,7 @@ object ConfigCommand extends Config {
     try {
       val opts = new ConfigCommandOptions(args)
 
-      CommandLineUtils.printHelpAndExitIfNeeded(opts, "This tool helps to manipulate and describe entity config for a topic, client, user, broker or IP")
+      CommandLineUtils.printHelpAndExitIfNeeded(opts, "This tool helps to manipulate and describe entity config for a topic, client, user, broker or ip")
 
       opts.checkArgs()
 
@@ -567,7 +567,7 @@ object ConfigCommand extends Config {
           val typeStr = entityType match {
             case ClientQuotaEntity.USER => "user-principal"
             case ClientQuotaEntity.CLIENT_ID => "client-id"
-            case ClientQuotaEntity.IP => "IP"
+            case ClientQuotaEntity.IP => "ip"
           }
           if (name != null) s"$typeStr '$name'"
           else s"the default $typeStr"
@@ -749,10 +749,10 @@ object ConfigCommand extends Config {
     val describeOpt = parser.accepts("describe", "List configs for the given entity.")
     val allOpt = parser.accepts("all", "List all configs for the given topic, broker, or broker-logger entity (includes static configuration when the entity type is brokers)")
 
-    val entityType = parser.accepts("entity-type", "Type of entity (topics/clients/users/brokers/broker-loggers/IPs)")
+    val entityType = parser.accepts("entity-type", "Type of entity (topics/clients/users/brokers/broker-loggers/ips)")
             .withRequiredArg
             .ofType(classOf[String])
-    val entityName = parser.accepts("entity-name", "Name of entity (topic name/client id/user principal name/broker id/IP)")
+    val entityName = parser.accepts("entity-name", "Name of entity (topic name/client id/user principal name/broker id/ip)")
             .withRequiredArg
             .ofType(classOf[String])
     val entityDefault = parser.accepts("entity-default", "Default entity name for clients/users/brokers (applies to corresponding entity type in command line)")
@@ -793,8 +793,8 @@ object ConfigCommand extends Config {
     val brokerLogger = parser.accepts("broker-logger", "The broker's ID for its logger config.")
       .withRequiredArg
       .ofType(classOf[String])
-    val ipDefaults = parser.accepts("IP-defaults", "The config defaults for all IPs.")
-    val ip = parser.accepts("IP", "The IP address.")
+    val ipDefaults = parser.accepts("ip-defaults", "The config defaults for all IPs.")
+    val ip = parser.accepts("ip", "The IP address.")
       .withRequiredArg
       .ofType(classOf[String])
     val zkTlsConfigFile = parser.accepts("zk-tls-config-file",
diff --git a/core/src/main/scala/kafka/server/DynamicConfigManager.scala b/core/src/main/scala/kafka/server/DynamicConfigManager.scala
index 67c8182bafb72..3eed3827782d7 100644
--- a/core/src/main/scala/kafka/server/DynamicConfigManager.scala
+++ b/core/src/main/scala/kafka/server/DynamicConfigManager.scala
@@ -38,7 +38,7 @@ object ConfigType {
   val Client = "clients"
   val User = "users"
   val Broker = "brokers"
-  val Ip = "IPs"
+  val Ip = "ips"
   val all = Seq(Topic, Client, User, Broker, Ip)
 }
 
diff --git a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
index eeb234ce49496..8fd943f316ada 100644
--- a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
+++ b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
@@ -135,12 +135,12 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
 
   @Test
   def shouldParseArgumentsForIpEntityType(): Unit = {
-    testArgumentParse("IPs", zkConfig = false)
+    testArgumentParse("ips", zkConfig = false)
   }
 
   @Test
   def shouldParseArgumentsForIpEntityTypeUsingZookeeper(): Unit = {
-    testArgumentParse("IPs", zkConfig = true)
+    testArgumentParse("ips", zkConfig = true)
   }
 
   def testArgumentParse(entityType: String, zkConfig: Boolean): Unit = {
@@ -306,7 +306,7 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
 
     testExpectedEntityTypeNames(List(ConfigType.Topic), List("A"), "--entity-type", "topics", "--entity-name", "A")
     testExpectedEntityTypeNames(List(ConfigType.Broker), List("0"), "--entity-name", "0", "--entity-type", "brokers")
-    testExpectedEntityTypeNames(List(ConfigType.Ip), List("1.2.3.4"), "--entity-name", "1.2.3.4", "--entity-type", "IPs")
+    testExpectedEntityTypeNames(List(ConfigType.Ip), List("1.2.3.4"), "--entity-name", "1.2.3.4", "--entity-type", "ips")
     testExpectedEntityTypeNames(List(ConfigType.User, ConfigType.Client), List("A", ""),
       "--entity-type", "users", "--entity-type", "clients", "--entity-name", "A", "--entity-default")
     testExpectedEntityTypeNames(List(ConfigType.User, ConfigType.Client), List("", "B"),
@@ -314,7 +314,7 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
 
     testExpectedEntityTypeNames(List(ConfigType.Topic), List("A"), "--topic", "A")
     testExpectedEntityTypeNames(List(ConfigType.Broker), List("0"), "--broker", "0")
-    testExpectedEntityTypeNames(List(ConfigType.Ip), List("1.2.3.4"), "--IP", "1.2.3.4")
+    testExpectedEntityTypeNames(List(ConfigType.Ip), List("1.2.3.4"), "--ip", "1.2.3.4")
     testExpectedEntityTypeNames(List(ConfigType.Client, ConfigType.User), List("B", "A"), "--client", "B", "--user", "A")
     testExpectedEntityTypeNames(List(ConfigType.Client, ConfigType.User), List("B", ""), "--client", "B", "--user-defaults")
     testExpectedEntityTypeNames(List(ConfigType.Client, ConfigType.User), List("A"),
@@ -323,7 +323,7 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
     testExpectedEntityTypeNames(List(ConfigType.Topic), List.empty, "--entity-type", "topics")
     testExpectedEntityTypeNames(List(ConfigType.User), List.empty, "--entity-type", "users")
     testExpectedEntityTypeNames(List(ConfigType.Broker), List.empty, "--entity-type", "brokers")
-    testExpectedEntityTypeNames(List(ConfigType.Ip), List.empty, "--entity-type", "IPs")
+    testExpectedEntityTypeNames(List(ConfigType.Ip), List.empty, "--entity-type", "ips")
   }
 
   @Test
@@ -395,7 +395,7 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
   @Test(expected = classOf[IllegalArgumentException])
   def shouldFailIfInvalidHost(): Unit = {
     val createOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092",
-      "--entity-name", "A,B", "--entity-type", "IPs", "--describe"))
+      "--entity-name", "A,B", "--entity-type", "ips", "--describe"))
     createOpts.checkArgs()
   }
 
@@ -422,7 +422,7 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
   def shouldAddIpConfigsUsingZookeeper(): Unit = {
     val createOpts = new ConfigCommandOptions(Array("--zookeeper", zkConnect,
       "--entity-name", "1.2.3.4",
-      "--entity-type", "IPs",
+      "--entity-type", "ips",
       "--alter",
       "--add-config", "a=b,c=d"))
 
@@ -515,7 +515,7 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
   @Test
   def shouldNotAlterNonQuotaIpConfigsUsingBootstrapServer(): Unit = {
     // when using --bootstrap-server, it should be illegal to alter anything that is not a connection quota
-    // for IP entities
+    // for ip entities
     val node = new Node(1, "localhost", 9092)
     val mockAdminClient = new MockAdminClient(util.Collections.singletonList(node), node)
 
@@ -529,10 +529,10 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
       assertTrue(s"Unexpected exception: $e", e.getMessage.contains("some_config"))
     }
 
-    verifyCommand("IPs", "--add-config", "connection_creation_rate=10000,some_config=10")
-    verifyCommand("IPs", "--add-config", "some_config=10")
-    verifyCommand("IPs", "--delete-config", "connection_creation_rate=10000,some_config=10")
-    verifyCommand("IPs", "--delete-config", "some_config=10")
+    verifyCommand("ips", "--add-config", "connection_creation_rate=10000,some_config=10")
+    verifyCommand("ips", "--add-config", "some_config=10")
+    verifyCommand("ips", "--delete-config", "connection_creation_rate=10000,some_config=10")
+    verifyCommand("ips", "--delete-config", "some_config=10")
   }
 
   @Test
@@ -546,7 +546,7 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
       }
 
       val describeOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092",
-        "--describe", "--entity-type", "IPs") ++ ipArgs)
+        "--describe", "--entity-type", "ips") ++ ipArgs)
 
       val expectedFilter = ClientQuotaFilter.containsOnly(List(filterComponent).asJava)
 
@@ -579,7 +579,7 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
   @Test
   def testAlterIpConfig(): Unit = {
     def testShouldAlterIpConfig(ip: Option[String], remove: Boolean): Unit = {
-      val (ipArgs, ipEntity, ipComponent) = toValues(ip, ClientQuotaEntity.IP, "IPs")
+      val (ipArgs, ipEntity, ipComponent) = toValues(ip, ClientQuotaEntity.IP, "ips")
       val alterOpts = if (remove)
         Array("--delete-config", "connection_creation_rate")
       else

From a59b1c850b0577d64aab5cd7286f29ce205d97fd Mon Sep 17 00:00:00 2001
From: David Mao <dmao@confluent.io>
Date: Mon, 7 Dec 2020 17:00:41 -0600
Subject: [PATCH 6/9] address feedback, refactor ConfigCommandTest

---
 .../scala/kafka/admin/ConfigCommand.scala     |   4 +-
 .../network/DynamicConnectionQuotaTest.scala  |  21 +-
 .../unit/kafka/admin/ConfigCommandTest.scala  | 440 +++++++-----------
 .../server/ClientQuotasRequestTest.scala      |   3 -
 4 files changed, 187 insertions(+), 281 deletions(-)

diff --git a/core/src/main/scala/kafka/admin/ConfigCommand.scala b/core/src/main/scala/kafka/admin/ConfigCommand.scala
index f9ff58e4a4f32..dc466cdaefb50 100644
--- a/core/src/main/scala/kafka/admin/ConfigCommand.scala
+++ b/core/src/main/scala/kafka/admin/ConfigCommand.scala
@@ -755,7 +755,7 @@ object ConfigCommand extends Config {
     val entityName = parser.accepts("entity-name", "Name of entity (topic name/client id/user principal name/broker id/ip)")
             .withRequiredArg
             .ofType(classOf[String])
-    val entityDefault = parser.accepts("entity-default", "Default entity name for clients/users/brokers (applies to corresponding entity type in command line)")
+    val entityDefault = parser.accepts("entity-default", "Default entity name for clients/users/brokers/ips (applies to corresponding entity type in command line)")
 
     val nl = System.getProperty("line.separator")
     val addConfig = parser.accepts("add-config", "Key Value pairs of configs to add. Square brackets can be used to group values which contain commas: 'k1=v1,k2=[v1,v2,v2],k3=v3'. The following is a list of valid configurations: " +
@@ -901,7 +901,7 @@ object ConfigCommand extends Config {
             entityTypeVals.contains(ConfigType.Broker) ||
             entityTypeVals.contains(ConfigType.Ip)) {
           if (!hasEntityName && !hasEntityDefault)
-            throw new IllegalArgumentException("an entity-name or default entity must be specified with --alter of users, clients or brokers")
+            throw new IllegalArgumentException("an entity-name or default entity must be specified with --alter of users, clients, brokers or ips")
         } else if (!hasEntityName)
           throw new IllegalArgumentException(s"an entity name must be specified with --alter of ${entityTypeVals.mkString(",")}")
 
diff --git a/core/src/test/scala/integration/kafka/network/DynamicConnectionQuotaTest.scala b/core/src/test/scala/integration/kafka/network/DynamicConnectionQuotaTest.scala
index 06dbaaf57b3d9..ac8f49ee9b7cf 100644
--- a/core/src/test/scala/integration/kafka/network/DynamicConnectionQuotaTest.scala
+++ b/core/src/test/scala/integration/kafka/network/DynamicConnectionQuotaTest.scala
@@ -266,15 +266,18 @@ class DynamicConnectionQuotaTest extends BaseRequestTest {
   private def updateIpConnectionRate(ip: Option[String], updatedRate: Int): Unit = {
     val initialConnectionCount = connectionCount
     val adminClient = createAdminClient()
-    val entity = new ClientQuotaEntity(Map(ClientQuotaEntity.IP -> ip.orNull).asJava)
-    val request = Map(entity -> Map(DynamicConfig.Ip.IpConnectionRateOverrideProp -> Some(updatedRate.toDouble)))
-    TestUtils.alterClientQuotas(adminClient, request).all.get()
-    // use a random throwaway address if ip isn't specified to get the default value
-    TestUtils.waitUntilTrue(() => servers.head.socketServer.connectionQuotas.
-      connectionRateForIp(InetAddress.getByName(ip.getOrElse(unknownHost))) == updatedRate,
-      s"Timed out waiting for connection rate update to propagate"
-    )
-    adminClient.close()
+    try {
+      val entity = new ClientQuotaEntity(Map(ClientQuotaEntity.IP -> ip.orNull).asJava)
+      val request = Map(entity -> Map(DynamicConfig.Ip.IpConnectionRateOverrideProp -> Some(updatedRate.toDouble)))
+      TestUtils.alterClientQuotas(adminClient, request).all.get()
+      // use a random throwaway address if ip isn't specified to get the default value
+      TestUtils.waitUntilTrue(() => servers.head.socketServer.connectionQuotas.
+        connectionRateForIp(InetAddress.getByName(ip.getOrElse(unknownHost))) == updatedRate,
+        s"Timed out waiting for connection rate update to propagate"
+      )
+    } finally {
+      adminClient.close()
+    }
     TestUtils.waitUntilTrue(() => initialConnectionCount == connectionCount,
       s"Admin client connection not closed (initial = $initialConnectionCount, current = $connectionCount)")
   }
diff --git a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
index 8fd943f316ada..d1e22d9ac9495 100644
--- a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
+++ b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
@@ -437,40 +437,98 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
     ConfigCommand.alterConfigWithZk(null, createOpts, new TestAdminZkClient(zkClient))
   }
 
-  private def toValues(entityName: Option[String], entityType: String, command: String):
-  (Array[String], Option[String], Option[ClientQuotaFilterComponent]) = {
+  private def toValues(entityName: Option[String], entityType: String): (Array[String], Map[String, String]) = {
+    val command = entityType match {
+      case ClientQuotaEntity.USER => "users"
+      case ClientQuotaEntity.CLIENT_ID => "clients"
+      case ClientQuotaEntity.IP => "ips"
+    }
     entityName match {
       case Some(null) =>
-        (Array("--entity-type", command, "--entity-default"), Some(null),
-          Some(ClientQuotaFilterComponent.ofDefaultEntity(entityType)))
+        (Array("--entity-type", command, "--entity-default"), Map(entityType -> null))
       case Some(name) =>
-        (Array("--entity-type", command, "--entity-name", name), Some(name),
-          Some(ClientQuotaFilterComponent.ofEntity(entityType, name)))
-      case None => (Array.empty, None, None)
+        (Array("--entity-type", command, "--entity-name", name), Map(entityType -> name))
+      case None => (Array.empty, Map.empty)
     }
   }
 
-  def testShouldAddClientConfig(user: Option[String], clientId: Option[String]): Unit = {
-    val (userArgs, userEntity, userComponent) = toValues(user, ClientQuotaEntity.USER, "users")
-    val (clientIdArgs, clientIdEntity, clientIdComponent) = toValues(clientId, ClientQuotaEntity.CLIENT_ID, "clients")
+  private def verifyAlterCommandFails(expectedErrorMessage: String, alterOpts: Seq[String]): Unit = {
+    val mockAdminClient: Admin = EasyMock.createStrictMock(classOf[Admin])
+    val opts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092",
+      "--alter") ++ alterOpts)
+    val e = intercept[IllegalArgumentException] {
+      ConfigCommand.alterConfig(mockAdminClient, opts)
+    }
+    assertTrue(s"Unexpected exception: $e", e.getMessage.contains(expectedErrorMessage))
+  }
 
-    val createOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092",
-      "--alter",
-      "--add-config", "consumer_byte_rate=20000,producer_byte_rate=10000",
-      "--delete-config", "request_percentage") ++ userArgs ++ clientIdArgs)
+  @Test
+  def shouldNotAlterNonQuotaIpConfigsUsingBootstrapServer(): Unit = {
+    // when using --bootstrap-server, it should be illegal to alter anything that is not a connection quota
+    // for ip entities
+    val ipEntityOpts = List("--entity-type", "ips", "--entity-name", "127.0.0.1")
+    val invalidProp = "some_config"
+    verifyAlterCommandFails(invalidProp, ipEntityOpts ++ List("--add-config", "connection_creation_rate=10000,some_config=10"))
+    verifyAlterCommandFails(invalidProp, ipEntityOpts ++ List("--add-config", "some_config=10"))
+    verifyAlterCommandFails(invalidProp, ipEntityOpts ++ List("--delete-config", "connection_creation_rate=10000,some_config=10"))
+    verifyAlterCommandFails(invalidProp, ipEntityOpts ++ List("--delete-config", "some_config=10"))
+  }
 
-    // Explicitly populate a HashMap to ensure nulls are recorded properly.
-    val entityMap = new java.util.HashMap[String, String]
-    userEntity.foreach(u => entityMap.put(ClientQuotaEntity.USER, u))
-    clientIdEntity.foreach(c => entityMap.put(ClientQuotaEntity.CLIENT_ID, c))
-    val entity = new ClientQuotaEntity(entityMap)
+  private def verifyDescribeQuotas(describeArgs: List[String], expectedFilter: ClientQuotaFilter): Unit = {
+    val describeOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092",
+      "--describe") ++ describeArgs)
+    val describeFuture = new KafkaFutureImpl[util.Map[ClientQuotaEntity, util.Map[String, java.lang.Double]]]
+    describeFuture.complete(Map.empty[ClientQuotaEntity, util.Map[String, java.lang.Double]].asJava)
+    val describeResult: DescribeClientQuotasResult = EasyMock.createNiceMock(classOf[DescribeClientQuotasResult])
+    EasyMock.expect(describeResult.entities()).andReturn(describeFuture)
+
+    var describedConfigs = false
+    val node = new Node(1, "localhost", 9092)
+    val mockAdminClient = new MockAdminClient(util.Collections.singletonList(node), node) {
+      override def describeClientQuotas(filter: ClientQuotaFilter, options: DescribeClientQuotasOptions): DescribeClientQuotasResult = {
+        assertTrue(filter.strict)
+        assertEquals(expectedFilter.components().asScala.toSet, filter.components.asScala.toSet)
+        describedConfigs = true
+        describeResult
+      }
+    }
+    EasyMock.replay(describeResult)
+    ConfigCommand.describeConfig(mockAdminClient, describeOpts)
+    assertTrue(describedConfigs)
+  }
+
+  @Test
+  def testDescribeIpConfigs(): Unit = {
+    val entityType = ClientQuotaEntity.IP
+    val knownHost = "1.2.3.4"
+    val defaultIpFilter = ClientQuotaFilter.containsOnly(List(ClientQuotaFilterComponent.ofDefaultEntity(entityType)).asJava)
+    val singleIpFilter = ClientQuotaFilter.containsOnly(List(ClientQuotaFilterComponent.ofEntity(entityType, knownHost)).asJava)
+    val allIpsFilter = ClientQuotaFilter.containsOnly(List(ClientQuotaFilterComponent.ofEntityType(entityType)).asJava)
+    verifyDescribeQuotas(List("--entity-default", "--entity-type", "ips"), defaultIpFilter)
+    verifyDescribeQuotas(List("--ip-defaults"), defaultIpFilter)
+    verifyDescribeQuotas(List("--entity-type", "ips", "--entity-name", knownHost), singleIpFilter)
+    verifyDescribeQuotas(List("--ip", knownHost), singleIpFilter)
+    verifyDescribeQuotas(List("--entity-type", "ips"), allIpsFilter)
+  }
+
+  def verifyAlterQuotas(alterOpts: Seq[String], expectedAlterEntity: ClientQuotaEntity,
+                        expectedProps: Map[String, java.lang.Double], expectedAlterOps: Set[ClientQuotaAlteration.Op]): Unit = {
+    val createOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092",
+      "--alter") ++ alterOpts)
 
     var describedConfigs = false
     val describeFuture = new KafkaFutureImpl[util.Map[ClientQuotaEntity, util.Map[String, java.lang.Double]]]
-    describeFuture.complete(Map((entity -> Map(("request_percentage" -> Double.box(50.0))).asJava)).asJava)
+    describeFuture.complete(Map(expectedAlterEntity -> expectedProps.asJava).asJava)
     val describeResult: DescribeClientQuotasResult = EasyMock.createNiceMock(classOf[DescribeClientQuotasResult])
     EasyMock.expect(describeResult.entities()).andReturn(describeFuture)
 
+    val expectedFilterComponents = expectedAlterEntity.entries.asScala.map { case (entityType, entityName) =>
+      if (entityName == null)
+        ClientQuotaFilterComponent.ofDefaultEntity(entityType)
+      else
+        ClientQuotaFilterComponent.ofEntity(entityType, entityName)
+    }.toSet
+
     var alteredConfigs = false
     val alterFuture = new KafkaFutureImpl[Void]
     alterFuture.complete(null)
@@ -481,9 +539,7 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
     val mockAdminClient = new MockAdminClient(util.Collections.singletonList(node), node) {
       override def describeClientQuotas(filter: ClientQuotaFilter, options: DescribeClientQuotasOptions): DescribeClientQuotasResult = {
         assertTrue(filter.strict)
-        val components = filter.components.asScala.toSeq
-        userComponent.foreach(c => assertTrue(components.contains(c)))
-        clientIdComponent.foreach(c => assertTrue(components.contains(c)))
+        assertEquals(expectedFilterComponents, filter.components().asScala.toSet)
         describedConfigs = true
         describeResult
       }
@@ -492,15 +548,9 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
         assertFalse(options.validateOnly)
         assertEquals(1, entries.size)
         val alteration = entries.asScala.head
-        assertEquals(entity, alteration.entity)
+        assertEquals(expectedAlterEntity, alteration.entity)
         val ops = alteration.ops.asScala
-        assertEquals(3, ops.size)
-        val expectedOps = Set(
-          new ClientQuotaAlteration.Op("consumer_byte_rate", Double.box(20000)),
-          new ClientQuotaAlteration.Op("producer_byte_rate", Double.box(10000)),
-          new ClientQuotaAlteration.Op("request_percentage", null)
-        )
-        assertEquals(expectedOps, ops.toSet)
+        assertEquals(expectedAlterOps, ops.toSet)
         alteredConfigs = true
         alterResult
       }
@@ -509,288 +559,144 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
     ConfigCommand.alterConfig(mockAdminClient, createOpts)
     assertTrue(describedConfigs)
     assertTrue(alteredConfigs)
-    EasyMock.reset(alterResult, describeResult)
   }
 
   @Test
-  def shouldNotAlterNonQuotaIpConfigsUsingBootstrapServer(): Unit = {
-    // when using --bootstrap-server, it should be illegal to alter anything that is not a connection quota
-    // for ip entities
-    val node = new Node(1, "localhost", 9092)
-    val mockAdminClient = new MockAdminClient(util.Collections.singletonList(node), node)
-
-    def verifyCommand(entityType: String, alterOpts: String*): Unit = {
-      val opts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092",
-        "--entity-type", entityType, "--entity-name", "admin",
-        "--alter") ++ alterOpts)
-      val e = intercept[IllegalArgumentException] {
-        ConfigCommand.alterConfig(mockAdminClient, opts)
-      }
-      assertTrue(s"Unexpected exception: $e", e.getMessage.contains("some_config"))
-    }
-
-    verifyCommand("ips", "--add-config", "connection_creation_rate=10000,some_config=10")
-    verifyCommand("ips", "--add-config", "some_config=10")
-    verifyCommand("ips", "--delete-config", "connection_creation_rate=10000,some_config=10")
-    verifyCommand("ips", "--delete-config", "some_config=10")
-  }
-
-  @Test
-  def testDescribeIpConfigs(): Unit = {
-    def testShouldDescribeIpConfig(ip: Option[String]): Unit = {
-      val entityType = ClientQuotaEntity.IP
-      val (ipArgs, filterComponent) = ip match {
-        case Some(null) => (Array("--entity-default"), ClientQuotaFilterComponent.ofDefaultEntity(entityType))
-        case Some(addr) => (Array("--entity-name", addr), ClientQuotaFilterComponent.ofEntity(entityType, addr))
-        case None => (Array.empty[String], ClientQuotaFilterComponent.ofEntityType(entityType))
-      }
-
-      val describeOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092",
-        "--describe", "--entity-type", "ips") ++ ipArgs)
+  def testAlterIpConfig(): Unit = {
+    val (singleIpArgs, singleIpEntry) = toValues(Some("1.2.3.4"), ClientQuotaEntity.IP)
+    val singleIpEntity = new ClientQuotaEntity(singleIpEntry.asJava)
+    val (defaultIpArgs, defaultIpEntry) = toValues(Some(null), ClientQuotaEntity.IP)
+    val defaultIpEntity = new ClientQuotaEntity(defaultIpEntry.asJava)
 
-      val expectedFilter = ClientQuotaFilter.containsOnly(List(filterComponent).asJava)
+    val deleteArgs = List("--delete-config", "connection_creation_rate")
+    val deleteAlterationOps = Set(new ClientQuotaAlteration.Op("connection_creation_rate", null))
+    val propsToDelete = Map("connection_creation_rate" -> Double.box(50.0))
 
-      var describedConfigs = false
-      val describeFuture = new KafkaFutureImpl[util.Map[ClientQuotaEntity, util.Map[String, java.lang.Double]]]
-      describeFuture.complete(Map.empty[ClientQuotaEntity, util.Map[String, java.lang.Double]].asJava)
-      val describeResult: DescribeClientQuotasResult = EasyMock.createNiceMock(classOf[DescribeClientQuotasResult])
-      EasyMock.expect(describeResult.entities()).andReturn(describeFuture)
+    val addArgs = List("--add-config", "connection_creation_rate=100")
+    val addAlterationOps = Set(new ClientQuotaAlteration.Op("connection_creation_rate", 100.0))
 
-      val node = new Node(1, "localhost", 9092)
-      val mockAdminClient = new MockAdminClient(util.Collections.singletonList(node), node) {
-        override def describeClientQuotas(filter: ClientQuotaFilter, options: DescribeClientQuotasOptions): DescribeClientQuotasResult = {
-          assertTrue(filter.strict)
-          assertEquals(expectedFilter, filter)
-          describedConfigs = true
-          describeResult
-        }
-      }
-      EasyMock.replay(describeResult)
-      ConfigCommand.describeConfig(mockAdminClient, describeOpts)
-      assertTrue(describedConfigs)
-      EasyMock.reset(describeResult)
-    }
-
-    testShouldDescribeIpConfig(Some("1.2.3.4"))
-    testShouldDescribeIpConfig(Some(null))
-    testShouldDescribeIpConfig(None)
+    verifyAlterQuotas(singleIpArgs ++ deleteArgs, singleIpEntity, propsToDelete, deleteAlterationOps)
+    verifyAlterQuotas(singleIpArgs ++ addArgs, singleIpEntity, Map.empty, addAlterationOps)
+    verifyAlterQuotas(defaultIpArgs ++ deleteArgs, defaultIpEntity, propsToDelete, deleteAlterationOps)
+    verifyAlterQuotas(defaultIpArgs ++ addArgs, defaultIpEntity, Map.empty, addAlterationOps)
   }
 
   @Test
-  def testAlterIpConfig(): Unit = {
-    def testShouldAlterIpConfig(ip: Option[String], remove: Boolean): Unit = {
-      val (ipArgs, ipEntity, ipComponent) = toValues(ip, ClientQuotaEntity.IP, "ips")
-      val alterOpts = if (remove)
-        Array("--delete-config", "connection_creation_rate")
-      else
-        Array("--add-config", "connection_creation_rate=100")
-      val createOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092",
-        "--alter") ++ ipArgs ++ alterOpts)
-
-      // Explicitly populate a HashMap to ensure nulls are recorded properly.
-      val entityMap = new java.util.HashMap[String, String]
-      ipEntity.foreach(u => entityMap.put(ClientQuotaEntity.IP, u))
-      val entity = new ClientQuotaEntity(entityMap)
-
-      var describedConfigs = false
-      val describeFuture = new KafkaFutureImpl[util.Map[ClientQuotaEntity, util.Map[String, java.lang.Double]]]
-      describeFuture.complete(Map(entity -> Map("connection_creation_rate" -> Double.box(50.0)).asJava).asJava)
-      val describeResult: DescribeClientQuotasResult = EasyMock.createNiceMock(classOf[DescribeClientQuotasResult])
-      EasyMock.expect(describeResult.entities()).andReturn(describeFuture)
-
-      var alteredConfigs = false
-      val alterFuture = new KafkaFutureImpl[Void]
-      alterFuture.complete(null)
-      val alterResult: AlterClientQuotasResult = EasyMock.createNiceMock(classOf[AlterClientQuotasResult])
-      EasyMock.expect(alterResult.all()).andReturn(alterFuture)
+  def shouldAddClientConfig(): Unit = {
+    val alterArgs = List("--add-config", "consumer_byte_rate=20000,producer_byte_rate=10000",
+      "--delete-config", "request_percentage")
+    val propsToDelete = Map("request_percentage" -> Double.box(50.0))
+
+    val alterationOps = Set(
+      new ClientQuotaAlteration.Op("consumer_byte_rate", Double.box(20000)),
+      new ClientQuotaAlteration.Op("producer_byte_rate", Double.box(10000)),
+      new ClientQuotaAlteration.Op("request_percentage", null)
+    )
 
-      val node = new Node(1, "localhost", 9092)
-      val mockAdminClient = new MockAdminClient(util.Collections.singletonList(node), node) {
-        override def describeClientQuotas(filter: ClientQuotaFilter, options: DescribeClientQuotasOptions): DescribeClientQuotasResult = {
-          assertTrue(filter.strict)
-          assertEquals(ClientQuotaFilter.containsOnly(List(ipComponent.get).asJava), filter)
-          describedConfigs = true
-          describeResult
-        }
+    def verifyAlterUserClientQuotas(userOpt: Option[String], clientOpt: Option[String]): Unit = {
+      val (userArgs, userEntry) = toValues(userOpt, ClientQuotaEntity.USER)
+      val (clientArgs, clientEntry) = toValues(clientOpt, ClientQuotaEntity.CLIENT_ID)
 
-        override def alterClientQuotas(entries: util.Collection[ClientQuotaAlteration], options: AlterClientQuotasOptions): AlterClientQuotasResult = {
-          assertFalse(options.validateOnly)
-          assertEquals(1, entries.size)
-          val alteration = entries.asScala.head
-          assertEquals(entity, alteration.entity)
-          val ops = alteration.ops.asScala
-          val expectedOps = if (remove)
-            Set(new ClientQuotaAlteration.Op("connection_creation_rate", null))
-          else
-            Set(new ClientQuotaAlteration.Op("connection_creation_rate", Double.box(100)))
-          assertEquals(expectedOps, ops.toSet)
-          alteredConfigs = true
-          alterResult
-        }
-      }
-      EasyMock.replay(alterResult, describeResult)
-      ConfigCommand.alterConfig(mockAdminClient, createOpts)
-      assertTrue(describedConfigs)
-      assertTrue(alteredConfigs)
-      EasyMock.reset(alterResult, describeResult)
+      val commandArgs = alterArgs ++ userArgs ++ clientArgs
+      val clientQuotaEntity = new ClientQuotaEntity((userEntry ++ clientEntry).asJava)
+      verifyAlterQuotas(commandArgs, clientQuotaEntity, propsToDelete, alterationOps)
     }
-
-    testShouldAlterIpConfig(Some("1.2.3.4"), remove = false)
-    testShouldAlterIpConfig(Some("1.2.3.4"), remove = true)
-    testShouldAlterIpConfig(Some(null), remove = false)
-    testShouldAlterIpConfig(Some(null), remove = true)
+    verifyAlterUserClientQuotas(Some("test-user-1"), Some("test-client-1"))
+    verifyAlterUserClientQuotas(Some("test-user-2"), Some(null))
+    verifyAlterUserClientQuotas(Some("test-user-3"), None)
+    verifyAlterUserClientQuotas(Some(null), Some("test-client-2"))
+    verifyAlterUserClientQuotas(Some(null), Some(null))
+    verifyAlterUserClientQuotas(Some(null), None)
+    verifyAlterUserClientQuotas(None, Some("test-client-3"))
+    verifyAlterUserClientQuotas(None, Some(null))
   }
 
-  @Test
-  def shouldAddClientConfig(): Unit = {
-    testShouldAddClientConfig(Some("test-user-1"), Some("test-client-1"))
-    testShouldAddClientConfig(Some("test-user-2"), Some(null))
-    testShouldAddClientConfig(Some("test-user-3"), None)
-    testShouldAddClientConfig(Some(null), Some("test-client-2"))
-    testShouldAddClientConfig(Some(null), Some(null))
-    testShouldAddClientConfig(Some(null), None)
-    testShouldAddClientConfig(None, Some("test-client-3"))
-    testShouldAddClientConfig(None, Some(null))
-  }
+  private val userEntityOpts = List("--entity-type", "users", "--entity-name", "admin")
+  private val clientEntityOpts = List("--entity-type", "clients", "--entity-name", "admin")
+  private val addScramOpts = List("--add-config", "SCRAM-SHA-256=[iterations=8192,password=foo-secret]")
+  private val deleteScramOpts = List("--delete-config", "SCRAM-SHA-256")
 
   @Test
   def shouldNotAlterNonQuotaNonScramUserOrClientConfigUsingBootstrapServer(): Unit = {
     // when using --bootstrap-server, it should be illegal to alter anything that is not a quota and not a SCRAM credential
     // for both user and client entities
-    val node = new Node(1, "localhost", 9092)
-    val mockAdminClient = new MockAdminClient(util.Collections.singletonList(node), node)
-
-    def verifyCommand(entityType: String, alterOpts: String*): Unit = {
-      val opts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092",
-        "--entity-type", entityType, "--entity-name", "admin",
-        "--alter") ++ alterOpts)
-      val e = intercept[IllegalArgumentException] {
-        ConfigCommand.alterConfig(mockAdminClient, opts)
-      }
-      assertTrue(s"Unexpected exception: $e", e.getMessage.contains("some_config"))
-    }
-
-    verifyCommand("users", "--add-config", "consumer_byte_rate=20000,producer_byte_rate=10000,some_config=10")
-    verifyCommand("users", "--add-config", "SCRAM-SHA-256=[iterations=8192,password=foo-secret],some_config=10")
-    verifyCommand("clients", "--add-config", "some_config=10")
-    verifyCommand("users", "--delete-config", "consumer_byte_rate,some_config")
-    verifyCommand("users", "--delete-config", "SCRAM-SHA-256,some_config")
-    verifyCommand("clients", "--delete-config", "some_config")
+    val invalidProp = "some_config"
+    verifyAlterCommandFails(invalidProp, userEntityOpts ++
+      List("-add-config", "consumer_byte_rate=20000,producer_byte_rate=10000,some_config=10"))
+    verifyAlterCommandFails(invalidProp, userEntityOpts ++
+      List("--add-config", "consumer_byte_rate=20000,producer_byte_rate=10000,some_config=10"))
+    verifyAlterCommandFails(invalidProp, clientEntityOpts ++ List("--add-config", "some_config=10"))
+    verifyAlterCommandFails(invalidProp, userEntityOpts ++ List("--delete-config", "consumer_byte_rate,some_config"))
+    verifyAlterCommandFails(invalidProp, userEntityOpts ++ List("--delete-config", "SCRAM-SHA-256,some_config"))
+    verifyAlterCommandFails(invalidProp, clientEntityOpts ++ List("--delete-config", "some_config"))
   }
 
   @Test
   def shouldNotAlterScramClientConfigUsingBootstrapServer(): Unit = {
     // when using --bootstrap-server, it should be illegal to alter SCRAM credentials for client entities
-    val node = new Node(1, "localhost", 9092)
-    val mockAdminClient = new MockAdminClient(util.Collections.singletonList(node), node)
-
-    def verifyCommand(entityType: String, alterOpts: String*): Unit = {
-      val opts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092",
-        "--entity-type", entityType, "--entity-name", "admin",
-        "--alter") ++ alterOpts)
-      val e = intercept[IllegalArgumentException] {
-        ConfigCommand.alterConfig(mockAdminClient, opts)
-      }
-      assertTrue(s"Unexpected exception: $e", e.getMessage.contains("SCRAM-SHA-256"))
-    }
-
-    verifyCommand("clients", "--add-config", "SCRAM-SHA-256=[iterations=8192,password=foo-secret]")
-    verifyCommand("clients", "--delete-config", "SCRAM-SHA-256")
+    verifyAlterCommandFails("SCRAM-SHA-256", clientEntityOpts ++ addScramOpts)
+    verifyAlterCommandFails("SCRAM-SHA-256", clientEntityOpts ++ deleteScramOpts)
   }
 
   @Test
   def shouldNotCreateUserScramCredentialConfigWithUnderMinimumIterationsUsingBootstrapServer(): Unit = {
     // when using --bootstrap-server, it should be illegal to create a SCRAM credential for a user
     // with an iterations value less than the minimum
-    val node = new Node(1, "localhost", 9092)
-    val mockAdminClient = new MockAdminClient(util.Collections.singletonList(node), node)
-
-    def verifyCommand(entityType: String, alterOpts: String*): Unit = {
-      val opts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092",
-        "--entity-type", entityType, "--entity-name", "admin",
-        "--alter") ++ alterOpts)
-      val e = intercept[IllegalArgumentException] {
-        ConfigCommand.alterConfig(mockAdminClient, opts)
-      }
-      assertTrue(s"Unexpected exception: $e", e.getMessage.contains("SCRAM-SHA-256"))
-    }
-
-    verifyCommand("users", "--add-config", "SCRAM-SHA-256=[iterations=100,password=foo-secret]")
+    verifyAlterCommandFails("SCRAM-SHA-256", userEntityOpts ++ List("--add-config", "SCRAM-SHA-256=[iterations=100,password=foo-secret]"))
   }
 
   @Test
   def shouldNotAlterUserScramCredentialAndClientQuotaConfigsSimultaneouslyUsingBootstrapServer(): Unit = {
     // when using --bootstrap-server, it should be illegal to alter both SCRAM credentials and quotas for user entities
-    val node = new Node(1, "localhost", 9092)
-    val mockAdminClient = new MockAdminClient(util.Collections.singletonList(node), node)
+    val expectedErrorMessage = "SCRAM-SHA-256"
+    val secondUserEntityOpts = List("--entity-type", "users", "--entity-name", "admin1")
+    val addQuotaOpts = List("--add-config", "consumer_byte_rate=20000")
+    val deleteQuotaOpts = List("--delete-config", "consumer_byte_rate")
 
-    def verifyCommand(alterOpts: String*): Unit = {
-      val opts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092", "--alter") ++ alterOpts)
-      val e = intercept[IllegalArgumentException] {
-        ConfigCommand.alterConfig(mockAdminClient, opts)
-      }
-      assertTrue(s"Unexpected exception: $e", e.getMessage.contains("SCRAM-SHA-256"))
-    }
+    verifyAlterCommandFails(expectedErrorMessage, userEntityOpts ++ addScramOpts ++ userEntityOpts ++ deleteQuotaOpts)
+    verifyAlterCommandFails(expectedErrorMessage, userEntityOpts ++ addScramOpts ++ secondUserEntityOpts ++ deleteQuotaOpts)
+    verifyAlterCommandFails(expectedErrorMessage, userEntityOpts ++ deleteScramOpts ++ userEntityOpts ++ addQuotaOpts)
+    verifyAlterCommandFails(expectedErrorMessage, userEntityOpts ++ deleteScramOpts ++ secondUserEntityOpts ++ addQuotaOpts)
 
-    verifyCommand("--entity-type", "users", "--entity-name", "admin", "--add-config", "SCRAM-SHA-256=[iterations=8192,password=foo-secret]",
-      "--entity-type", "users", "--entity-name", "admin", "--delete-config", "consumer_byte_rate")
-    verifyCommand("--entity-type", "users", "--entity-name", "admin", "--add-config", "SCRAM-SHA-256=[iterations=8192,password=foo-secret]",
-      "--entity-type", "users", "--entity-name", "admin1", "--delete-config", "consumer_byte_rate")
-    verifyCommand("--entity-type", "users", "--entity-name", "admin", "--delete-config", "SCRAM-SHA-256",
-      "--entity-type", "users", "--entity-name", "admin", "--add-config", "consumer_byte_rate=20000")
-    verifyCommand("--entity-type", "users", "--entity-name", "admin", "--delete-config", "SCRAM-SHA-256",
-      "--entity-type", "users", "--entity-name", "admin1", "--add-config", "consumer_byte_rate=20000")
-
-    verifyCommand("--entity-type", "clients", "--entity-name", "admin", "--delete-config", "consumer_byte_rate",
-      "--entity-type", "users", "--entity-name", "admin", "--add-config", "SCRAM-SHA-256=[iterations=8192,password=foo-secret]")
-    verifyCommand( "--entity-type", "clients", "--entity-name", "admin1", "--delete-config", "consumer_byte_rate",
-      "--entity-type", "users", "--entity-name", "admin", "--add-config", "SCRAM-SHA-256=[iterations=8192,password=foo-secret]")
-    verifyCommand("--entity-type", "clients", "--entity-name", "admin", "--add-config", "consumer_byte_rate=20000",
-      "--entity-type", "users", "--entity-name", "admin", "--delete-config", "SCRAM-SHA-256")
-    verifyCommand("--entity-type", "users", "--entity-name", "admin1", "--add-config", "consumer_byte_rate=20000",
-      "--entity-type", "users", "--entity-name", "admin", "--delete-config", "SCRAM-SHA-256")
+    // change order of quota/SCRAM commands, verify alter still fails
+    verifyAlterCommandFails(expectedErrorMessage, userEntityOpts ++ deleteQuotaOpts ++ userEntityOpts ++ addScramOpts)
+    verifyAlterCommandFails(expectedErrorMessage, secondUserEntityOpts ++ deleteQuotaOpts ++ userEntityOpts ++ addScramOpts)
+    verifyAlterCommandFails(expectedErrorMessage, userEntityOpts ++ addQuotaOpts ++ userEntityOpts ++ deleteScramOpts)
+    verifyAlterCommandFails(expectedErrorMessage, secondUserEntityOpts ++ addQuotaOpts ++ userEntityOpts ++ deleteScramOpts)
   }
 
   @Test
   def shouldNotDescribeUserScramCredentialsWithEntityDefaultUsingBootstrapServer(): Unit = {
-    // User SCRAM credentials should not be described when specifying
-    // --describe --entity-type users --entity-default (or --user-defaults) with --bootstrap-server
-    val describeFuture = new KafkaFutureImpl[util.Map[ClientQuotaEntity, util.Map[String, java.lang.Double]]]
-    describeFuture.complete(Map((new ClientQuotaEntity(Map("" -> "").asJava) -> Map(("request_percentage" -> Double.box(50.0))).asJava)).asJava)
-    val describeClientQuotasResult: DescribeClientQuotasResult = EasyMock.createNiceMock(classOf[DescribeClientQuotasResult])
-    EasyMock.expect(describeClientQuotasResult.entities()).andReturn(describeFuture).times(2)
-    EasyMock.replay(describeClientQuotasResult)
-
-    val node = new Node(1, "localhost", 9092)
-    val mockAdminClient = new MockAdminClient(util.Collections.singletonList(node), node) {
-      override def describeClientQuotas(filter: ClientQuotaFilter, options: DescribeClientQuotasOptions):  DescribeClientQuotasResult = {
-        describeClientQuotasResult
-      }
-      override def describeUserScramCredentials(users: util.List[String], options: DescribeUserScramCredentialsOptions): DescribeUserScramCredentialsResult = {
-        throw new IllegalStateException("Incorrectly described SCRAM credentials when specifying --entity-default with --bootstrap-server")
-      }
-    }
-
-    def verifyCommand(expectedMessage: String, alterOrDescribeOpt: String, requestOpts: String*): Unit = {
-      val opts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092",
-        alterOrDescribeOpt) ++ requestOpts)
-      if (alterOrDescribeOpt.equals("--describe"))
-        ConfigCommand.describeConfig(mockAdminClient, opts) // fails if describeUserScramCredentials() is invoked
-      else {
-        val e = intercept[IllegalArgumentException] {
-          ConfigCommand.alterConfig(mockAdminClient, opts)
+    def verifyUserScramCredentialsNotDescribed(requestOpts: List[String]): Unit = {
+      // User SCRAM credentials should not be described when specifying
+      // --describe --entity-type users --entity-default (or --user-defaults) with --bootstrap-server
+      val describeFuture = new KafkaFutureImpl[util.Map[ClientQuotaEntity, util.Map[String, java.lang.Double]]]
+      describeFuture.complete(Map((new ClientQuotaEntity(Map("" -> "").asJava) -> Map(("request_percentage" -> Double.box(50.0))).asJava)).asJava)
+      val describeClientQuotasResult: DescribeClientQuotasResult = EasyMock.createNiceMock(classOf[DescribeClientQuotasResult])
+      EasyMock.expect(describeClientQuotasResult.entities()).andReturn(describeFuture)
+      EasyMock.replay(describeClientQuotasResult)
+      val node = new Node(1, "localhost", 9092)
+      val mockAdminClient = new MockAdminClient(util.Collections.singletonList(node), node) {
+        override def describeClientQuotas(filter: ClientQuotaFilter, options: DescribeClientQuotasOptions):  DescribeClientQuotasResult = {
+          describeClientQuotasResult
+        }
+        override def describeUserScramCredentials(users: util.List[String], options: DescribeUserScramCredentialsOptions): DescribeUserScramCredentialsResult = {
+          throw new IllegalStateException("Incorrectly described SCRAM credentials when specifying --entity-default with --bootstrap-server")
         }
-        assertTrue(s"Unexpected exception: $e", e.getMessage.contains(expectedMessage))
       }
+      val opts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092", "--describe") ++ requestOpts)
+      ConfigCommand.describeConfig(mockAdminClient, opts) // fails if describeUserScramCredentials() is invoked
     }
 
     val expectedMsg = "The use of --entity-default or --user-defaults is not allowed with User SCRAM Credentials using --bootstrap-server."
-    verifyCommand(expectedMsg, "--alter", "--entity-type", "users", "--entity-default", "--add-config", "SCRAM-SHA-256=[iterations=8192,password=foo-secret]")
-    verifyCommand(expectedMsg, "--alter", "--entity-type", "users", "--entity-default", "--delete-config", "SCRAM-SHA-256")
-    verifyCommand(expectedMsg, "--describe", "--entity-type", "users", "--entity-default")
-    verifyCommand(expectedMsg, "--alter", "--user-defaults", "--add-config", "SCRAM-SHA-256=[iterations=8192,password=foo-secret]")
-    verifyCommand(expectedMsg, "--alter", "--user-defaults", "--delete-config", "SCRAM-SHA-256")
-    verifyCommand(expectedMsg, "--describe", "--user-defaults")
+    val defaultUserOpt = List("--user-defaults")
+    val verboseDefaultUserOpts = List("--entity-type", "users", "--entity-default")
+    verifyAlterCommandFails(expectedMsg, verboseDefaultUserOpts ++ addScramOpts)
+    verifyAlterCommandFails(expectedMsg, verboseDefaultUserOpts ++ deleteScramOpts)
+    verifyUserScramCredentialsNotDescribed(verboseDefaultUserOpts)
+    verifyAlterCommandFails(expectedMsg, defaultUserOpt ++ addScramOpts)
+    verifyAlterCommandFails(expectedMsg, defaultUserOpt ++ deleteScramOpts)
+    verifyUserScramCredentialsNotDescribed(defaultUserOpt)
   }
 
   @Test
diff --git a/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala b/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala
index 2b9a75e9df3a4..6bcbb3d6e9feb 100644
--- a/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala
@@ -326,9 +326,6 @@ class ClientQuotasRequestTest extends BaseRequestTest {
     }.toMap
     val result = alterClientQuotas(userClientQuotas ++ ipQuotas, validateOnly = false)
     (matchUserClientEntities ++ matchIpEntities).foreach(e => result(e._1).get(10, TimeUnit.SECONDS))
-
-    // Allow time for watch callbacks to be triggered.
-    Thread.sleep(500)
   }
 
   @Test

From 6bee345cb763c303215fcd5fe073ba7ca65f95fa Mon Sep 17 00:00:00 2001
From: David Mao <dmao@confluent.io>
Date: Mon, 7 Dec 2020 18:19:08 -0600
Subject: [PATCH 7/9] More thorough verification of IP entity input

---
 .../scala/kafka/admin/ConfigCommand.scala     |  5 +---
 .../scala/kafka/server/DynamicConfig.scala    | 15 +++++++++++
 .../main/scala/kafka/zk/AdminZkClient.scala   | 10 +++-----
 .../unit/kafka/admin/ConfigCommandTest.scala  | 25 +++++++++++++++++--
 .../server/ClientQuotasRequestTest.scala      |  8 ++++++
 .../unit/kafka/server/DynamicConfigTest.scala |  9 +++++--
 6 files changed, 58 insertions(+), 14 deletions(-)

diff --git a/core/src/main/scala/kafka/admin/ConfigCommand.scala b/core/src/main/scala/kafka/admin/ConfigCommand.scala
index dc466cdaefb50..773ade981e1bc 100644
--- a/core/src/main/scala/kafka/admin/ConfigCommand.scala
+++ b/core/src/main/scala/kafka/admin/ConfigCommand.scala
@@ -886,10 +886,7 @@ object ConfigCommand extends Config {
       }
 
       if (hasEntityName && entityTypeVals.contains(ConfigType.Ip)) {
-        Seq(entityName, ip).filter(options.has(_)).map(options.valueOf(_)).foreach { ipAddress =>
-          if (!Utils.validHostPattern(ipAddress))
-            throw new IllegalArgumentException(s"The entity name for ${entityTypeVals.head} must be a valid IP, but it is: $ipAddress")
-        }
+        Seq(entityName, ip).filter(options.has(_)).map(options.valueOf(_)).foreach(DynamicConfig.Ip.validateIpOrHost)
       }
 
       if (options.has(describeOpt) && entityTypeVals.contains(BrokerLoggerConfigType) && !hasEntityName)
diff --git a/core/src/main/scala/kafka/server/DynamicConfig.scala b/core/src/main/scala/kafka/server/DynamicConfig.scala
index e963e1a639171..ac494310b7565 100644
--- a/core/src/main/scala/kafka/server/DynamicConfig.scala
+++ b/core/src/main/scala/kafka/server/DynamicConfig.scala
@@ -17,6 +17,7 @@
 
 package kafka.server
 
+import java.net.{InetAddress, UnknownHostException}
 import java.util.Properties
 
 import kafka.log.LogConfig
@@ -25,6 +26,8 @@ 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.common.errors.InvalidRequestException
+import org.apache.kafka.common.utils.Utils
 
 import scala.jdk.CollectionConverters._
 
@@ -141,6 +144,18 @@ object DynamicConfig {
     def names = ipConfigs.names
 
     def validate(props: Properties) = DynamicConfig.validate(ipConfigs, props, customPropsAllowed = false)
+
+    def validateIpOrHost(ip: String): Unit = {
+      if (ip != ConfigEntityName.Default) {
+        if (!Utils.validHostPattern(ip))
+          throw new InvalidRequestException(s"$ip is not a valid hostname")
+        try {
+          InetAddress.getByName(ip)
+        } catch {
+          case _ :UnknownHostException => throw new InvalidRequestException(s"$ip is not a valid IP or resolvable hostname")
+        }
+      }
+    }
   }
 
   private def validate(configDef: ConfigDef, props: Properties, customPropsAllowed: Boolean) = {
diff --git a/core/src/main/scala/kafka/zk/AdminZkClient.scala b/core/src/main/scala/kafka/zk/AdminZkClient.scala
index e558aa0eb0859..b6974f362961c 100644
--- a/core/src/main/scala/kafka/zk/AdminZkClient.scala
+++ b/core/src/main/scala/kafka/zk/AdminZkClient.scala
@@ -28,7 +28,6 @@ import kafka.utils.Implicits._
 import org.apache.kafka.common.{TopicPartition, Uuid}
 import org.apache.kafka.common.errors._
 import org.apache.kafka.common.internals.Topic
-import org.apache.kafka.common.utils.Utils
 import org.apache.zookeeper.KeeperException.NodeExistsException
 
 import scala.collection.{Map, Seq}
@@ -351,7 +350,7 @@ class AdminZkClient(zkClient: KafkaZkClient) extends Logging {
       case ConfigType.User => changeUserOrUserClientIdConfig(entityName, configs)
       case ConfigType.Broker => changeBrokerConfig(parseBroker(entityName), configs)
       case ConfigType.Ip => changeIpConfig(entityName, configs)
-      case _ => throw new IllegalArgumentException(s"$entityType is not a known entityType. Should be one of ${ConfigType.Topic}, ${ConfigType.Client}, ${ConfigType.Broker}")
+      case _ => throw new IllegalArgumentException(s"$entityType is not a known entityType. Should be one of ${ConfigType.all}")
     }
   }
 
@@ -394,8 +393,7 @@ class AdminZkClient(zkClient: KafkaZkClient) extends Logging {
    * @param configs properties to validate for the IP
    */
   def validateIpConfig(ip: String, configs: Properties): Unit = {
-    if (ip != ConfigEntityName.Default && !Utils.validHostPattern(ip))
-      throw new AdminOperationException(s"IP $ip is not a valid address.")
+    DynamicConfig.Ip.validateIpOrHost(ip)
     DynamicConfig.Ip.validate(configs)
   }
 
@@ -480,8 +478,8 @@ class AdminZkClient(zkClient: KafkaZkClient) extends Logging {
   }
 
   /**
-   * Read the entity (topic, broker, client, user or <user, client>) config (if any) from zk
-   * sanitizedEntityName is <topic>, <broker>, <client-id>, <user> or <user>/clients/<client-id>.
+   * Read the entity (topic, broker, client, user, <user, client> or <ip>) config (if any) from zk
+   * sanitizedEntityName is <topic>, <broker>, <client-id>, <user>, <user>/clients/<client-id> or <ip>.
    * @param rootEntityType
    * @param sanitizedEntityName
    * @return
diff --git a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
index d1e22d9ac9495..ef831902c5f10 100644
--- a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
+++ b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
@@ -28,7 +28,7 @@ import kafka.zk.{AdminZkClient, BrokerInfo, KafkaZkClient, ZooKeeperTestHarness}
 import org.apache.kafka.clients.admin._
 import org.apache.kafka.common.Node
 import org.apache.kafka.common.config.{ConfigException, ConfigResource}
-import org.apache.kafka.common.errors.InvalidConfigurationException
+import org.apache.kafka.common.errors.{InvalidConfigurationException, InvalidRequestException}
 import org.apache.kafka.common.internals.KafkaFutureImpl
 import org.apache.kafka.common.network.ListenerName
 import org.apache.kafka.common.quota.{ClientQuotaAlteration, ClientQuotaEntity, ClientQuotaFilter, ClientQuotaFilterComponent}
@@ -392,13 +392,34 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
     createOpts.checkArgs()
   }
 
-  @Test(expected = classOf[IllegalArgumentException])
+  @Test(expected = classOf[InvalidRequestException])
   def shouldFailIfInvalidHost(): Unit = {
     val createOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092",
       "--entity-name", "A,B", "--entity-type", "ips", "--describe"))
     createOpts.checkArgs()
   }
 
+  @Test(expected = classOf[InvalidRequestException])
+  def shouldFailIfInvalidHostUsingZookeeper(): Unit = {
+    val createOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092",
+      "--entity-name", "A,B", "--entity-type", "ips", "--describe"))
+    ConfigCommand.alterConfigWithZk(null, createOpts, new DummyAdminZkClient(zkClient))
+  }
+
+  @Test(expected = classOf[InvalidRequestException])
+  def shouldFailIfUnresolvableHost(): Unit = {
+    val createOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092",
+      "--entity-name", "admin", "--entity-type", "ips", "--describe"))
+    createOpts.checkArgs()
+  }
+
+  @Test(expected = classOf[InvalidRequestException])
+  def shouldFailIfUnresolvableHostUsingZookeeper(): Unit = {
+    val createOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092",
+      "--entity-name", "admin", "--entity-type", "ips", "--describe"))
+    ConfigCommand.alterConfigWithZk(null, createOpts, new DummyAdminZkClient(zkClient))
+  }
+
   @Test
   def shouldAddClientConfigUsingZookeeper(): Unit = {
     val createOpts = new ConfigCommandOptions(Array("--zookeeper", zkConnect,
diff --git a/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala b/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala
index 6bcbb3d6e9feb..c2be2e729f6fa 100644
--- a/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala
@@ -287,6 +287,14 @@ class ClientQuotasRequestTest extends BaseRequestTest {
     assertThrows(classOf[InvalidRequestException], () => alterEntityQuotas(clientAndIpEntity, Map(RequestPercentageProp -> Some(12.34)), validateOnly = true))
   }
 
+  @Test
+  def testAlterClientQuotasBadIp(): Unit = {
+    val invalidHostPatternEntity = new ClientQuotaEntity(Map(ClientQuotaEntity.IP -> "abc-123").asJava)
+    val unresolvableHostEntity = new ClientQuotaEntity(Map(ClientQuotaEntity.IP -> "ip").asJava)
+    assertThrows(classOf[InvalidRequestException], () => alterEntityQuotas(invalidHostPatternEntity, Map(RequestPercentageProp -> Some(12.34)), validateOnly = true))
+    assertThrows(classOf[InvalidRequestException], () => alterEntityQuotas(unresolvableHostEntity, Map(RequestPercentageProp -> Some(12.34)), validateOnly = true))
+  }
+
   @Test
   def testDescribeClientQuotasInvalidFilterCombination(): Unit = {
     val ipFilterComponent = ClientQuotaFilterComponent.ofEntityType(ClientQuotaEntity.IP)
diff --git a/core/src/test/scala/unit/kafka/server/DynamicConfigTest.scala b/core/src/test/scala/unit/kafka/server/DynamicConfigTest.scala
index 3f8ffdf6bb24a..3eda6ee92840e 100644
--- a/core/src/test/scala/unit/kafka/server/DynamicConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/server/DynamicConfigTest.scala
@@ -16,10 +16,10 @@
   */
 package kafka.server
 
-import kafka.admin.AdminOperationException
 import kafka.utils.CoreUtils._
 import kafka.zk.ZooKeeperTestHarness
 import org.apache.kafka.common.config._
+import org.apache.kafka.common.errors.InvalidRequestException
 import org.junit.Test
 
 class DynamicConfigTest extends ZooKeeperTestHarness {
@@ -53,8 +53,13 @@ class DynamicConfigTest extends ZooKeeperTestHarness {
     adminZkClient.changeIpConfig("1.2.3.4", propsWith(DynamicConfig.Ip.IpConnectionRateOverrideProp, "-1"))
   }
 
-  @Test(expected = classOf[AdminOperationException])
+  @Test(expected = classOf[InvalidRequestException])
   def shouldFailIpConfigsWithInvalidIpv4Entity(): Unit = {
     adminZkClient.changeIpConfig("1,1.1.1", propsWith(DynamicConfig.Ip.IpConnectionRateOverrideProp, "2"));
   }
+
+  @Test(expected = classOf[InvalidRequestException])
+  def shouldFailIpConfigsWithBadHost(): Unit = {
+    adminZkClient.changeIpConfig("ip", propsWith(DynamicConfig.Ip.IpConnectionRateOverrideProp, "2"));
+  }
 }

From 1b0ec2a3ca79dc8da5908c41afc4ef87d2288d84 Mon Sep 17 00:00:00 2001
From: David Mao <dmao@confluent.io>
Date: Tue, 8 Dec 2020 10:39:45 -0600
Subject: [PATCH 8/9] address feedback

---
 core/src/main/scala/kafka/server/AdminManager.scala    |  8 +++++++-
 core/src/main/scala/kafka/server/DynamicConfig.scala   |  6 +-----
 .../scala/unit/kafka/admin/ConfigCommandTest.scala     | 10 +++++-----
 .../unit/kafka/server/ClientQuotasRequestTest.scala    |  4 ++--
 .../scala/unit/kafka/server/DynamicConfigTest.scala    |  5 ++---
 5 files changed, 17 insertions(+), 16 deletions(-)

diff --git a/core/src/main/scala/kafka/server/AdminManager.scala b/core/src/main/scala/kafka/server/AdminManager.scala
index 33a2435534572..45aae175ace59 100644
--- a/core/src/main/scala/kafka/server/AdminManager.scala
+++ b/core/src/main/scala/kafka/server/AdminManager.scala
@@ -995,7 +995,13 @@ class AdminManager(val config: KafkaConfig,
         case (Some(user), Some(clientId), None) => (user + "/clients/" + clientId, ConfigType.User, DynamicConfig.User.configKeys)
         case (Some(user), None, None) => (user, ConfigType.User, DynamicConfig.User.configKeys)
         case (None, Some(clientId), None) => (clientId, ConfigType.Client, DynamicConfig.Client.configKeys)
-        case (None, None, Some(ip)) => (ip, ConfigType.Ip, DynamicConfig.Ip.configKeys)
+        case (None, None, Some(ip)) =>
+          try {
+            DynamicConfig.Ip.validateIpOrHost(ip)
+          } catch {
+            case e: IllegalArgumentException => throw new InvalidRequestException(e.getMessage)
+          }
+          (ip, ConfigType.Ip, DynamicConfig.Ip.configKeys)
         case (_, _, Some(_)) => throw new InvalidRequestException(s"Invalid quota entity combination, " +
           s"IP entity should not be used with user/client ID entity.")
         case _ => throw new InvalidRequestException("Invalid client quota entity")
diff --git a/core/src/main/scala/kafka/server/DynamicConfig.scala b/core/src/main/scala/kafka/server/DynamicConfig.scala
index ac494310b7565..2decca4857c05 100644
--- a/core/src/main/scala/kafka/server/DynamicConfig.scala
+++ b/core/src/main/scala/kafka/server/DynamicConfig.scala
@@ -26,8 +26,6 @@ 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.common.errors.InvalidRequestException
-import org.apache.kafka.common.utils.Utils
 
 import scala.jdk.CollectionConverters._
 
@@ -147,12 +145,10 @@ object DynamicConfig {
 
     def validateIpOrHost(ip: String): Unit = {
       if (ip != ConfigEntityName.Default) {
-        if (!Utils.validHostPattern(ip))
-          throw new InvalidRequestException(s"$ip is not a valid hostname")
         try {
           InetAddress.getByName(ip)
         } catch {
-          case _ :UnknownHostException => throw new InvalidRequestException(s"$ip is not a valid IP or resolvable hostname")
+          case _: UnknownHostException => throw new IllegalArgumentException(s"$ip is not a valid IP or resolvable hostname")
         }
       }
     }
diff --git a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
index ef831902c5f10..f1247bbdfec81 100644
--- a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
+++ b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
@@ -28,7 +28,7 @@ import kafka.zk.{AdminZkClient, BrokerInfo, KafkaZkClient, ZooKeeperTestHarness}
 import org.apache.kafka.clients.admin._
 import org.apache.kafka.common.Node
 import org.apache.kafka.common.config.{ConfigException, ConfigResource}
-import org.apache.kafka.common.errors.{InvalidConfigurationException, InvalidRequestException}
+import org.apache.kafka.common.errors.InvalidConfigurationException
 import org.apache.kafka.common.internals.KafkaFutureImpl
 import org.apache.kafka.common.network.ListenerName
 import org.apache.kafka.common.quota.{ClientQuotaAlteration, ClientQuotaEntity, ClientQuotaFilter, ClientQuotaFilterComponent}
@@ -392,28 +392,28 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
     createOpts.checkArgs()
   }
 
-  @Test(expected = classOf[InvalidRequestException])
+  @Test(expected = classOf[IllegalArgumentException])
   def shouldFailIfInvalidHost(): Unit = {
     val createOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092",
       "--entity-name", "A,B", "--entity-type", "ips", "--describe"))
     createOpts.checkArgs()
   }
 
-  @Test(expected = classOf[InvalidRequestException])
+  @Test(expected = classOf[IllegalArgumentException])
   def shouldFailIfInvalidHostUsingZookeeper(): Unit = {
     val createOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092",
       "--entity-name", "A,B", "--entity-type", "ips", "--describe"))
     ConfigCommand.alterConfigWithZk(null, createOpts, new DummyAdminZkClient(zkClient))
   }
 
-  @Test(expected = classOf[InvalidRequestException])
+  @Test(expected = classOf[IllegalArgumentException])
   def shouldFailIfUnresolvableHost(): Unit = {
     val createOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092",
       "--entity-name", "admin", "--entity-type", "ips", "--describe"))
     createOpts.checkArgs()
   }
 
-  @Test(expected = classOf[InvalidRequestException])
+  @Test(expected = classOf[IllegalArgumentException])
   def shouldFailIfUnresolvableHostUsingZookeeper(): Unit = {
     val createOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092",
       "--entity-name", "admin", "--entity-type", "ips", "--describe"))
diff --git a/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala b/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala
index c2be2e729f6fa..de0cb175ef946 100644
--- a/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala
@@ -291,8 +291,8 @@ class ClientQuotasRequestTest extends BaseRequestTest {
   def testAlterClientQuotasBadIp(): Unit = {
     val invalidHostPatternEntity = new ClientQuotaEntity(Map(ClientQuotaEntity.IP -> "abc-123").asJava)
     val unresolvableHostEntity = new ClientQuotaEntity(Map(ClientQuotaEntity.IP -> "ip").asJava)
-    assertThrows(classOf[InvalidRequestException], () => alterEntityQuotas(invalidHostPatternEntity, Map(RequestPercentageProp -> Some(12.34)), validateOnly = true))
-    assertThrows(classOf[InvalidRequestException], () => alterEntityQuotas(unresolvableHostEntity, Map(RequestPercentageProp -> Some(12.34)), validateOnly = true))
+    assertThrows(classOf[InvalidRequestException], () => alterEntityQuotas(invalidHostPatternEntity, Map(IpConnectionRateProp -> Some(50.0)), validateOnly = true))
+    assertThrows(classOf[InvalidRequestException], () => alterEntityQuotas(unresolvableHostEntity, Map(IpConnectionRateProp -> Some(50.0)), validateOnly = true))
   }
 
   @Test
diff --git a/core/src/test/scala/unit/kafka/server/DynamicConfigTest.scala b/core/src/test/scala/unit/kafka/server/DynamicConfigTest.scala
index 3eda6ee92840e..eefae1bcd20e8 100644
--- a/core/src/test/scala/unit/kafka/server/DynamicConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/server/DynamicConfigTest.scala
@@ -19,7 +19,6 @@ package kafka.server
 import kafka.utils.CoreUtils._
 import kafka.zk.ZooKeeperTestHarness
 import org.apache.kafka.common.config._
-import org.apache.kafka.common.errors.InvalidRequestException
 import org.junit.Test
 
 class DynamicConfigTest extends ZooKeeperTestHarness {
@@ -53,12 +52,12 @@ class DynamicConfigTest extends ZooKeeperTestHarness {
     adminZkClient.changeIpConfig("1.2.3.4", propsWith(DynamicConfig.Ip.IpConnectionRateOverrideProp, "-1"))
   }
 
-  @Test(expected = classOf[InvalidRequestException])
+  @Test(expected = classOf[IllegalArgumentException])
   def shouldFailIpConfigsWithInvalidIpv4Entity(): Unit = {
     adminZkClient.changeIpConfig("1,1.1.1", propsWith(DynamicConfig.Ip.IpConnectionRateOverrideProp, "2"));
   }
 
-  @Test(expected = classOf[InvalidRequestException])
+  @Test(expected = classOf[IllegalArgumentException])
   def shouldFailIpConfigsWithBadHost(): Unit = {
     adminZkClient.changeIpConfig("ip", propsWith(DynamicConfig.Ip.IpConnectionRateOverrideProp, "2"));
   }

From a781ee70fd166c1e0773da21112a2dcd721b658a Mon Sep 17 00:00:00 2001
From: David Mao <dmao@confluent.io>
Date: Wed, 9 Dec 2020 11:21:55 -0600
Subject: [PATCH 9/9] address feedback

---
 .../scala/kafka/admin/ConfigCommand.scala     |  5 +++-
 .../scala/kafka/server/AdminManager.scala     |  7 ++---
 .../scala/kafka/server/DynamicConfig.scala    |  5 ++--
 .../main/scala/kafka/zk/AdminZkClient.scala   |  3 ++-
 .../unit/kafka/admin/ConfigCommandTest.scala  |  8 +++---
 .../server/ClientQuotasRequestTest.scala      | 27 ++++++++++++++-----
 .../unit/kafka/server/DynamicConfigTest.scala |  5 ++--
 7 files changed, 38 insertions(+), 22 deletions(-)

diff --git a/core/src/main/scala/kafka/admin/ConfigCommand.scala b/core/src/main/scala/kafka/admin/ConfigCommand.scala
index 773ade981e1bc..7932c8dddb397 100644
--- a/core/src/main/scala/kafka/admin/ConfigCommand.scala
+++ b/core/src/main/scala/kafka/admin/ConfigCommand.scala
@@ -886,7 +886,10 @@ object ConfigCommand extends Config {
       }
 
       if (hasEntityName && entityTypeVals.contains(ConfigType.Ip)) {
-        Seq(entityName, ip).filter(options.has(_)).map(options.valueOf(_)).foreach(DynamicConfig.Ip.validateIpOrHost)
+        Seq(entityName, ip).filter(options.has(_)).map(options.valueOf(_)).foreach { ipEntity =>
+          if (!DynamicConfig.Ip.isValidIpEntity(ipEntity))
+            throw new IllegalArgumentException(s"The entity name for ${entityTypeVals.head} must be a valid IP or resolvable host, but it is: $ipEntity")
+        }
       }
 
       if (options.has(describeOpt) && entityTypeVals.contains(BrokerLoggerConfigType) && !hasEntityName)
diff --git a/core/src/main/scala/kafka/server/AdminManager.scala b/core/src/main/scala/kafka/server/AdminManager.scala
index 45aae175ace59..f1bd1e276c4f1 100644
--- a/core/src/main/scala/kafka/server/AdminManager.scala
+++ b/core/src/main/scala/kafka/server/AdminManager.scala
@@ -996,11 +996,8 @@ class AdminManager(val config: KafkaConfig,
         case (Some(user), None, None) => (user, ConfigType.User, DynamicConfig.User.configKeys)
         case (None, Some(clientId), None) => (clientId, ConfigType.Client, DynamicConfig.Client.configKeys)
         case (None, None, Some(ip)) =>
-          try {
-            DynamicConfig.Ip.validateIpOrHost(ip)
-          } catch {
-            case e: IllegalArgumentException => throw new InvalidRequestException(e.getMessage)
-          }
+          if (!DynamicConfig.Ip.isValidIpEntity(ip))
+            throw new InvalidRequestException(s"$ip is not a valid IP or resolvable host.")
           (ip, ConfigType.Ip, DynamicConfig.Ip.configKeys)
         case (_, _, Some(_)) => throw new InvalidRequestException(s"Invalid quota entity combination, " +
           s"IP entity should not be used with user/client ID entity.")
diff --git a/core/src/main/scala/kafka/server/DynamicConfig.scala b/core/src/main/scala/kafka/server/DynamicConfig.scala
index 2decca4857c05..f00c610776aa4 100644
--- a/core/src/main/scala/kafka/server/DynamicConfig.scala
+++ b/core/src/main/scala/kafka/server/DynamicConfig.scala
@@ -143,14 +143,15 @@ object DynamicConfig {
 
     def validate(props: Properties) = DynamicConfig.validate(ipConfigs, props, customPropsAllowed = false)
 
-    def validateIpOrHost(ip: String): Unit = {
+    def isValidIpEntity(ip: String): Boolean = {
       if (ip != ConfigEntityName.Default) {
         try {
           InetAddress.getByName(ip)
         } catch {
-          case _: UnknownHostException => throw new IllegalArgumentException(s"$ip is not a valid IP or resolvable hostname")
+          case _: UnknownHostException => return false
         }
       }
+      true
     }
   }
 
diff --git a/core/src/main/scala/kafka/zk/AdminZkClient.scala b/core/src/main/scala/kafka/zk/AdminZkClient.scala
index b6974f362961c..1e994b4ecac85 100644
--- a/core/src/main/scala/kafka/zk/AdminZkClient.scala
+++ b/core/src/main/scala/kafka/zk/AdminZkClient.scala
@@ -393,7 +393,8 @@ class AdminZkClient(zkClient: KafkaZkClient) extends Logging {
    * @param configs properties to validate for the IP
    */
   def validateIpConfig(ip: String, configs: Properties): Unit = {
-    DynamicConfig.Ip.validateIpOrHost(ip)
+    if (!DynamicConfig.Ip.isValidIpEntity(ip))
+      throw new AdminOperationException(s"$ip is not a valid IP or resolvable host.")
     DynamicConfig.Ip.validate(configs)
   }
 
diff --git a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
index f1247bbdfec81..1eb18c9815c7f 100644
--- a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
+++ b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
@@ -401,9 +401,9 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
 
   @Test(expected = classOf[IllegalArgumentException])
   def shouldFailIfInvalidHostUsingZookeeper(): Unit = {
-    val createOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092",
+    val createOpts = new ConfigCommandOptions(Array("--zookeeper", zkConnect,
       "--entity-name", "A,B", "--entity-type", "ips", "--describe"))
-    ConfigCommand.alterConfigWithZk(null, createOpts, new DummyAdminZkClient(zkClient))
+    createOpts.checkArgs()
   }
 
   @Test(expected = classOf[IllegalArgumentException])
@@ -415,9 +415,9 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
 
   @Test(expected = classOf[IllegalArgumentException])
   def shouldFailIfUnresolvableHostUsingZookeeper(): Unit = {
-    val createOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092",
+    val createOpts = new ConfigCommandOptions(Array("--zookeeper", zkConnect,
       "--entity-name", "admin", "--entity-type", "ips", "--describe"))
-    ConfigCommand.alterConfigWithZk(null, createOpts, new DummyAdminZkClient(zkClient))
+    createOpts.checkArgs()
   }
 
   @Test
diff --git a/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala b/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala
index de0cb175ef946..b72f3b9121b17 100644
--- a/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala
@@ -279,20 +279,31 @@ class ClientQuotasRequestTest extends BaseRequestTest {
     alterEntityQuotas(entity, Map((ProducerByteRateProp -> Some(10000.5))), validateOnly = true)
   }
 
+  private def expectInvalidRequestWithMessage(runnable: => Unit, expectedMessage: String): Unit = {
+    val exception = assertThrows(classOf[InvalidRequestException], () => runnable)
+    assertTrue(s"Expected message $exception to contain $expectedMessage", exception.getMessage.contains(expectedMessage))
+  }
+
   @Test
   def testAlterClientQuotasInvalidEntityCombination(): Unit = {
     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)
-    assertThrows(classOf[InvalidRequestException], () => alterEntityQuotas(userAndIpEntity, Map(RequestPercentageProp -> Some(12.34)), validateOnly = true))
-    assertThrows(classOf[InvalidRequestException], () => alterEntityQuotas(clientAndIpEntity, Map(RequestPercentageProp -> Some(12.34)), validateOnly = true))
+    val expectedExceptionMessage = "Invalid quota entity combination"
+    expectInvalidRequestWithMessage(alterEntityQuotas(userAndIpEntity, Map(RequestPercentageProp -> Some(12.34)),
+      validateOnly = true), expectedExceptionMessage)
+    expectInvalidRequestWithMessage(alterEntityQuotas(clientAndIpEntity, Map(RequestPercentageProp -> Some(12.34)),
+      validateOnly = true), expectedExceptionMessage)
   }
 
   @Test
   def testAlterClientQuotasBadIp(): Unit = {
     val invalidHostPatternEntity = new ClientQuotaEntity(Map(ClientQuotaEntity.IP -> "abc-123").asJava)
     val unresolvableHostEntity = new ClientQuotaEntity(Map(ClientQuotaEntity.IP -> "ip").asJava)
-    assertThrows(classOf[InvalidRequestException], () => alterEntityQuotas(invalidHostPatternEntity, Map(IpConnectionRateProp -> Some(50.0)), validateOnly = true))
-    assertThrows(classOf[InvalidRequestException], () => alterEntityQuotas(unresolvableHostEntity, Map(IpConnectionRateProp -> Some(50.0)), validateOnly = true))
+    val expectedExceptionMessage = "not a valid IP"
+    expectInvalidRequestWithMessage(alterEntityQuotas(invalidHostPatternEntity, Map(IpConnectionRateProp -> Some(50.0)),
+      validateOnly = true), expectedExceptionMessage)
+    expectInvalidRequestWithMessage(alterEntityQuotas(unresolvableHostEntity, Map(IpConnectionRateProp -> Some(50.0)),
+      validateOnly = true), expectedExceptionMessage)
   }
 
   @Test
@@ -300,9 +311,11 @@ class ClientQuotasRequestTest extends BaseRequestTest {
     val ipFilterComponent = ClientQuotaFilterComponent.ofEntityType(ClientQuotaEntity.IP)
     val userFilterComponent = ClientQuotaFilterComponent.ofEntityType(ClientQuotaEntity.USER)
     val clientIdFilterComponent = ClientQuotaFilterComponent.ofEntityType(ClientQuotaEntity.CLIENT_ID)
-
-    assertThrows(classOf[InvalidRequestException], () => describeClientQuotas(ClientQuotaFilter.contains(List(ipFilterComponent, userFilterComponent).asJava)))
-    assertThrows(classOf[InvalidRequestException], () => describeClientQuotas(ClientQuotaFilter.contains(List(ipFilterComponent, clientIdFilterComponent).asJava)))
+    val expectedExceptionMessage = "Invalid entity filter component combination"
+    expectInvalidRequestWithMessage(describeClientQuotas(ClientQuotaFilter.contains(List(ipFilterComponent, userFilterComponent).asJava)),
+      expectedExceptionMessage)
+    expectInvalidRequestWithMessage(describeClientQuotas(ClientQuotaFilter.contains(List(ipFilterComponent, clientIdFilterComponent).asJava)),
+      expectedExceptionMessage)
   }
 
   // Entities to be matched against.
diff --git a/core/src/test/scala/unit/kafka/server/DynamicConfigTest.scala b/core/src/test/scala/unit/kafka/server/DynamicConfigTest.scala
index eefae1bcd20e8..ce65df9b72566 100644
--- a/core/src/test/scala/unit/kafka/server/DynamicConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/server/DynamicConfigTest.scala
@@ -16,6 +16,7 @@
   */
 package kafka.server
 
+import kafka.admin.AdminOperationException
 import kafka.utils.CoreUtils._
 import kafka.zk.ZooKeeperTestHarness
 import org.apache.kafka.common.config._
@@ -52,12 +53,12 @@ class DynamicConfigTest extends ZooKeeperTestHarness {
     adminZkClient.changeIpConfig("1.2.3.4", propsWith(DynamicConfig.Ip.IpConnectionRateOverrideProp, "-1"))
   }
 
-  @Test(expected = classOf[IllegalArgumentException])
+  @Test(expected = classOf[AdminOperationException])
   def shouldFailIpConfigsWithInvalidIpv4Entity(): Unit = {
     adminZkClient.changeIpConfig("1,1.1.1", propsWith(DynamicConfig.Ip.IpConnectionRateOverrideProp, "2"));
   }
 
-  @Test(expected = classOf[IllegalArgumentException])
+  @Test(expected = classOf[AdminOperationException])
   def shouldFailIpConfigsWithBadHost(): Unit = {
     adminZkClient.changeIpConfig("ip", propsWith(DynamicConfig.Ip.IpConnectionRateOverrideProp, "2"));
   }