Repository: apache/kafka
File: core/src/main/scala/kafka/server/ClientQuotaManager.scala (Scala)

Code before:
    } else if (!shouldAdd && isActive) {
      activeQuotaEntities.compute(activeQuotaType, (_, currentValue) => if (currentValue <= 1) 0 else currentValue - 1)
      if (activeQuotaEntities.get(activeQuotaType) == 0) {
        quotaTypesEnabled &= ~activeQuotaType
      }
    }

    val quotaTypes = List(
      QuotaTypes.UserClientIdQuotaEnabled -> "UserClientIdQuota",
      QuotaTypes.ClientIdQuotaEnabled     -> "ClientIdQuota",
      QuotaTypes.UserQuotaEnabled         -> "UserQuota"
    )

    val activeEntities = quotaTypes.collect {
      case (k, name) if activeQuotaEntities.get(k) > 0 => name
    }.mkString(", ")
    info(s"Quota types enabled has been changed to $quotaTypesEnabled with active quota entities: [$activeEntities]")
  }

  /**
   * Updates metrics configs. This is invoked when quota configs are updated when partitions leaders change 
   * and custom callbacks that implement partition-based quotas have updated quotas.
   *
   * @param updatedQuotaEntity If set to one entity and quotas have only been enabled at one
   *    level, then an optimized update is performed with a single metric update. If None is provided,
   *    or if custom callbacks are used or if multi-level quotas have been enabled, all metric configs
   *    are checked and updated if required.
   */
  def updateQuotaMetricConfigs(updatedQuotaEntity: Option[KafkaQuotaEntity] = None): Unit = {
    val allMetrics = metrics.metrics()

    // If using custom quota callbacks or if multiple-levels of quotas are defined or
    // if this is a default quota update, traverse metrics to find all affected values.
    // Otherwise, update just the single matching one.
    val singleUpdate = quotaTypesEnabled match {
      case QuotaTypes.NoQuotas | QuotaTypes.ClientIdQuotaEnabled | QuotaTypes.UserQuotaEnabled | QuotaTypes.UserClientIdQuotaEnabled =>
        updatedQuotaEntity.nonEmpty
      case _ => false
    }
    if (singleUpdate) {
      val quotaEntity = updatedQuotaEntity.getOrElse(throw new IllegalStateException("Quota entity not specified"))
      val user = quotaEntity.sanitizedUser
      val clientId = quotaEntity.clientId
      val metricTags = Map(DefaultTags.User -> user, DefaultTags.ClientId -> clientId)

      val quotaMetricName = clientQuotaMetricName(metricTags)
      // Change the underlying metric config if the sensor has been created
      val metric = allMetrics.get(quotaMetricName)
      if (metric != null) {
        Option(quotaLimit(metricTags.asJava)).foreach { newQuota =>
          info(s"Sensor for $quotaEntity already exists. Changing quota to $newQuota in MetricConfig")

Review comment (suggestion):
We should probably only do `updateQuotaTypes()` only if custom quotas are not used. 

Code after:
      case KafkaQuotaEntity(Some(_), Some(_)) => QuotaTypes.UserClientIdQuotaEnabled
      case KafkaQuotaEntity(Some(_), None) => QuotaTypes.UserQuotaEnabled
      case KafkaQuotaEntity(None, Some(_)) => QuotaTypes.ClientIdQuotaEnabled
      case _ => QuotaTypes.NoQuotas
    }

    if (shouldAdd && !isActive) {
      activeQuotaEntities.compute(activeQuotaType, (_, currentValue) => if (currentValue == 0) 1 else currentValue + 1)
      quotaTypesEnabled |= activeQuotaType
    } else if (!shouldAdd && isActive) {
      activeQuotaEntities.compute(activeQuotaType, (_, currentValue) => if (currentValue <= 1) 0 else currentValue - 1)
      if (activeQuotaEntities.get(activeQuotaType) == 0) {
        quotaTypesEnabled &= ~activeQuotaType
      }
    }

    val quotaTypes = List(
      QuotaTypes.UserClientIdQuotaEnabled -> "UserClientIdQuota",
      QuotaTypes.ClientIdQuotaEnabled     -> "ClientIdQuota",
      QuotaTypes.UserQuotaEnabled         -> "UserQuota"
    )

    val activeEntities = quotaTypes.collect {
      case (k, name) if activeQuotaEntities.get(k) > 0 => name
    }.mkString(", ")
    info(s"Quota types enabled has been changed to $quotaTypesEnabled with active quota entities: [$activeEntities]")
  }

  /**
   * Updates metrics configs. This is invoked when quota configs are updated when partitions leaders change 
   * and custom callbacks that implement partition-based quotas have updated quotas.
   *
   * @param updatedQuotaEntity If set to one entity and quotas have only been enabled at one
   *    level, then an optimized update is performed with a single metric update. If None is provided,
   *    or if custom callbacks are used or if multi-level quotas have been enabled, all metric configs
   *    are checked and updated if required.
   */
  def updateQuotaMetricConfigs(updatedQuotaEntity: Option[KafkaQuotaEntity] = None): Unit = {
    val allMetrics = metrics.metrics()

    // If using custom quota callbacks or if multiple-levels of quotas are defined or
    // if this is a default quota update, traverse metrics to find all affected values.
    // Otherwise, update just the single matching one.
    val singleUpdate = quotaTypesEnabled match {
      case QuotaTypes.NoQuotas | QuotaTypes.ClientIdQuotaEnabled | QuotaTypes.UserQuotaEnabled | QuotaTypes.UserClientIdQuotaEnabled =>
        updatedQuotaEntity.nonEmpty
      case _ => false
    }
    if (singleUpdate) {
      val quotaEntity = updatedQuotaEntity.getOrElse(throw new IllegalStateException("Quota entity not specified"))
      val user = quotaEntity.sanitizedUser