Repository: apache/kafka
File: core/src/main/scala/kafka/admin/ConfigCommand.scala (Scala)

Code before:
    if (entityTypes.size != entityNames.size)
      throw new IllegalArgumentException("Exactly one entity name must be specified for every entity type")
    getAllClientQuotasConfigs(adminClient, entityTypes, entityNames).headOption.map(_._2.asScala).getOrElse(Map.empty)
  }

  private def getAllClientQuotasConfigs(adminClient: Admin, entityTypes: List[String], entityNames: List[String]) = {
    val components = entityTypes.map(Some(_)).zipAll(entityNames.map(Some(_)), None, None).map { case (entityTypeOpt, entityNameOpt) =>
      val entityType = entityTypeOpt match {
        case Some(UserType) => ClientQuotaEntity.USER
        case Some(ClientType) => ClientQuotaEntity.CLIENT_ID
        case Some(IpType) => 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")
      }
      entityNameOpt match {
        case Some("") => ClientQuotaFilterComponent.ofDefaultEntity(entityType)
        case Some(name) => ClientQuotaFilterComponent.ofEntity(entityType, name)
        case None => ClientQuotaFilterComponent.ofEntityType(entityType)
      }
    }

    adminClient.describeClientQuotas(ClientQuotaFilter.containsOnly(components.asJava)).entities.get(30, TimeUnit.SECONDS).asScala
  }

  private def listGroupConfigResources(adminClient: Admin): Option[java.util.Collection[ConfigResource]] = {
    try {
      Some(adminClient.listConfigResources(java.util.Set.of(ConfigResource.Type.GROUP), new ListConfigResourcesOptions).all.get)
    } catch {
      // (KIP-1142) 4.1 and later admin client connecting to an older broker will not be able to list group config resources
      case e: ExecutionException =>
        e.getCause match {
          case _: UnsupportedVersionException =>
            Option.empty
          case _ => throw e
        }
      case e: Throwable => throw e
    }
  }


  class ConfigCommandOptions(args: Array[String]) extends CommandDefaultOptions(args) {
    val bootstrapServerOpt: OptionSpec[String] = parser.accepts("bootstrap-server", "The Kafka servers to connect to.")
      .withRequiredArg
      .describedAs("server to connect to")
      .ofType(classOf[String])
    val bootstrapControllerOpt: OptionSpec[String] = parser.accepts("bootstrap-controller", "The Kafka controllers to connect to.")
      .withRequiredArg
      .describedAs("controller to connect to")
      .ofType(classOf[String])
    val commandConfigOpt: OptionSpec[String] = parser.accepts("command-config", "Property file containing configs to be passed to Admin Client. " +
      "This is used only with --bootstrap-server option for describing and altering broker configs.")

Review comment (suggestion):
```scala
    try {
      Some(adminClient.listConfigResources(java.util.Set.of(ConfigResource.Type.GROUP), new ListConfigResourcesOptions).all.get)
    } catch {
      // (KIP-1142) 4.1+ admin client vs older broker: treat UnsupportedVersion as None
      case e: ExecutionException if e.getCause.isInstanceOf[UnsupportedVersionException] => None
    }
```

Code after:
    if (entityTypes.size != entityNames.size)
      throw new IllegalArgumentException("Exactly one entity name must be specified for every entity type")
    getAllClientQuotasConfigs(adminClient, entityTypes, entityNames).headOption.map(_._2.asScala).getOrElse(Map.empty)
  }

  private def getAllClientQuotasConfigs(adminClient: Admin, entityTypes: List[String], entityNames: List[String]) = {
    val components = entityTypes.map(Some(_)).zipAll(entityNames.map(Some(_)), None, None).map { case (entityTypeOpt, entityNameOpt) =>
      val entityType = entityTypeOpt match {
        case Some(UserType) => ClientQuotaEntity.USER
        case Some(ClientType) => ClientQuotaEntity.CLIENT_ID
        case Some(IpType) => 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")
      }
      entityNameOpt match {
        case Some("") => ClientQuotaFilterComponent.ofDefaultEntity(entityType)
        case Some(name) => ClientQuotaFilterComponent.ofEntity(entityType, name)
        case None => ClientQuotaFilterComponent.ofEntityType(entityType)
      }
    }

    adminClient.describeClientQuotas(ClientQuotaFilter.containsOnly(components.asJava)).entities.get(30, TimeUnit.SECONDS).asScala
  }

  private def listGroupConfigResources(adminClient: Admin): Option[java.util.Collection[ConfigResource]] = {
    try {
      Some(adminClient.listConfigResources(java.util.Set.of(ConfigResource.Type.GROUP), new ListConfigResourcesOptions).all.get)
    } catch {
      // (KIP-1142) 4.1+ admin client vs older broker: treat UnsupportedVersionException and ClusterAuthorizationException as None
      case e: ExecutionException if e.getCause.isInstanceOf[UnsupportedVersionException] => None
      case e: ExecutionException if e.getCause.isInstanceOf[ClusterAuthorizationException] => None
      case e: ExecutionException => throw e.getCause
    }
  }


  class ConfigCommandOptions(args: Array[String]) extends CommandDefaultOptions(args) {
    val bootstrapServerOpt: OptionSpec[String] = parser.accepts("bootstrap-server", "The Kafka servers to connect to.")
      .withRequiredArg
      .describedAs("server to connect to")
      .ofType(classOf[String])
    val bootstrapControllerOpt: OptionSpec[String] = parser.accepts("bootstrap-controller", "The Kafka controllers to connect to.")
      .withRequiredArg
      .describedAs("controller to connect to")
      .ofType(classOf[String])
    val commandConfigOpt: OptionSpec[String] = parser.accepts("command-config", "Property file containing configs to be passed to Admin Client. " +
      "This is used only with --bootstrap-server option for describing and altering broker configs.")
      .withRequiredArg
      .describedAs("command config property file")
      .ofType(classOf[String])
    val alterOpt: OptionSpecBuilder = parser.accepts("alter", "Alter the configuration for the entity.")