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

Code before:
          warn("Received a wakeup request outside of typical usage.")

        case null => // continue
      }
    }
    completeShutdown()
  }

  private def completeShutdown(): Unit = {
    requestLocal.close()
    threadRequestChannel.remove()
    shutdownComplete.countDown()
  }

  def stop(): Unit = {
    stopped = true
  }

  def initiateShutdown(): Unit = requestChannel.sendShutdownRequest()

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

}

object KafkaRequestHandlerPool {
  val sharedAggregateTotalThreads = new AtomicInteger(0)
}

class KafkaRequestHandlerPool(
  val brokerId: Int,
  val requestChannel: RequestChannel,
  val apis: ApiRequestHandler,
  time: Time,
  numThreads: Int,
  requestHandlerAvgIdleMetricName: String,
  nodeName: String = "broker"
) extends Logging {
  private val metricsGroup = new KafkaMetricsGroup(this.getClass)

  val threadPoolSize: AtomicInteger = new AtomicInteger(numThreads)
  /* Per-pool idle meter (broker-only or controller-only) */
  private val perPoolIdleMeterName = if (nodeName == "controller") "ControllerRequestHandlerAvgIdlePercent" else "BrokerRequestHandlerAvgIdlePercent"
  private val perPoolIdleMeter = metricsGroup.newMeter(perPoolIdleMeterName, "percent", TimeUnit.NANOSECONDS)
  /* Aggregate meter to track the average free capacity of the request handlers */
  private val aggregateIdleMeter = metricsGroup.newMeter(requestHandlerAvgIdleMetricName, "percent", TimeUnit.NANOSECONDS)

  this.logIdent = s"[data-plane Kafka Request Handler on ${nodeName.capitalize} $brokerId] "
  val runnables = new mutable.ArrayBuffer[KafkaRequestHandler](numThreads)
  // when using shared aggregate counter, register this pool's threads
  sharedAggregateTotalThreads.addAndGet(numThreads)
  for (i <- 0 until numThreads) {

Review comment (suggestion):
`sharedAggregateTotalThreads` is redundant. We can just name this `totalThreads` or `aggregateThreads`.

Code after:
          warn("Received a wakeup request outside of typical usage.")

        case null => // continue
      }
    }
    completeShutdown()
  }

  private def completeShutdown(): Unit = {
    requestLocal.close()
    threadRequestChannel.remove()
    shutdownComplete.countDown()
  }

  def stop(): Unit = {
    stopped = true
  }

  def initiateShutdown(): Unit = requestChannel.sendShutdownRequest()

  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,