package kafka.api
import java.nio.ByteBuffer
import kafka.api.ApiUtils._
import collection.mutable.ListBuffer
import kafka.network.{BoundedByteBufferSend, RequestChannel}
import kafka.common.{TopicAndPartition, ErrorMapping}
import kafka.network.RequestChannel.Response
import kafka.utils.Logging
object ControlledShutdownRequest extends Logging {
val CurrentVersion = 0.shortValue
val DefaultClientId = ""
def readFrom(buffer: ByteBuffer): ControlledShutdownRequest = {
val versionId = buffer.getShort
val correlationId = buffer.getInt
val brokerId = buffer.getInt
new ControlledShutdownRequest(versionId, correlationId, brokerId)
}
}
case class ControlledShutdownRequest(val versionId: Short,
val correlationId: Int,
val brokerId: Int)
extends RequestOrResponse(Some(RequestKeys.ControlledShutdownKey)){
def this(correlationId: Int, brokerId: Int) =
this(ControlledShutdownRequest.CurrentVersion, correlationId, brokerId)
def writeTo(buffer: ByteBuffer) {
buffer.putShort(versionId)
buffer.putInt(correlationId)
buffer.putInt(brokerId)
}
def sizeInBytes(): Int = {
2 +
4 +
4
}
override def toString(): String = {
describe(true)
}
override def handleError(e: Throwable, requestChannel: RequestChannel, request: RequestChannel.Request): Unit = {
val errorResponse = ControlledShutdownResponse(correlationId, ErrorMapping.codeFor(e.getCause.asInstanceOf[Class[Throwable]]), Set.empty[TopicAndPartition])
requestChannel.sendResponse(new Response(request, new BoundedByteBufferSend(errorResponse)))
}
override def describe(details: Boolean = false): String = {
val controlledShutdownRequest = new StringBuilder
controlledShutdownRequest.append("Name: " + this.getClass.getSimpleName)
controlledShutdownRequest.append("; Version: " + versionId)
controlledShutdownRequest.append("; CorrelationId: " + correlationId)
controlledShutdownRequest.append("; BrokerId: " + brokerId)
controlledShutdownRequest.toString()
}
}