package kafka.producer
import kafka.metrics.{KafkaTimer, KafkaMetricsGroup}
import java.util.concurrent.TimeUnit
import kafka.utils.Pool
import kafka.common.{ClientIdAllBrokers, ClientIdBroker, ClientIdAndBroker}
class ProducerRequestMetrics(metricId: ClientIdBroker) extends KafkaMetricsGroup {
val tags = metricId match {
case ClientIdAndBroker(clientId, brokerHost, brokerPort) => Map("clientId" -> clientId, "brokerHost" -> brokerHost, "brokerPort" -> brokerPort.toString)
case ClientIdAllBrokers(clientId) => Map("clientId" -> clientId)
}
val requestTimer = new KafkaTimer(newTimer("ProducerRequestRateAndTimeMs", TimeUnit.MILLISECONDS, TimeUnit.SECONDS, tags))
val requestSizeHist = newHistogram("ProducerRequestSize", biased = true, tags)
}
class ProducerRequestStats(clientId: String) {
private val valueFactory = (k: ClientIdBroker) => new ProducerRequestMetrics(k)
private val stats = new Pool[ClientIdBroker, ProducerRequestMetrics](Some(valueFactory))
private val = new ProducerRequestMetrics(new ClientIdAllBrokers(clientId))
def (): ProducerRequestMetrics = allBrokersStats
def getProducerRequestStats(brokerHost: String, brokerPort: Int): ProducerRequestMetrics = {
stats.getAndMaybePut(new ClientIdAndBroker(clientId, brokerHost, brokerPort))
}
}
object ProducerRequestStatsRegistry {
private val valueFactory = (k: String) => new ProducerRequestStats(k)
private val globalStats = new Pool[String, ProducerRequestStats](Some(valueFactory))
def getProducerRequestStats(clientId: String) = {
globalStats.getAndMaybePut(clientId)
}
def removeProducerRequestStats(clientId: String) {
globalStats.remove(clientId)
}
}