MINOR: Simplify SensorAccess usage

I was investigating an exception in this code and found a few
opportunities for making it clearer.

I also added the `out` folder to `.gitignore` as IntelliJ sometimes
uses that as the build folder.

Author: Ismael Juma <ismael@juma.me.uk>

Reviewers: Rajini Sivaram <rajinisivaram@googlemail.com>

Closes #3552 from ijuma/minor-quota-improvements

.gitignore

@@ -3,6 +3,7 @@ dist
 target/
 build/
 build_eclipse/
+out/
 .gradle/
 lib_managed/
 src_managed/

core/src/main/scala/kafka/server/ClientQuotaManager.scala

@@ -138,10 +138,12 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
                          private val time: Time) extends Logging {
   private val overriddenQuota = new ConcurrentHashMap[QuotaId, Quota]()
   private val staticConfigClientIdQuota = Quota.upperBound(config.quotaBytesPerSecondDefault)
-  @volatile private var quotaTypesEnabled = if (config.quotaBytesPerSecondDefault == Long.MaxValue) QuotaTypes.NoQuotas else QuotaTypes.ClientIdQuotaEnabled
+  @volatile private var quotaTypesEnabled =
+    if (config.quotaBytesPerSecondDefault == Long.MaxValue) QuotaTypes.NoQuotas
+    else QuotaTypes.ClientIdQuotaEnabled
   private val lock = new ReentrantReadWriteLock()
   private val delayQueue = new DelayQueue[ThrottledResponse]()
-  private val sensorAccessor = new SensorAccess
+  private val sensorAccessor = new SensorAccess(lock, metrics)
   val throttledRequestReaper = new ThrottledRequestReaper(delayQueue)
 
   private val delayQueueSensor = metrics.sensor(quotaType + "-delayQueue")
@@ -392,24 +394,19 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
       sensorAccessor.getOrCreate(
         getQuotaSensorName(clientQuotaEntity.quotaId),
         ClientQuotaManagerConfig.InactiveSensorExpirationTimeSeconds,
-        lock, metrics,
+        clientRateMetricName(clientQuotaEntity.sanitizedUser, clientQuotaEntity.clientId),
-        () => clientRateMetricName(clientQuotaEntity.sanitizedUser, clientQuotaEntity.clientId),
+        Some(getQuotaMetricConfig(clientQuotaEntity.quota)),
-        () => getQuotaMetricConfig(clientQuotaEntity.quota),
+        new Rate
-        () => measurableStat
       ),
       sensorAccessor.getOrCreate(getThrottleTimeSensorName(clientQuotaEntity.quotaId),
         ClientQuotaManagerConfig.InactiveSensorExpirationTimeSeconds,
-        lock,
+        throttleMetricName(clientQuotaEntity),
-        metrics,
+        None,
-        () => throttleMetricName(clientQuotaEntity),
+        new Avg
-        () => null,
-        () => new Avg()
       )
     )
   }
 
-  private def measurableStat: MeasurableStat = new Rate()
-
   private def getThrottleTimeSensorName(quotaId: QuotaId): String = quotaType + "ThrottleTime-" + quotaId.sanitizedUser.getOrElse("") + ':' + quotaId.clientId.getOrElse("")
 
   private def getQuotaSensorName(quotaId: QuotaId): String = quotaType + "-" + quotaId.sanitizedUser.getOrElse("") + ':' + quotaId.clientId.getOrElse("")
@@ -425,10 +422,9 @@ class ClientQuotaManager(private val config: ClientQuotaManagerConfig,
     sensorAccessor.getOrCreate(
         sensorName,
         ClientQuotaManagerConfig.InactiveSensorExpirationTimeSeconds,
-        lock, metrics,
+        metricName,
-        () => metricName,
+        None,
-        () => null,
+        new Rate
-        () => measurableStat
       )
   }
 

core/src/main/scala/kafka/server/ReplicationQuotaManager.scala

@@ -74,8 +74,9 @@ class ReplicationQuotaManager(val config: ReplicationQuotaManagerConfig,
   private val lock = new ReentrantReadWriteLock()
   private val throttledPartitions = new ConcurrentHashMap[String, Seq[Int]]()
   private var quota: Quota = null
-  private val sensorAccess = new SensorAccess
+  private val sensorAccess = new SensorAccess(lock, metrics)
-  private val rateMetricName = metrics.metricName("byte-rate", replicationType.toString, s"Tracking byte-rate for ${replicationType}")
+  private val rateMetricName = metrics.metricName("byte-rate", replicationType.toString,
+    s"Tracking byte-rate for ${replicationType}")
 
   /**
     * Update the quota
@@ -194,11 +195,9 @@ class ReplicationQuotaManager(val config: ReplicationQuotaManagerConfig,
     sensorAccess.getOrCreate(
       replicationType.toString,
       InactiveSensorExpirationTimeSeconds,
-      lock,
+      rateMetricName,
-      metrics,
+      Some(getQuotaMetricConfig(quota)),
-      () => rateMetricName,
+      new SimpleRate
-      () => getQuotaMetricConfig(quota),
-      () => new SimpleRate()
     )
   }
 }

core/src/main/scala/kafka/server/SensorAccess.scala

@@ -16,10 +16,10 @@
   */
 package kafka.server
 
-import java.util.concurrent.locks.ReentrantReadWriteLock
+import java.util.concurrent.locks.ReadWriteLock
 
 import org.apache.kafka.common.MetricName
-import org.apache.kafka.common.metrics.{Metrics, Sensor, MeasurableStat, MetricConfig}
+import org.apache.kafka.common.metrics.{MeasurableStat, MetricConfig, Metrics, Sensor}
 
 /**
   * Class which centralises the logic for creating/accessing sensors.
@@ -27,9 +27,10 @@ import org.apache.kafka.common.metrics.{Metrics, Sensor, MeasurableStat, MetricC
   *
   * The later arguments are passed as methods as they are only called when the sensor is instantiated.
   */
-class SensorAccess {
+class SensorAccess(lock: ReadWriteLock, metrics: Metrics) {
 
-  def getOrCreate(sensorName: String, expirationTime: Long, lock: ReentrantReadWriteLock, metrics: Metrics, metricName: () => MetricName, config: () => MetricConfig, measure: () => MeasurableStat): Sensor = {
+  def getOrCreate(sensorName: String, expirationTime: Long,
+                  metricName: => MetricName, config: => Option[MetricConfig], measure: => MeasurableStat): Sensor = {
     var sensor: Sensor = null
 
     /* Acquire the read lock to fetch the sensor. It is safe to call getSensor from multiple threads.
@@ -41,12 +42,8 @@ class SensorAccess {
      * at which point it is safe to read
      */
     lock.readLock().lock()
-    try {
+    try sensor = metrics.getSensor(sensorName)
-      sensor = metrics.getSensor(sensorName)
+    finally lock.readLock().unlock()
-    }
-    finally {
-      lock.readLock().unlock()
-    }
 
     /* If the sensor is null, try to create it else return the existing sensor
      * The sensor can be null, hence the null checks
@@ -64,8 +61,8 @@ class SensorAccess {
         // ensure that we initialise `ClientSensors` with non-null parameters.
         sensor = metrics.getSensor(sensorName)
         if (sensor == null) {
-          sensor = metrics.sensor(sensorName, config(), expirationTime)
+          sensor = metrics.sensor(sensorName, config.orNull, expirationTime)
-          sensor.add(metricName(), measure())
+          sensor.add(metricName, measure)
         }
       } finally {
         lock.writeLock().unlock()