package kafka.server
import kafka.common.TopicAndPartition
import kafka.metrics.KafkaMetricsGroup
import kafka.utils.Pool
import kafka.network.{BoundedByteBufferSend, RequestChannel}
import java.util.concurrent.TimeUnit
class ProducerRequestPurgatory(replicaManager: ReplicaManager, offsetManager: OffsetManager, requestChannel: RequestChannel)
extends RequestPurgatory[DelayedProduce](replicaManager.config.brokerId, replicaManager.config.producerPurgatoryPurgeIntervalRequests) {
this.logIdent = "[ProducerRequestPurgatory-%d] ".format(replicaManager.config.brokerId)
private class DelayedProducerRequestMetrics(metricId: Option[TopicAndPartition]) extends KafkaMetricsGroup {
val tags: scala.collection.Map[String, String] = metricId match {
case Some(topicAndPartition) => Map("topic" -> topicAndPartition.topic, "partition" -> topicAndPartition.partition.toString)
case None => Map.empty
}
val expiredRequestMeter = newMeter("ExpiresPerSecond", "requests", TimeUnit.SECONDS, tags)
}
private val producerRequestMetricsForKey = {
val valueFactory = (k: TopicAndPartition) => new DelayedProducerRequestMetrics(Some(k))
new Pool[TopicAndPartition, DelayedProducerRequestMetrics](Some(valueFactory))
}
private val aggregateProduceRequestMetrics = new DelayedProducerRequestMetrics(None)
private def recordDelayedProducerKeyExpired(metricId: TopicAndPartition) {
val keyMetrics = producerRequestMetricsForKey.getAndMaybePut(metricId)
List(keyMetrics, aggregateProduceRequestMetrics).foreach(_.expiredRequestMeter.mark())
}
def checkSatisfied(delayedProduce: DelayedProduce) = delayedProduce.isSatisfied(replicaManager)
def expire(delayedProduce: DelayedProduce) {
debug("Expiring produce request %s.".format(delayedProduce.produce))
for ((topicPartition, responseStatus) <- delayedProduce.partitionStatus if responseStatus.acksPending)
recordDelayedProducerKeyExpired(topicPartition)
respond(delayedProduce)
}
def respond(delayedProduce: DelayedProduce) {
val response = delayedProduce.respond(offsetManager)
requestChannel.sendResponse(new RequestChannel.Response(delayedProduce.request, new BoundedByteBufferSend(response)))
}
}