package kafka.consumer
import java.util.concurrent.TimeUnit
import kafka.common.{ClientIdAllBrokers, ClientIdBroker, ClientIdAndBroker}
import kafka.metrics.{KafkaMetricsGroup, KafkaTimer}
import kafka.utils.Pool
class FetchRequestAndResponseMetrics(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("FetchRequestRateAndTimeMs", TimeUnit.MILLISECONDS, TimeUnit.SECONDS, tags))
val requestSizeHist = newHistogram("FetchResponseSize", biased = true, tags)
}
class FetchRequestAndResponseStats(clientId: String) {
private val valueFactory = (k: ClientIdBroker) => new FetchRequestAndResponseMetrics(k)
private val stats = new Pool[ClientIdBroker, FetchRequestAndResponseMetrics](Some(valueFactory))
private val allBrokersStats = new FetchRequestAndResponseMetrics(new ClientIdAllBrokers(clientId))
def getFetchRequestAndResponseAllBrokersStats(): FetchRequestAndResponseMetrics = allBrokersStats
def getFetchRequestAndResponseStats(brokerHost: String, brokerPort: Int): FetchRequestAndResponseMetrics = {
stats.getAndMaybePut(new ClientIdAndBroker(clientId, brokerHost, brokerPort))
}
}
object FetchRequestAndResponseStatsRegistry {
private val valueFactory = (k: String) => new FetchRequestAndResponseStats(k)
private val globalStats = new Pool[String, FetchRequestAndResponseStats](Some(valueFactory))
def getFetchRequestAndResponseStats(clientId: String) = {
globalStats.getAndMaybePut(clientId)
}
def removeConsumerFetchRequestAndResponseStats(clientId: String) {
val pattern = (".*" + clientId + ".*").r
val keys = globalStats.keys
for (key <- keys) {
pattern.findFirstIn(key) match {
case Some(_) => globalStats.remove(key)
case _ =>
}
}
}
}