You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by jk...@apache.org on 2012/12/07 05:01:31 UTC
svn commit: r1418181 - in /kafka/branches/0.8/core/src:
main/scala/kafka/api/ main/scala/kafka/server/ test/scala/unit/kafka/api/
test/scala/unit/kafka/producer/
Author: jkreps
Date: Fri Dec 7 04:01:29 2012
New Revision: 1418181
URL: http://svn.apache.org/viewvc?rev=1418181&view=rev
Log:
KAFKA-642 Remove the response version and baseline all versions/magic bytes etc to 0.
Modified:
kafka/branches/0.8/core/src/main/scala/kafka/api/FetchRequest.scala
kafka/branches/0.8/core/src/main/scala/kafka/api/FetchResponse.scala
kafka/branches/0.8/core/src/main/scala/kafka/api/LeaderAndIsrRequest.scala
kafka/branches/0.8/core/src/main/scala/kafka/api/LeaderAndIsrResponse.scala
kafka/branches/0.8/core/src/main/scala/kafka/api/OffsetRequest.scala
kafka/branches/0.8/core/src/main/scala/kafka/api/OffsetResponse.scala
kafka/branches/0.8/core/src/main/scala/kafka/api/ProducerRequest.scala
kafka/branches/0.8/core/src/main/scala/kafka/api/ProducerResponse.scala
kafka/branches/0.8/core/src/main/scala/kafka/api/StopReplicaRequest.scala
kafka/branches/0.8/core/src/main/scala/kafka/api/StopReplicaResponse.scala
kafka/branches/0.8/core/src/main/scala/kafka/api/TopicMetadataRequest.scala
kafka/branches/0.8/core/src/main/scala/kafka/api/TopicMetadataResponse.scala
kafka/branches/0.8/core/src/main/scala/kafka/server/KafkaApis.scala
kafka/branches/0.8/core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala
kafka/branches/0.8/core/src/test/scala/unit/kafka/producer/AsyncProducerTest.scala
Modified: kafka/branches/0.8/core/src/main/scala/kafka/api/FetchRequest.scala
URL: http://svn.apache.org/viewvc/kafka/branches/0.8/core/src/main/scala/kafka/api/FetchRequest.scala?rev=1418181&r1=1418180&r2=1418181&view=diff
==============================================================================
--- kafka/branches/0.8/core/src/main/scala/kafka/api/FetchRequest.scala (original)
+++ kafka/branches/0.8/core/src/main/scala/kafka/api/FetchRequest.scala Fri Dec 7 04:01:29 2012
@@ -30,7 +30,7 @@ case class PartitionFetchInfo(offset: Lo
object FetchRequest {
- val CurrentVersion = 1.shortValue()
+ val CurrentVersion = 0.shortValue
val DefaultMaxWait = 0
val DefaultMinBytes = 0
val ReplicaFetcherClientId = "replica-fetcher"
Modified: kafka/branches/0.8/core/src/main/scala/kafka/api/FetchResponse.scala
URL: http://svn.apache.org/viewvc/kafka/branches/0.8/core/src/main/scala/kafka/api/FetchResponse.scala?rev=1418181&r1=1418180&r2=1418181&view=diff
==============================================================================
--- kafka/branches/0.8/core/src/main/scala/kafka/api/FetchResponse.scala (original)
+++ kafka/branches/0.8/core/src/main/scala/kafka/api/FetchResponse.scala Fri Dec 7 04:01:29 2012
@@ -136,12 +136,10 @@ class TopicDataSend(val topicData: Topic
object FetchResponse {
val headerSize =
- 2 + /* versionId */
4 + /* correlationId */
4 /* topic count */
def readFrom(buffer: ByteBuffer): FetchResponse = {
- val versionId = buffer.getShort
val correlationId = buffer.getInt
val topicCount = buffer.getInt
val pairs = (1 to topicCount).flatMap(_ => {
@@ -151,13 +149,12 @@ object FetchResponse {
(TopicAndPartition(topicData.topic, partitionId), partitionData)
}
})
- FetchResponse(versionId, correlationId, Map(pairs:_*))
+ FetchResponse(correlationId, Map(pairs:_*))
}
}
-case class FetchResponse(versionId: Short,
- correlationId: Int,
+case class FetchResponse(correlationId: Int,
data: Map[TopicAndPartition, FetchResponsePartitionData]) {
/**
@@ -206,7 +203,6 @@ class FetchResponseSend(val fetchRespons
private val buffer = ByteBuffer.allocate(4 /* for size */ + FetchResponse.headerSize)
buffer.putInt(size)
- buffer.putShort(fetchResponse.versionId)
buffer.putInt(fetchResponse.correlationId)
buffer.putInt(fetchResponse.dataGroupedByTopic.size) // topic count
buffer.rewind()
Modified: kafka/branches/0.8/core/src/main/scala/kafka/api/LeaderAndIsrRequest.scala
URL: http://svn.apache.org/viewvc/kafka/branches/0.8/core/src/main/scala/kafka/api/LeaderAndIsrRequest.scala?rev=1418181&r1=1418180&r2=1418181&view=diff
==============================================================================
--- kafka/branches/0.8/core/src/main/scala/kafka/api/LeaderAndIsrRequest.scala (original)
+++ kafka/branches/0.8/core/src/main/scala/kafka/api/LeaderAndIsrRequest.scala Fri Dec 7 04:01:29 2012
@@ -79,7 +79,7 @@ case class PartitionStateInfo(val leader
}
object LeaderAndIsrRequest {
- val CurrentVersion = 1.shortValue()
+ val CurrentVersion = 0.shortValue
val DefaultClientId = ""
val IsInit: Boolean = true
val NotInit: Boolean = false
Modified: kafka/branches/0.8/core/src/main/scala/kafka/api/LeaderAndIsrResponse.scala
URL: http://svn.apache.org/viewvc/kafka/branches/0.8/core/src/main/scala/kafka/api/LeaderAndIsrResponse.scala?rev=1418181&r1=1418180&r2=1418181&view=diff
==============================================================================
--- kafka/branches/0.8/core/src/main/scala/kafka/api/LeaderAndIsrResponse.scala (original)
+++ kafka/branches/0.8/core/src/main/scala/kafka/api/LeaderAndIsrResponse.scala Fri Dec 7 04:01:29 2012
@@ -26,7 +26,6 @@ import collection.Map
object LeaderAndIsrResponse {
def readFrom(buffer: ByteBuffer): LeaderAndIsrResponse = {
- val versionId = buffer.getShort
val correlationId = buffer.getInt
val errorCode = buffer.getShort
val numEntries = buffer.getInt
@@ -37,19 +36,17 @@ object LeaderAndIsrResponse {
val partitionErrorCode = buffer.getShort
responseMap.put((topic, partition), partitionErrorCode)
}
- new LeaderAndIsrResponse(versionId, correlationId, responseMap, errorCode)
+ new LeaderAndIsrResponse(correlationId, responseMap, errorCode)
}
}
-case class LeaderAndIsrResponse(versionId: Short,
- correlationId: Int,
+case class LeaderAndIsrResponse(correlationId: Int,
responseMap: Map[(String, Int), Short],
errorCode: Short = ErrorMapping.NoError)
extends RequestOrResponse {
def sizeInBytes(): Int ={
var size =
- 2 /* version id */ +
4 /* correlation id */ +
2 /* error code */ +
4 /* number of responses */
@@ -63,7 +60,6 @@ case class LeaderAndIsrResponse(versionI
}
def writeTo(buffer: ByteBuffer) {
- buffer.putShort(versionId)
buffer.putInt(correlationId)
buffer.putShort(errorCode)
buffer.putInt(responseMap.size)
Modified: kafka/branches/0.8/core/src/main/scala/kafka/api/OffsetRequest.scala
URL: http://svn.apache.org/viewvc/kafka/branches/0.8/core/src/main/scala/kafka/api/OffsetRequest.scala?rev=1418181&r1=1418180&r2=1418181&view=diff
==============================================================================
--- kafka/branches/0.8/core/src/main/scala/kafka/api/OffsetRequest.scala (original)
+++ kafka/branches/0.8/core/src/main/scala/kafka/api/OffsetRequest.scala Fri Dec 7 04:01:29 2012
@@ -23,7 +23,7 @@ import kafka.api.ApiUtils._
object OffsetRequest {
- val CurrentVersion = 1.shortValue()
+ val CurrentVersion = 0.shortValue
val DefaultClientId = ""
val SmallestTimeString = "smallest"
Modified: kafka/branches/0.8/core/src/main/scala/kafka/api/OffsetResponse.scala
URL: http://svn.apache.org/viewvc/kafka/branches/0.8/core/src/main/scala/kafka/api/OffsetResponse.scala?rev=1418181&r1=1418180&r2=1418181&view=diff
==============================================================================
--- kafka/branches/0.8/core/src/main/scala/kafka/api/OffsetResponse.scala (original)
+++ kafka/branches/0.8/core/src/main/scala/kafka/api/OffsetResponse.scala Fri Dec 7 04:01:29 2012
@@ -25,7 +25,6 @@ import kafka.api.ApiUtils._
object OffsetResponse {
def readFrom(buffer: ByteBuffer): OffsetResponse = {
- val versionId = buffer.getShort
val correlationId = buffer.getInt
val numTopics = buffer.getInt
val pairs = (1 to numTopics).flatMap(_ => {
@@ -39,7 +38,7 @@ object OffsetResponse {
(TopicAndPartition(topic, partition), PartitionOffsetsResponse(error, offsets))
})
})
- OffsetResponse(versionId, correlationId, Map(pairs:_*))
+ OffsetResponse(correlationId, Map(pairs:_*))
}
}
@@ -48,8 +47,7 @@ object OffsetResponse {
case class PartitionOffsetsResponse(error: Short, offsets: Seq[Long])
-case class OffsetResponse(versionId: Short,
- correlationId: Int,
+case class OffsetResponse(correlationId: Int,
partitionErrorAndOffsets: Map[TopicAndPartition, PartitionOffsetsResponse])
extends RequestOrResponse {
@@ -58,7 +56,6 @@ case class OffsetResponse(versionId: Sho
def hasError = partitionErrorAndOffsets.values.exists(_.error != ErrorMapping.NoError)
val sizeInBytes = {
- 2 + /* versionId */
4 + /* correlation id */
4 + /* topic count */
offsetsGroupedByTopic.foldLeft(0)((foldedTopics, currTopic) => {
@@ -77,7 +74,6 @@ case class OffsetResponse(versionId: Sho
}
def writeTo(buffer: ByteBuffer) {
- buffer.putShort(versionId)
buffer.putInt(correlationId)
buffer.putInt(offsetsGroupedByTopic.size) // topic count
offsetsGroupedByTopic.foreach {
Modified: kafka/branches/0.8/core/src/main/scala/kafka/api/ProducerRequest.scala
URL: http://svn.apache.org/viewvc/kafka/branches/0.8/core/src/main/scala/kafka/api/ProducerRequest.scala?rev=1418181&r1=1418180&r2=1418181&view=diff
==============================================================================
--- kafka/branches/0.8/core/src/main/scala/kafka/api/ProducerRequest.scala (original)
+++ kafka/branches/0.8/core/src/main/scala/kafka/api/ProducerRequest.scala Fri Dec 7 04:01:29 2012
@@ -25,7 +25,7 @@ import kafka.api.ApiUtils._
object ProducerRequest {
- val CurrentVersion: Short = 0
+ val CurrentVersion = 0.shortValue
def readFrom(buffer: ByteBuffer): ProducerRequest = {
val versionId: Short = buffer.getShort
Modified: kafka/branches/0.8/core/src/main/scala/kafka/api/ProducerResponse.scala
URL: http://svn.apache.org/viewvc/kafka/branches/0.8/core/src/main/scala/kafka/api/ProducerResponse.scala?rev=1418181&r1=1418180&r2=1418181&view=diff
==============================================================================
--- kafka/branches/0.8/core/src/main/scala/kafka/api/ProducerResponse.scala (original)
+++ kafka/branches/0.8/core/src/main/scala/kafka/api/ProducerResponse.scala Fri Dec 7 04:01:29 2012
@@ -25,7 +25,6 @@ import kafka.api.ApiUtils._
object ProducerResponse {
def readFrom(buffer: ByteBuffer): ProducerResponse = {
- val versionId = buffer.getShort
val correlationId = buffer.getInt
val topicCount = buffer.getInt
val statusPairs = (1 to topicCount).flatMap(_ => {
@@ -39,15 +38,14 @@ object ProducerResponse {
})
})
- ProducerResponse(versionId, correlationId, Map(statusPairs:_*))
+ ProducerResponse(correlationId, Map(statusPairs:_*))
}
}
case class ProducerResponseStatus(error: Short, offset: Long)
-case class ProducerResponse(versionId: Short,
- correlationId: Int,
+case class ProducerResponse(correlationId: Int,
status: Map[TopicAndPartition, ProducerResponseStatus]) extends RequestOrResponse {
/**
@@ -59,7 +57,6 @@ case class ProducerResponse(versionId: S
val sizeInBytes = {
val groupedStatus = statusGroupedByTopic
- 2 + /* version id */
4 + /* correlation id */
4 + /* topic count */
groupedStatus.foldLeft (0) ((foldedTopics, currTopic) => {
@@ -76,8 +73,6 @@ case class ProducerResponse(versionId: S
def writeTo(buffer: ByteBuffer) {
val groupedStatus = statusGroupedByTopic
-
- buffer.putShort(versionId)
buffer.putInt(correlationId)
buffer.putInt(groupedStatus.size) // topic count
Modified: kafka/branches/0.8/core/src/main/scala/kafka/api/StopReplicaRequest.scala
URL: http://svn.apache.org/viewvc/kafka/branches/0.8/core/src/main/scala/kafka/api/StopReplicaRequest.scala?rev=1418181&r1=1418180&r2=1418181&view=diff
==============================================================================
--- kafka/branches/0.8/core/src/main/scala/kafka/api/StopReplicaRequest.scala (original)
+++ kafka/branches/0.8/core/src/main/scala/kafka/api/StopReplicaRequest.scala Fri Dec 7 04:01:29 2012
@@ -25,7 +25,7 @@ import kafka.network.InvalidRequestExcep
object StopReplicaRequest extends Logging {
- val CurrentVersion = 1.shortValue()
+ val CurrentVersion = 0.shortValue
val DefaultClientId = ""
val DefaultAckTimeout = 100
Modified: kafka/branches/0.8/core/src/main/scala/kafka/api/StopReplicaResponse.scala
URL: http://svn.apache.org/viewvc/kafka/branches/0.8/core/src/main/scala/kafka/api/StopReplicaResponse.scala?rev=1418181&r1=1418180&r2=1418181&view=diff
==============================================================================
--- kafka/branches/0.8/core/src/main/scala/kafka/api/StopReplicaResponse.scala (original)
+++ kafka/branches/0.8/core/src/main/scala/kafka/api/StopReplicaResponse.scala Fri Dec 7 04:01:29 2012
@@ -26,7 +26,6 @@ import kafka.api.ApiUtils._
object StopReplicaResponse {
def readFrom(buffer: ByteBuffer): StopReplicaResponse = {
- val versionId = buffer.getShort
val correlationId = buffer.getInt
val errorCode = buffer.getShort
val numEntries = buffer.getInt
@@ -38,18 +37,16 @@ object StopReplicaResponse {
val partitionErrorCode = buffer.getShort()
responseMap.put((topic, partition), partitionErrorCode)
}
- new StopReplicaResponse(versionId, correlationId, responseMap.toMap, errorCode)
+ new StopReplicaResponse(correlationId, responseMap.toMap, errorCode)
}
}
-case class StopReplicaResponse(val versionId: Short,
- val correlationId: Int,
+case class StopReplicaResponse(val correlationId: Int,
val responseMap: Map[(String, Int), Short],
val errorCode: Short = ErrorMapping.NoError) extends RequestOrResponse{
def sizeInBytes(): Int ={
var size =
- 2 /* version id */ +
4 /* correlation id */ +
2 /* error code */ +
4 /* number of responses */
@@ -63,7 +60,6 @@ case class StopReplicaResponse(val versi
}
def writeTo(buffer: ByteBuffer) {
- buffer.putShort(versionId)
buffer.putInt(correlationId)
buffer.putShort(errorCode)
buffer.putInt(responseMap.size)
Modified: kafka/branches/0.8/core/src/main/scala/kafka/api/TopicMetadataRequest.scala
URL: http://svn.apache.org/viewvc/kafka/branches/0.8/core/src/main/scala/kafka/api/TopicMetadataRequest.scala?rev=1418181&r1=1418180&r2=1418181&view=diff
==============================================================================
--- kafka/branches/0.8/core/src/main/scala/kafka/api/TopicMetadataRequest.scala (original)
+++ kafka/branches/0.8/core/src/main/scala/kafka/api/TopicMetadataRequest.scala Fri Dec 7 04:01:29 2012
@@ -23,7 +23,7 @@ import collection.mutable.ListBuffer
import kafka.utils.Logging
object TopicMetadataRequest extends Logging {
- val CurrentVersion = 1.shortValue()
+ val CurrentVersion = 0.shortValue
val DefaultClientId = ""
/**
Modified: kafka/branches/0.8/core/src/main/scala/kafka/api/TopicMetadataResponse.scala
URL: http://svn.apache.org/viewvc/kafka/branches/0.8/core/src/main/scala/kafka/api/TopicMetadataResponse.scala?rev=1418181&r1=1418180&r2=1418181&view=diff
==============================================================================
--- kafka/branches/0.8/core/src/main/scala/kafka/api/TopicMetadataResponse.scala (original)
+++ kafka/branches/0.8/core/src/main/scala/kafka/api/TopicMetadataResponse.scala Fri Dec 7 04:01:29 2012
@@ -23,27 +23,24 @@ import java.nio.ByteBuffer
object TopicMetadataResponse {
def readFrom(buffer: ByteBuffer): TopicMetadataResponse = {
- val versionId = buffer.getShort
val correlationId = buffer.getInt
val brokerCount = buffer.getInt
val brokers = (0 until brokerCount).map(_ => Broker.readFrom(buffer))
val brokerMap = brokers.map(b => (b.id, b)).toMap
val topicCount = buffer.getInt
val topicsMetadata = (0 until topicCount).map(_ => TopicMetadata.readFrom(buffer, brokerMap))
- new TopicMetadataResponse(versionId, topicsMetadata, correlationId)
+ new TopicMetadataResponse(topicsMetadata, correlationId)
}
}
-case class TopicMetadataResponse(versionId: Short,
- topicsMetadata: Seq[TopicMetadata],
+case class TopicMetadataResponse(topicsMetadata: Seq[TopicMetadata],
correlationId: Int) extends RequestOrResponse {
val sizeInBytes: Int = {
val brokers = extractBrokers(topicsMetadata).values
- 2 + 4 + 4 + brokers.map(_.sizeInBytes).sum + 4 + topicsMetadata.map(_.sizeInBytes).sum
+ 4 + 4 + brokers.map(_.sizeInBytes).sum + 4 + topicsMetadata.map(_.sizeInBytes).sum
}
def writeTo(buffer: ByteBuffer) {
- buffer.putShort(versionId)
buffer.putInt(correlationId)
/* brokers */
val brokers = extractBrokers(topicsMetadata).values
Modified: kafka/branches/0.8/core/src/main/scala/kafka/server/KafkaApis.scala
URL: http://svn.apache.org/viewvc/kafka/branches/0.8/core/src/main/scala/kafka/server/KafkaApis.scala?rev=1418181&r1=1418180&r2=1418181&view=diff
==============================================================================
--- kafka/branches/0.8/core/src/main/scala/kafka/server/KafkaApis.scala (original)
+++ kafka/branches/0.8/core/src/main/scala/kafka/server/KafkaApis.scala Fri Dec 7 04:01:29 2012
@@ -70,7 +70,7 @@ class KafkaApis(val requestChannel: Requ
case (topicAndPartition, data) =>
(topicAndPartition, ProducerResponseStatus(ErrorMapping.codeFor(e.getClass.asInstanceOf[Class[Throwable]]), -1l))
}
- val errorResponse = ProducerResponse(apiRequest.versionId, apiRequest.correlationId, producerResponseStatus)
+ val errorResponse = ProducerResponse(apiRequest.correlationId, producerResponseStatus)
requestChannel.sendResponse(new Response(request, new BoundedByteBufferSend(errorResponse)))
error("error when handling request %s".format(apiRequest), e)
case RequestKeys.FetchKey =>
@@ -79,7 +79,7 @@ class KafkaApis(val requestChannel: Requ
case (topicAndPartition, data) =>
(topicAndPartition, FetchResponsePartitionData(ErrorMapping.codeFor(e.getClass.asInstanceOf[Class[Throwable]]), -1, null))
}
- val errorResponse = FetchResponse(apiRequest.versionId, apiRequest.correlationId, fetchResponsePartitionData)
+ val errorResponse = FetchResponse(apiRequest.correlationId, fetchResponsePartitionData)
requestChannel.sendResponse(new RequestChannel.Response(request, new FetchResponseSend(errorResponse)))
error("error when handling request %s".format(apiRequest), e)
case RequestKeys.OffsetsKey =>
@@ -88,7 +88,7 @@ class KafkaApis(val requestChannel: Requ
case (topicAndPartition, partitionOffsetRequest) =>
(topicAndPartition, PartitionOffsetsResponse(ErrorMapping.codeFor(e.getClass.asInstanceOf[Class[Throwable]]), null))
}
- val errorResponse = OffsetResponse(apiRequest.versionId, apiRequest.correlationId, partitionOffsetResponseMap)
+ val errorResponse = OffsetResponse(apiRequest.correlationId, partitionOffsetResponseMap)
requestChannel.sendResponse(new Response(request, new BoundedByteBufferSend(errorResponse)))
error("error when handling request %s".format(apiRequest), e)
case RequestKeys.MetadataKey =>
@@ -96,7 +96,7 @@ class KafkaApis(val requestChannel: Requ
val topicMeatadata = apiRequest.topics.map {
topic => TopicMetadata(topic, Nil, ErrorMapping.codeFor(e.getClass.asInstanceOf[Class[Throwable]]))
}
- val errorResponse = TopicMetadataResponse(apiRequest.versionId, topicMeatadata, apiRequest.correlationId)
+ val errorResponse = TopicMetadataResponse(topicMeatadata, apiRequest.correlationId)
requestChannel.sendResponse(new Response(request, new BoundedByteBufferSend(errorResponse)))
error("error when handling request %s".format(apiRequest), e)
case RequestKeys.LeaderAndIsrKey =>
@@ -104,7 +104,7 @@ class KafkaApis(val requestChannel: Requ
val responseMap = apiRequest.partitionStateInfos.map {
case (topicAndPartition, partitionAndState) => (topicAndPartition, ErrorMapping.codeFor(e.getClass.asInstanceOf[Class[Throwable]]))
}
- val errorResponse = LeaderAndIsrResponse(apiRequest.versionId, apiRequest.correlationId, responseMap)
+ val errorResponse = LeaderAndIsrResponse(apiRequest.correlationId, responseMap)
requestChannel.sendResponse(new Response(request, new BoundedByteBufferSend(errorResponse)))
error("error when handling request %s".format(apiRequest), e)
case RequestKeys.StopReplicaKey =>
@@ -113,7 +113,7 @@ class KafkaApis(val requestChannel: Requ
case topicAndPartition => (topicAndPartition, ErrorMapping.codeFor(e.getClass.asInstanceOf[Class[Throwable]]))
}.toMap
error("error when handling request %s".format(apiRequest), e)
- val errorResponse = StopReplicaResponse(apiRequest.versionId, apiRequest.correlationId, responseMap)
+ val errorResponse = StopReplicaResponse(apiRequest.correlationId, responseMap)
requestChannel.sendResponse(new Response(request, new BoundedByteBufferSend(errorResponse)))
}
} finally
@@ -127,7 +127,7 @@ class KafkaApis(val requestChannel: Requ
trace("Handling leader and ISR request " + leaderAndIsrRequest)
try {
val (response, error) = replicaManager.becomeLeaderOrFollower(leaderAndIsrRequest)
- val leaderAndIsrResponse = new LeaderAndIsrResponse(leaderAndIsrRequest.versionId, leaderAndIsrRequest.correlationId, response, error)
+ val leaderAndIsrResponse = new LeaderAndIsrResponse(leaderAndIsrRequest.correlationId, response, error)
requestChannel.sendResponse(new Response(request, new BoundedByteBufferSend(leaderAndIsrResponse)))
} catch {
case e: KafkaStorageException =>
@@ -144,7 +144,7 @@ class KafkaApis(val requestChannel: Requ
trace("Handling stop replica request " + stopReplicaRequest)
val (response, error) = replicaManager.stopReplicas(stopReplicaRequest)
- val stopReplicaResponse = new StopReplicaResponse(stopReplicaRequest.versionId, stopReplicaRequest.correlationId, response.toMap, error)
+ val stopReplicaResponse = new StopReplicaResponse(stopReplicaRequest.correlationId, response.toMap, error)
requestChannel.sendResponse(new Response(request, new BoundedByteBufferSend(stopReplicaResponse)))
replicaManager.replicaFetcherManager.shutdownIdleFetcherThreads()
@@ -161,7 +161,7 @@ class KafkaApis(val requestChannel: Requ
// send any newly unblocked responses
for(fetchReq <- satisfied) {
val topicData = readMessageSets(fetchReq.fetch)
- val response = FetchResponse(FetchRequest.CurrentVersion, fetchReq.fetch.correlationId, topicData)
+ val response = FetchResponse(fetchReq.fetch.correlationId, topicData)
requestChannel.sendResponse(new RequestChannel.Response(fetchReq.request, new FetchResponseSend(response)))
}
}
@@ -192,7 +192,7 @@ class KafkaApis(val requestChannel: Requ
allPartitionHaveReplicationFactorOne ||
numPartitionsInError == produceRequest.numPartitions) {
val statuses = localProduceResults.map(r => r.key -> ProducerResponseStatus(r.errorCode, r.start)).toMap
- val response = ProducerResponse(produceRequest.versionId, produceRequest.correlationId, statuses)
+ val response = ProducerResponse(produceRequest.correlationId, statuses)
requestChannel.sendResponse(new RequestChannel.Response(request, new BoundedByteBufferSend(response)))
} else {
// create a list of (topic, partition) pairs to use as keys for this delayed request
@@ -292,7 +292,7 @@ class KafkaApis(val requestChannel: Requ
bytesReadable >= fetchRequest.minBytes ||
fetchRequest.numPartitions <= 0) {
debug("Returning fetch response %s for fetch request with correlation id %d".format(dataRead.values.map(_.error).mkString(","), fetchRequest.correlationId))
- val response = new FetchResponse(FetchRequest.CurrentVersion, fetchRequest.correlationId, dataRead)
+ val response = new FetchResponse(fetchRequest.correlationId, dataRead)
requestChannel.sendResponse(new RequestChannel.Response(request, new FetchResponseSend(response)))
} else {
debug("Putting fetch request into purgatory")
@@ -408,7 +408,7 @@ class KafkaApis(val requestChannel: Requ
(topicAndPartition, PartitionOffsetsResponse(ErrorMapping.codeFor(e.getClass.asInstanceOf[Class[Throwable]]), Nil) )
}
})
- val response = OffsetResponse(OffsetRequest.CurrentVersion, offsetRequest.correlationId, responseMap)
+ val response = OffsetResponse(offsetRequest.correlationId, responseMap)
requestChannel.sendResponse(new RequestChannel.Response(request, new BoundedByteBufferSend(response)))
}
@@ -457,7 +457,7 @@ class KafkaApis(val requestChannel: Requ
}
})
topicsMetadata.foreach(metadata => trace("Sending topic metadata " + metadata.toString))
- val response = new TopicMetadataResponse(metadataRequest.versionId, topicsMetadata.toSeq, metadataRequest.correlationId)
+ val response = new TopicMetadataResponse(topicsMetadata.toSeq, metadataRequest.correlationId)
requestChannel.sendResponse(new RequestChannel.Response(request, new BoundedByteBufferSend(response)))
}
@@ -514,7 +514,7 @@ class KafkaApis(val requestChannel: Requ
debug("Expiring fetch request %s.".format(delayed.fetch))
try {
val topicData = readMessageSets(delayed.fetch)
- val response = FetchResponse(FetchRequest.CurrentVersion, delayed.fetch.correlationId, topicData)
+ val response = FetchResponse(delayed.fetch.correlationId, topicData)
val fromFollower = delayed.fetch.isFromFollower
delayedRequestMetrics.recordDelayedFetchExpired(fromFollower)
requestChannel.sendResponse(new RequestChannel.Response(delayed.request, new FetchResponseSend(response)))
@@ -564,7 +564,7 @@ class KafkaApis(val requestChannel: Requ
(status._1, ProducerResponseStatus(pstat.error, pstat.requiredOffset))
})
- val response = ProducerResponse(produce.versionId, produce.correlationId, finalErrorsAndOffsets)
+ val response = ProducerResponse(produce.correlationId, finalErrorsAndOffsets)
requestChannel.sendResponse(new RequestChannel.Response(
request, new BoundedByteBufferSend(response)))
Modified: kafka/branches/0.8/core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala
URL: http://svn.apache.org/viewvc/kafka/branches/0.8/core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala?rev=1418181&r1=1418180&r2=1418181&view=diff
==============================================================================
--- kafka/branches/0.8/core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala (original)
+++ kafka/branches/0.8/core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala Fri Dec 7 04:01:29 2012
@@ -95,7 +95,7 @@ object SerializationTestUtils{
def createTestLeaderAndIsrResponse() : LeaderAndIsrResponse = {
val responseMap = Map(((topic1, 0), ErrorMapping.NoError),
((topic2, 0), ErrorMapping.NoError))
- new LeaderAndIsrResponse(1, 1, responseMap)
+ new LeaderAndIsrResponse(1, responseMap)
}
def createTestStopReplicaRequest() : StopReplicaRequest = {
@@ -105,7 +105,7 @@ object SerializationTestUtils{
def createTestStopReplicaResponse() : StopReplicaResponse = {
val responseMap = Map(((topic1, 0), ErrorMapping.NoError),
((topic2, 0), ErrorMapping.NoError))
- new StopReplicaResponse(1, 0, responseMap.toMap)
+ new StopReplicaResponse(0, responseMap.toMap)
}
def createTestProducerRequest: ProducerRequest = {
@@ -113,7 +113,7 @@ object SerializationTestUtils{
}
def createTestProducerResponse: ProducerResponse =
- ProducerResponse(1, 1, Map(
+ ProducerResponse(1, Map(
TopicAndPartition(topic1, 0) -> ProducerResponseStatus(0.toShort, 10001),
TopicAndPartition(topic2, 0) -> ProducerResponseStatus(0.toShort, 20001)
))
@@ -123,7 +123,7 @@ object SerializationTestUtils{
}
def createTestFetchResponse: FetchResponse = {
- FetchResponse(1, 1, topicDataFetchResponse)
+ FetchResponse(1, topicDataFetchResponse)
}
def createTestOffsetRequest = new OffsetRequest(
@@ -132,7 +132,7 @@ object SerializationTestUtils{
)
def createTestOffsetResponse: OffsetResponse = {
- new OffsetResponse(OffsetRequest.CurrentVersion, 0, collection.immutable.Map(
+ new OffsetResponse(0, collection.immutable.Map(
TopicAndPartition(topic1, 1) -> PartitionOffsetsResponse(ErrorMapping.NoError, Seq(1000l, 2000l, 3000l, 4000l)))
)
}
@@ -142,7 +142,7 @@ object SerializationTestUtils{
}
def createTestTopicMetadataResponse: TopicMetadataResponse = {
- new TopicMetadataResponse(1, Seq(topicmetaData1, topicmetaData2), 1)
+ new TopicMetadataResponse(Seq(topicmetaData1, topicmetaData2), 1)
}
}
Modified: kafka/branches/0.8/core/src/test/scala/unit/kafka/producer/AsyncProducerTest.scala
URL: http://svn.apache.org/viewvc/kafka/branches/0.8/core/src/test/scala/unit/kafka/producer/AsyncProducerTest.scala?rev=1418181&r1=1418180&r2=1418181&view=diff
==============================================================================
--- kafka/branches/0.8/core/src/test/scala/unit/kafka/producer/AsyncProducerTest.scala (original)
+++ kafka/branches/0.8/core/src/test/scala/unit/kafka/producer/AsyncProducerTest.scala Fri Dec 7 04:01:29 2012
@@ -429,11 +429,11 @@ class AsyncProducerTest extends JUnit3Su
// On the third try for partition 0, let it succeed.
val request1 = TestUtils.produceRequestWithAcks(List(topic1), List(0, 1), messagesToSet(msgs), acks = 0, correlationId = 0)
val request2 = TestUtils.produceRequestWithAcks(List(topic1), List(0, 1), messagesToSet(msgs), acks = 0, correlationId = 1)
- val response1 = ProducerResponse(ProducerRequest.CurrentVersion, 0,
+ val response1 = ProducerResponse(0,
Map((TopicAndPartition("topic1", 0), ProducerResponseStatus(ErrorMapping.NotLeaderForPartitionCode.toShort, 0L)),
(TopicAndPartition("topic1", 1), ProducerResponseStatus(ErrorMapping.NoError, 0L))))
val request3 = TestUtils.produceRequest(topic1, 0, messagesToSet(msgs), correlationId = 2)
- val response2 = ProducerResponse(ProducerRequest.CurrentVersion, 0,
+ val response2 = ProducerResponse(0,
Map((TopicAndPartition("topic1", 0), ProducerResponseStatus(ErrorMapping.NoError, 0L))))
val mockSyncProducer = EasyMock.createMock(classOf[SyncProducer])
EasyMock.expect(mockSyncProducer.send(request1)).andThrow(new RuntimeException) // simulate SocketTimeoutException