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

Code before:
      updateQuotaMetricConfigs(updatedEntity)

    } finally {
      lock.writeLock().unlock()
    }
  }

  /**
   * Helper method to update quota types counts and quotaTypesEnabled flag.
   * @param quotaTypeKey The QuotaTypes constant (e.g., QuotaTypes.UserClientIdQuotaEnabled)
   * @param increment True to increment count, false to decrement
   */
  private def updateQuotaTypes(quotaTypeKey: Int, increment: Boolean): Unit = {
    if (quotaTypeKey == QuotaTypes.NoQuotas) {
      return
    }
    val previousQuotaTypesEnabled = quotaTypesEnabled

    // Update activeQuotaTypes counts
    activeQuotaTypes.compute(quotaTypeKey, (_, count) =>
      if (increment) Option(count).getOrElse(0) + 1
      else if (Option(count).exists(_ > 1)) count - 1
      else 0
    )

    // Update quotaTypesEnabled based on activeQuotaTypes counts
    quotaTypesEnabled = clientQuotaCallbackPlugin match {
      case Some(_) => QuotaTypes.CustomQuotas
      case None =>
        var newQuotaTypes = QuotaTypes.NoQuotas
        if (activeQuotaTypes.getOrDefault(QuotaTypes.UserClientIdQuotaEnabled, 0) > 0)
          newQuotaTypes |= QuotaTypes.UserClientIdQuotaEnabled
        if (activeQuotaTypes.getOrDefault(QuotaTypes.UserQuotaEnabled, 0) > 0)
          newQuotaTypes |= QuotaTypes.UserQuotaEnabled
        if (activeQuotaTypes.getOrDefault(QuotaTypes.ClientIdQuotaEnabled, 0) > 0)
          newQuotaTypes |= QuotaTypes.ClientIdQuotaEnabled
        newQuotaTypes
    }

    if (previousQuotaTypesEnabled != quotaTypesEnabled) {
      debug(s"Quota types enabled changed from $previousQuotaTypesEnabled to $quotaTypesEnabled")
    }
  }

  /**
   * 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

Review comment (suggestion):
maybe would be helpful if this comment explained the bitwise operation, potentially with an example. 

e.g. quotaTypesEnabled is the bitwise representation of what quota types are enabled. For instance, if quotaTypesEnabled == 6, then we know we have UserClientIdQuotaEnabled and UserQuotaEnabled since their bit representations are 4 and 2. (4 | 2 = 6)

Code after:
      lock.writeLock().unlock()
    }
  }

  /**
   * Updates `quotaTypesEnabled` by performing a bitwise OR operation to combine the enabled quota types.
   * This method ensures that the `quotaTypesEnabled` field reflects the active quota types based on the
   * current state of `activeQuotaEntities`.
   * For example:
   *  - If UserQuotaEnabled = 2 and ClientIdQuotaEnabled = 1, then quotaTypesEnabled = 3 (2 | 1 = 3)
   *  - If UserClientIdQuotaEnabled = 4 and UserQuotaEnabled = 1, then quotaTypesEnabled =  (4 | 1 = 5)
   *  - If UserClientIdQuotaEnabled = 4 and ClientIdQuotaEnabled = 2, then quotaTypesEnabled = 6 (4 | 2 = 6)
   *  - If all three are enabled (1 | 2 | 4), then quotaTypesEnabled = 7
   *
   *  @param quotaEntity The entity for which the quota is being updated, which can be a combination of user and client-id.
   *  @param shouldAdd   A boolean indicating whether to add or remove the quota entity.
   */
  private def updateQuotaTypes(quotaEntity: KafkaQuotaEntity, shouldAdd: Boolean): Unit = {

    if (quotaTypesEnabled == QuotaTypes.CustomQuotas) {
      // If custom quotas are enabled, we do not need to update quota types
      return
    }

    val isActive =  quotaCallback match {
      case callback: DefaultQuotaCallback => callback.getActiveQuotasEntities.contains(quotaEntity)
      case _ => true
    }

    val activeQuotaType = quotaEntity match {
      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"
    )