package kafka.api
import java.nio.ByteBuffer
import kafka.api.ApiUtils._
import kafka.utils.{SystemTime, Logging}
import kafka.network.{RequestChannel, BoundedByteBufferSend}
import kafka.common.{OffsetAndMetadata, ErrorMapping, TopicAndPartition}
import kafka.network.RequestChannel.Response
import scala.collection._
object OffsetCommitRequest extends Logging {
val CurrentVersion: Short = 1
val DefaultClientId = ""
def readFrom(buffer: ByteBuffer): OffsetCommitRequest = {
val versionId = buffer.getShort
assert(versionId == 0 || versionId == 1,
"Version " + versionId + " is invalid for OffsetCommitRequest. Valid versions are 0 or 1.")
val correlationId = buffer.getInt
val clientId = readShortString(buffer)
val consumerGroupId = readShortString(buffer)
var groupGenerationId: Int = org.apache.kafka.common.requests.OffsetCommitRequest.DEFAULT_GENERATION_ID
var consumerId: String = org.apache.kafka.common.requests.OffsetCommitRequest.DEFAULT_CONSUMER_ID
if (versionId == 1) {
groupGenerationId = buffer.getInt
consumerId = readShortString(buffer)
}
val topicCount = buffer.getInt
val pairs = (1 to topicCount).flatMap(_ => {
val topic = readShortString(buffer)
val partitionCount = buffer.getInt
(1 to partitionCount).map(_ => {
val partitionId = buffer.getInt
val offset = buffer.getLong
val timestamp = {
if (versionId == 1) {
val given = buffer.getLong
given
} else
OffsetAndMetadata.InvalidTime
}
val metadata = readShortString(buffer)
(TopicAndPartition(topic, partitionId), OffsetAndMetadata(offset, metadata, timestamp))
})
})
OffsetCommitRequest(consumerGroupId, immutable.Map(pairs:_*), versionId, correlationId, clientId, groupGenerationId, consumerId)
}
def changeInvalidTimeToCurrentTime(offsetCommitRequest: OffsetCommitRequest) {
val now = SystemTime.milliseconds
for ( (topicAndPartiiton, offsetAndMetadata) <- offsetCommitRequest.requestInfo)
if (offsetAndMetadata.timestamp == OffsetAndMetadata.InvalidTime)
offsetAndMetadata.timestamp = now
}
}
case class OffsetCommitRequest(groupId: String,
requestInfo: immutable.Map[TopicAndPartition, OffsetAndMetadata],
versionId: Short = OffsetCommitRequest.CurrentVersion,
correlationId: Int = 0,
clientId: String = OffsetCommitRequest.DefaultClientId,
groupGenerationId: Int = org.apache.kafka.common.requests.OffsetCommitRequest.DEFAULT_GENERATION_ID,
consumerId: String = org.apache.kafka.common.requests.OffsetCommitRequest.DEFAULT_CONSUMER_ID)
extends RequestOrResponse(Some(RequestKeys.OffsetCommitKey)) {
assert(versionId == 0 || versionId == 1,
"Version " + versionId + " is invalid for OffsetCommitRequest. Valid versions are 0 or 1.")
lazy val requestInfoGroupedByTopic = requestInfo.groupBy(_._1.topic)
def filterLargeMetadata(maxMetadataSize: Int) =
requestInfo.filter(info => info._2.metadata == null || info._2.metadata.length <= maxMetadataSize)
def responseFor(errorCode: Short, offsetMetadataMaxSize: Int) = {
val commitStatus = requestInfo.map {info =>
(info._1, if (info._2.metadata != null && info._2.metadata.length > offsetMetadataMaxSize)
ErrorMapping.OffsetMetadataTooLargeCode
else if (errorCode == ErrorMapping.UnknownTopicOrPartitionCode)
ErrorMapping.ConsumerCoordinatorNotAvailableCode
else if (errorCode == ErrorMapping.NotLeaderForPartitionCode)
ErrorMapping.NotCoordinatorForConsumerCode
else
errorCode)
}.toMap
OffsetCommitResponse(commitStatus, correlationId)
}
def writeTo(buffer: ByteBuffer) {
buffer.putShort(versionId)
buffer.putInt(correlationId)
writeShortString(buffer, clientId)
writeShortString(buffer, groupId)
if (versionId == 1) {
buffer.putInt(groupGenerationId)
writeShortString(buffer, consumerId)
}
buffer.putInt(requestInfoGroupedByTopic.size)
requestInfoGroupedByTopic.foreach( t1 => {
writeShortString(buffer, t1._1)
buffer.putInt(t1._2.size)
t1._2.foreach( t2 => {
buffer.putInt(t2._1.partition)
buffer.putLong(t2._2.offset)
if (versionId == 1)
buffer.putLong(t2._2.timestamp)
writeShortString(buffer, t2._2.metadata)
})
})
}
override def sizeInBytes =
2 +
4 +
shortStringLength(clientId) +
shortStringLength(groupId) +
(if (versionId == 1) 4 + shortStringLength(consumerId) else 0) +
4 +
requestInfoGroupedByTopic.foldLeft(0)((count, topicAndOffsets) => {
val (topic, offsets) = topicAndOffsets
count +
shortStringLength(topic) +
4 +
offsets.foldLeft(0)((innerCount, offsetAndMetadata) => {
innerCount +
4 +
8 +
(if (versionId == 1) 8 else 0 ) +
shortStringLength(offsetAndMetadata._2.metadata)
})
})
override def handleError(e: Throwable, requestChannel: RequestChannel, request: RequestChannel.Request): Unit = {
val errorCode = ErrorMapping.codeFor(e.getClass.asInstanceOf[Class[Throwable]])
val errorResponse = responseFor(errorCode, Int.MaxValue)
requestChannel.sendResponse(new Response(request, new BoundedByteBufferSend(errorResponse)))
}
override def describe(details: Boolean): String = {
val offsetCommitRequest = new StringBuilder
offsetCommitRequest.append("Name: " + this.getClass.getSimpleName)
offsetCommitRequest.append("; Version: " + versionId)
offsetCommitRequest.append("; CorrelationId: " + correlationId)
offsetCommitRequest.append("; ClientId: " + clientId)
offsetCommitRequest.append("; GroupId: " + groupId)
offsetCommitRequest.append("; GroupGenerationId: " + groupGenerationId)
offsetCommitRequest.append("; ConsumerId: " + consumerId)
if(details)
offsetCommitRequest.append("; RequestInfo: " + requestInfo.mkString(","))
offsetCommitRequest.toString()
}
override def toString = {
describe(details = true)
}
}