package kafka.api
import java.nio.ByteBuffer
import kafka.network.{BoundedByteBufferSend, RequestChannel}
import kafka.network.RequestChannel.Response
import kafka.common.ErrorMapping
object ConsumerMetadataRequest {
val CurrentVersion = 0.shortValue
val DefaultClientId = ""
def readFrom(buffer: ByteBuffer) = {
val versionId = buffer.getShort
val correlationId = buffer.getInt
val clientId = ApiUtils.readShortString(buffer)
val group = ApiUtils.readShortString(buffer)
ConsumerMetadataRequest(group, versionId, correlationId, clientId)
}
}
case class ConsumerMetadataRequest(group: String,
versionId: Short = ConsumerMetadataRequest.CurrentVersion,
correlationId: Int = 0,
clientId: String = ConsumerMetadataRequest.DefaultClientId)
extends RequestOrResponse(Some(RequestKeys.ConsumerMetadataKey)) {
def sizeInBytes =
2 +
4 +
ApiUtils.shortStringLength(clientId) +
ApiUtils.shortStringLength(group)
def writeTo(buffer: ByteBuffer) {
buffer.putShort(versionId)
buffer.putInt(correlationId)
ApiUtils.writeShortString(buffer, clientId)
ApiUtils.writeShortString(buffer, group)
}
override def handleError(e: Throwable, requestChannel: RequestChannel, request: RequestChannel.Request): Unit = {
val errorResponse = ConsumerMetadataResponse(None, ErrorMapping.ConsumerCoordinatorNotAvailableCode)
requestChannel.sendResponse(new Response(request, new BoundedByteBufferSend(errorResponse)))
}
def describe(details: Boolean) = {
val consumerMetadataRequest = new StringBuilder
consumerMetadataRequest.append("Name: " + this.getClass.getSimpleName)
consumerMetadataRequest.append("; Version: " + versionId)
consumerMetadataRequest.append("; CorrelationId: " + correlationId)
consumerMetadataRequest.append("; ClientId: " + clientId)
consumerMetadataRequest.append("; Group: " + group)
consumerMetadataRequest.toString()
}
}