package kafka.api
import java.nio.ByteBuffer
import kafka.common.{TopicAndPartition, ErrorMapping}
import kafka.api.ApiUtils._
import collection.Set
object ControlledShutdownResponse {
def readFrom(buffer: ByteBuffer): ControlledShutdownResponse = {
val correlationId = buffer.getInt
val errorCode = buffer.getShort
val numEntries = buffer.getInt
var partitionsRemaining = Set[TopicAndPartition]()
for (i<- 0 until numEntries){
val topic = readShortString(buffer)
val partition = buffer.getInt
partitionsRemaining += new TopicAndPartition(topic, partition)
}
new ControlledShutdownResponse(correlationId, errorCode, partitionsRemaining)
}
}
case class ControlledShutdownResponse(val correlationId: Int,
val errorCode: Short = ErrorMapping.NoError,
val partitionsRemaining: Set[TopicAndPartition])
extends RequestOrResponse() {
def sizeInBytes(): Int ={
var size =
4 +
2 +
4
for (topicAndPartition <- partitionsRemaining) {
size +=
2 + topicAndPartition.topic.length +
4
}
size
}
def writeTo(buffer: ByteBuffer) {
buffer.putInt(correlationId)
buffer.putShort(errorCode)
buffer.putInt(partitionsRemaining.size)
for (topicAndPartition:TopicAndPartition <- partitionsRemaining){
writeShortString(buffer, topicAndPartition.topic)
buffer.putInt(topicAndPartition.partition)
}
}
override def describe(details: Boolean):String = { toString }
}