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

Code before:

  def awaitShutdown(): Unit = shutdownComplete.await()

}

/**
 * Factory for creating KafkaRequestHandlerPool instances with shared aggregate metrics.
 * All pools created by the same factory share the same aggregateThreads counter.
 */
class KafkaRequestHandlerPoolFactory {
  private[this] val aggregateThreads = new AtomicInteger(0)
  private[this] val RequestHandlerAvgIdleMetricName = "RequestHandlerAvgIdlePercent"
  
  def createPool(
    brokerId: Int,
    requestChannel: RequestChannel,
    apis: ApiRequestHandler,
    time: Time,
    numThreads: Int,
    nodeName: String
  ): KafkaRequestHandlerPool = {
    new KafkaRequestHandlerPool(aggregateThreads, RequestHandlerAvgIdleMetricName, brokerId, requestChannel, apis, time, numThreads, nodeName)
  }

  // Only used for test purpose
  def getAggregateThreads: AtomicInteger = aggregateThreads
}

class KafkaRequestHandlerPool(
  val aggregateThreads: AtomicInteger,
  val requestHandlerAvgIdleMetricName: String,
  val brokerId: Int,
  val requestChannel: RequestChannel,
  val apis: ApiRequestHandler,
  time: Time,
  numThreads: Int,
  nodeName: String
) extends Logging {
  // Changing the package or class name may cause incompatibility with existing code and metrics configuration
  private val metricsPackage = "kafka.server"
  private val metricsClassName = "KafkaRequestHandlerPool"
  private val metricsGroup = new KafkaMetricsGroup(metricsPackage, metricsClassName)

  val threadPoolSize: AtomicInteger = new AtomicInteger(numThreads)
  private val perPoolIdleMeterName = if (nodeName == "broker") {
    "BrokerRequestHandlerAvgIdlePercent"
  } else if (nodeName == "controller") {
    "ControllerRequestHandlerAvgIdlePercent"
  } else {
    throw new IllegalArgumentException("Invalid node name:" + nodeName)
  }

Review comment (suggestion):
In Scala read methods are not prefixed with the string "get". E.g. `aggregateThreads`.

You don't want public users of this type to be able to modify the atomic integer. You can protect this by exposing this property instead: `def aggregateThreadCount: Int = aggregateThread.get()`.

Code after:

  def awaitShutdown(): Unit = shutdownComplete.await()

}

/**
 * Factory for creating KafkaRequestHandlerPool instances with shared aggregate metrics.
 * All pools created by the same factory share the same aggregateThreads counter.
 */
class KafkaRequestHandlerPoolFactory {
  private[this] val aggregateThreads = new AtomicInteger(0)
  private[this] val RequestHandlerAvgIdleMetricName = "RequestHandlerAvgIdlePercent"
  
  def createPool(
    brokerId: Int,
    requestChannel: RequestChannel,
    apis: ApiRequestHandler,
    time: Time,
    numThreads: Int,
    nodeName: String
  ): KafkaRequestHandlerPool = {
    new KafkaRequestHandlerPool(aggregateThreads, RequestHandlerAvgIdleMetricName, brokerId, requestChannel, apis, time, numThreads, nodeName)
  }

  // Only used for test purpose
  def aggregateThreadCount: Int = aggregateThreads.get()
}

class KafkaRequestHandlerPool(
  val aggregateThreads: AtomicInteger,
  val requestHandlerAvgIdleMetricName: String,
  val brokerId: Int,
  val requestChannel: RequestChannel,
  val apis: ApiRequestHandler,
  time: Time,
  numThreads: Int,
  nodeName: String
) extends Logging {
  // Changing the package or class name may cause incompatibility with existing code and metrics configuration
  private val metricsPackage = "kafka.server"
  private val metricsClassName = "KafkaRequestHandlerPool"
  private val metricsGroup = new KafkaMetricsGroup(metricsPackage, metricsClassName)

  val threadPoolSize: AtomicInteger = new AtomicInteger(numThreads)
  private val perPoolIdleMeterName = if (nodeName == "broker") {
    "BrokerRequestHandlerAvgIdlePercent"
  } else if (nodeName == "controller") {
    "ControllerRequestHandlerAvgIdlePercent"
  } else {
    throw new IllegalArgumentException("Invalid node name:" + nodeName)
  }