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

Code before:
 * @param config @ClientQuotaManagerConfig quota configs
 * @param metrics @Metrics Metrics instance
 * @param quotaType Quota type of this quota manager
 * @param time @Time object to use
 * @param threadNamePrefix The thread prefix to use
 * @param clientQuotaCallbackPlugin An optional @ClientQuotaCallback and 
 *                                  wrap it in a {@link org.apache.kafka.common.internals.Plugin}
 */
class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
                         private val metrics: Metrics,
                         private val quotaType: QuotaType,
                         private val time: Time,
                         private val threadNamePrefix: String,
                         private val clientQuotaCallbackPlugin: Option[Plugin[ClientQuotaCallback]] = None) extends Logging {

  private val lock = new ReentrantReadWriteLock()
  private val sensorAccessor = new SensorAccess(lock, metrics)
  private val quotaCallback = clientQuotaCallbackPlugin match {
    case Some(plugin) => plugin.get()
    case None => new DefaultQuotaCallback
  }
  private val clientQuotaType = QuotaType.toClientQuotaType(quotaType)
  private val activeQuotaEntities = new ConcurrentHashMap[KafkaQuotaEntity, Boolean]()

  @volatile
  private var quotaTypesEnabled = clientQuotaCallbackPlugin match {
    case Some(_) => QuotaTypes.CustomQuotas
    case None => QuotaTypes.NoQuotas
  }

  private val delayQueueSensor = metrics.sensor(quotaType.toString + "-delayQueue")
  delayQueueSensor.add(metrics.metricName("queue-size", quotaType.toString,
    "Tracks the size of the delay queue"), new CumulativeSum())

  private val delayQueue = new DelayQueue[ThrottledChannel]()
  private[server] val throttledChannelReaper = new ThrottledChannelReaper(delayQueue, threadNamePrefix)
  start() // Use start method to keep spotbugs happy
  private def start(): Unit = {
    throttledChannelReaper.start()
  }

  /**
   * Reaper thread that triggers channel unmute callbacks on all throttled channels
   * @param delayQueue DelayQueue to dequeue from
   */
  class ThrottledChannelReaper(delayQueue: DelayQueue[ThrottledChannel], prefix: String) extends ShutdownableThread(
    s"${prefix}ThrottledChannelReaper-$quotaType", false) {

    override def doWork(): Unit = {
      val throttledChannel: ThrottledChannel = delayQueue.poll(1, TimeUnit.SECONDS)
      if (throttledChannel != null) {

Review comment (performance):
Nit: Since we're doing a lot of bit-manipulation on this value, it could make sense to convert it to an [enum set](https://docs.oracle.com/javase/8/docs/api/java/util/EnumSet.html). This would provide an efficient bit-vector interface and allow explicit "get/set/remove" operations. But feel free to ignore this comment if it doesn't make sense in scala. 

Code after:
 * @param config @ClientQuotaManagerConfig quota configs
 * @param metrics @Metrics Metrics instance
 * @param quotaType Quota type of this quota manager
 * @param time @Time object to use
 * @param threadNamePrefix The thread prefix to use
 * @param clientQuotaCallbackPlugin An optional @ClientQuotaCallback and 
 *                                  wrap it in a {@link org.apache.kafka.common.internals.Plugin}
 */
class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
                         private val metrics: Metrics,
                         private val quotaType: QuotaType,
                         private val time: Time,
                         private val threadNamePrefix: String,
                         private val clientQuotaCallbackPlugin: Option[Plugin[ClientQuotaCallback]] = None) extends Logging {

  private val lock = new ReentrantReadWriteLock()
  private val sensorAccessor = new SensorAccess(lock, metrics)
  private val quotaCallback = clientQuotaCallbackPlugin match {
    case Some(plugin) => plugin.get()
    case None => new DefaultQuotaCallback
  }
  private val clientQuotaType = QuotaType.toClientQuotaType(quotaType)
  private val activeQuotaEntities = new ConcurrentHashMap[Int, Int]() // Key is QuotaTypes, value is the number of active quota entities of that type



  @volatile
  var quotaTypesEnabled = clientQuotaCallbackPlugin match {
    case Some(_) => QuotaTypes.CustomQuotas
    case None => QuotaTypes.NoQuotas
  }

  private val delayQueueSensor = metrics.sensor(quotaType.toString + "-delayQueue")
  delayQueueSensor.add(metrics.metricName("queue-size", quotaType.toString,
    "Tracks the size of the delay queue"), new CumulativeSum())

  private val delayQueue = new DelayQueue[ThrottledChannel]()
  private[server] val throttledChannelReaper = new ThrottledChannelReaper(delayQueue, threadNamePrefix)
  start() // Use start method to keep spotbugs happy
  private def start(): Unit = {
    throttledChannelReaper.start()
  }

  /**
   * Reaper thread that triggers channel unmute callbacks on all throttled channels
   * @param delayQueue DelayQueue to dequeue from
   */
  class ThrottledChannelReaper(delayQueue: DelayQueue[ThrottledChannel], prefix: String) extends ShutdownableThread(
    s"${prefix}ThrottledChannelReaper-$quotaType", false) {

    override def doWork(): Unit = {