From 8f4f0e492d2ec9bda5b9b613e649ac8d8bcfb93d Mon Sep 17 00:00:00 2001 From: Anton Karamanov Date: Mon, 1 Sep 2014 18:41:48 +0400 Subject: [PATCH] KAFKA-1620; Make kafka api protocol implementation public --- core/src/main/scala/kafka/api/FetchRequest.scala | 14 +++++++------- .../src/main/scala/kafka/api/GenericRequestAndHeader.scala | 14 +++++++------- .../main/scala/kafka/api/GenericResponseAndHeader.scala | 10 +++++----- core/src/main/scala/kafka/api/RequestOrResponse.scala | 2 +- core/src/main/scala/kafka/cluster/Broker.scala | 4 ++-- 5 files changed, 22 insertions(+), 22 deletions(-) diff --git a/core/src/main/scala/kafka/api/FetchRequest.scala b/core/src/main/scala/kafka/api/FetchRequest.scala index 51cdccf..59c0915 100644 --- a/core/src/main/scala/kafka/api/FetchRequest.scala +++ b/core/src/main/scala/kafka/api/FetchRequest.scala @@ -59,13 +59,13 @@ object FetchRequest { } } -case class FetchRequest private[kafka] (versionId: Short = FetchRequest.CurrentVersion, - correlationId: Int = FetchRequest.DefaultCorrelationId, - clientId: String = ConsumerConfig.DefaultClientId, - replicaId: Int = Request.OrdinaryConsumerId, - maxWait: Int = FetchRequest.DefaultMaxWait, - minBytes: Int = FetchRequest.DefaultMinBytes, - requestInfo: Map[TopicAndPartition, PartitionFetchInfo]) +case class FetchRequest(versionId: Short = FetchRequest.CurrentVersion, + correlationId: Int = FetchRequest.DefaultCorrelationId, + clientId: String = ConsumerConfig.DefaultClientId, + replicaId: Int = Request.OrdinaryConsumerId, + maxWait: Int = FetchRequest.DefaultMaxWait, + minBytes: Int = FetchRequest.DefaultMinBytes, + requestInfo: Map[TopicAndPartition, PartitionFetchInfo]) extends RequestOrResponse(Some(RequestKeys.FetchKey)) { /** diff --git a/core/src/main/scala/kafka/api/GenericRequestAndHeader.scala b/core/src/main/scala/kafka/api/GenericRequestAndHeader.scala index f40e19f..f3f6f7e 100644 --- a/core/src/main/scala/kafka/api/GenericRequestAndHeader.scala +++ b/core/src/main/scala/kafka/api/GenericRequestAndHeader.scala @@ -17,12 +17,12 @@ import java.nio.ByteBuffer import org.apache.kafka.common.requests.AbstractRequestResponse import kafka.api.ApiUtils._ -private[kafka] abstract class GenericRequestAndHeader(val versionId: Short, - val correlationId: Int, - val clientId: String, - val body: AbstractRequestResponse, - val name: String, - override val requestId: Option[Short] = None) +abstract class GenericRequestAndHeader(val versionId: Short, + val correlationId: Int, + val clientId: String, + val body: AbstractRequestResponse, + val name: String, + override val requestId: Option[Short] = None) extends RequestOrResponse(requestId) { def writeTo(buffer: ByteBuffer) { @@ -52,4 +52,4 @@ private[kafka] abstract class GenericRequestAndHeader(val versionId: Short, strBuffer.append("; Body: " + body.toString) strBuffer.toString() } -} \ No newline at end of file +} diff --git a/core/src/main/scala/kafka/api/GenericResponseAndHeader.scala b/core/src/main/scala/kafka/api/GenericResponseAndHeader.scala index a4879e2..ac33687 100644 --- a/core/src/main/scala/kafka/api/GenericResponseAndHeader.scala +++ b/core/src/main/scala/kafka/api/GenericResponseAndHeader.scala @@ -16,10 +16,10 @@ package kafka.api import java.nio.ByteBuffer import org.apache.kafka.common.requests.AbstractRequestResponse -private[kafka] abstract class GenericResponseAndHeader(val correlationId: Int, - val body: AbstractRequestResponse, - val name: String, - override val requestId: Option[Short] = None) +abstract class GenericResponseAndHeader(val correlationId: Int, + val body: AbstractRequestResponse, + val name: String, + override val requestId: Option[Short] = None) extends RequestOrResponse(requestId) { def writeTo(buffer: ByteBuffer) { @@ -43,4 +43,4 @@ private[kafka] abstract class GenericResponseAndHeader(val correlationId: Int, strBuffer.append("; Body: " + body.toString) strBuffer.toString() } -} \ No newline at end of file +} diff --git a/core/src/main/scala/kafka/api/RequestOrResponse.scala b/core/src/main/scala/kafka/api/RequestOrResponse.scala index 0334343..73ec1d9 100644 --- a/core/src/main/scala/kafka/api/RequestOrResponse.scala +++ b/core/src/main/scala/kafka/api/RequestOrResponse.scala @@ -30,7 +30,7 @@ object Request { } -private[kafka] abstract class RequestOrResponse(val requestId: Option[Short] = None) extends Logging { +abstract class RequestOrResponse(val requestId: Option[Short] = None) extends Logging { def sizeInBytes: Int diff --git a/core/src/main/scala/kafka/cluster/Broker.scala b/core/src/main/scala/kafka/cluster/Broker.scala index 9407ed2..ccc3fc1 100644 --- a/core/src/main/scala/kafka/cluster/Broker.scala +++ b/core/src/main/scala/kafka/cluster/Broker.scala @@ -26,7 +26,7 @@ import kafka.common.{KafkaException, BrokerNotAvailableException} /** * A Kafka broker */ -private[kafka] object Broker { +object Broker { def createBroker(id: Int, brokerInfoString: String): Broker = { if(brokerInfoString == null) @@ -54,7 +54,7 @@ private[kafka] object Broker { } } -private[kafka] case class Broker(val id: Int, val host: String, val port: Int) { +case class Broker(val id: Int, val host: String, val port: Int) { override def toString(): String = new String("id:" + id + ",host:" + host + ",port:" + port) -- 1.8.3.2