kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From j...@apache.org
Subject [1/5] kafka git commit: MINOR: Remove Struct from Request/Response classes
Date Fri, 17 Feb 2017 22:33:54 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk 1f2ee5f0a -> fc1cfe475


http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/core/src/test/scala/unit/kafka/server/BaseRequestTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/server/BaseRequestTest.scala b/core/src/test/scala/unit/kafka/server/BaseRequestTest.scala
index b864e5d..a26bc2e 100644
--- a/core/src/test/scala/unit/kafka/server/BaseRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/BaseRequestTest.scala
@@ -26,8 +26,9 @@ import kafka.integration.KafkaServerTestHarness
 import kafka.network.SocketServer
 import kafka.utils._
 import org.apache.kafka.common.network.ListenerName
+import org.apache.kafka.common.protocol.types.Struct
 import org.apache.kafka.common.protocol.{ApiKeys, SecurityProtocol}
-import org.apache.kafka.common.requests.{AbstractRequest, RequestHeader, ResponseHeader}
+import org.apache.kafka.common.requests.{AbstractRequest, AbstractRequestResponse, RequestHeader,
ResponseHeader}
 
 abstract class BaseRequestTest extends KafkaServerTestHarness {
   private var correlationId = 0
@@ -97,41 +98,64 @@ abstract class BaseRequestTest extends KafkaServerTestHarness {
   }
 
   /**
-    *
-    * @param request
-    * @param apiKey
     * @param destination An optional SocketServer ot send the request to. If not set, any
available server is used.
     * @param protocol An optional SecurityProtocol to use. If not set, PLAINTEXT is used.
-    * @return
+    * @return A ByteBuffer containing the response (without the response header)
     */
-  def send(request: AbstractRequest, apiKey: ApiKeys,
-           destination: SocketServer = anySocketServer, protocol: SecurityProtocol = SecurityProtocol.PLAINTEXT):
ByteBuffer = {
+  def connectAndSend(request: AbstractRequest, apiKey: ApiKeys,
+                     destination: SocketServer = anySocketServer,
+                     apiVersion: Option[Short] = None,
+                     protocol: SecurityProtocol = SecurityProtocol.PLAINTEXT): ByteBuffer
= {
     val socket = connect(destination, protocol)
-    try {
-      send(request, apiKey, socket)
-    } finally {
-      socket.close()
-    }
+    try send(request, apiKey, socket, apiVersion)
+    finally socket.close()
   }
 
   /**
-    * Serializes and send the request to the given api.
+    * @param destination An optional SocketServer ot send the request to. If not set, any
available server is used.
+    * @param protocol An optional SecurityProtocol to use. If not set, PLAINTEXT is used.
+    * @return A ByteBuffer containing the response (without the response header).
+    */
+  def connectAndSendStruct(requestStruct: Struct, apiKey: ApiKeys, apiVersion: Short,
+                           destination: SocketServer = anySocketServer,
+                           protocol: SecurityProtocol = SecurityProtocol.PLAINTEXT): ByteBuffer
= {
+    val socket = connect(destination, protocol)
+    try sendStruct(requestStruct, apiKey, socket, apiVersion)
+    finally socket.close()
+  }
+
+  /**
+    * Serializes and sends the request to the given api.
     * A ByteBuffer containing the response is returned.
     */
-  def send(request: AbstractRequest, apiKey: ApiKeys, socket: Socket): ByteBuffer = {
-    correlationId += 1
-    val serializedBytes = {
-      val header = new RequestHeader(apiKey.id, request.version, "client-id", correlationId)
-      val byteBuffer = ByteBuffer.allocate(header.sizeOf() + request.sizeOf)
-      header.writeTo(byteBuffer)
-      request.writeTo(byteBuffer)
-      byteBuffer.array()
-    }
+  def send(request: AbstractRequest, apiKey: ApiKeys, socket: Socket, apiVersion: Option[Short]
= None): ByteBuffer = {
+    val header = nextRequestHeader(apiKey, apiVersion.getOrElse(request.version))
+    val serializedBytes = request.serialize(header).array
+    val response = requestAndReceive(socket, serializedBytes)
+    skipResponseHeader(response)
+  }
 
+  /**
+    * Serializes and sends the requestStruct to the given api.
+    * A ByteBuffer containing the response (without the response header) is returned.
+    */
+  def sendStruct(requestStruct: Struct, apiKey: ApiKeys, socket: Socket, apiVersion: Short):
ByteBuffer = {
+    val header = nextRequestHeader(apiKey, apiVersion)
+    val serializedBytes = AbstractRequestResponse.serialize(header.toStruct, requestStruct).array
     val response = requestAndReceive(socket, serializedBytes)
+    skipResponseHeader(response)
+  }
 
+  private def skipResponseHeader(response: Array[Byte]): ByteBuffer = {
     val responseBuffer = ByteBuffer.wrap(response)
-    ResponseHeader.parse(responseBuffer) // Parse the header to ensure its valid and move
the buffer forward
+    // Parse the header to ensure its valid and move the buffer forward
+    ResponseHeader.parse(responseBuffer)
     responseBuffer
   }
+
+  def nextRequestHeader(apiKey: ApiKeys, apiVersion: Short): RequestHeader = {
+    correlationId += 1
+    new RequestHeader(apiKey.id, apiVersion, "client-id", correlationId)
+  }
+
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/core/src/test/scala/unit/kafka/server/CreateTopicsRequestTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/server/CreateTopicsRequestTest.scala b/core/src/test/scala/unit/kafka/server/CreateTopicsRequestTest.scala
index 6efa189..4ab9520 100644
--- a/core/src/test/scala/unit/kafka/server/CreateTopicsRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/CreateTopicsRequestTest.scala
@@ -116,30 +116,27 @@ class CreateTopicsRequestTest extends AbstractCreateTopicsRequestTest
{
     // Duplicate
     val singleRequest = new CreateTopicsRequest.Builder(Map("duplicate-topic" ->
         new CreateTopicsRequest.TopicDetails(1, 1.toShort)).asJava, 1000).build()
-    val duplicateRequest = duplicateFirstTopic(singleRequest)
-    assertFalse("Request doesn't have duplicate topics", duplicateRequest.duplicateTopics().isEmpty)
-    validateErrorCreateTopicsRequests(duplicateRequest, Map("duplicate-topic" -> error(Errors.INVALID_REQUEST,
-      Some("""Create topics request from client `client-id` contains multiple entries for
the following topics: duplicate-topic"""))))
+    validateErrorCreateTopicsRequests(singleRequest, Map("duplicate-topic" -> error(Errors.INVALID_REQUEST,
+      Some("""Create topics request from client `client-id` contains multiple entries for
the following topics: duplicate-topic"""))),
+      requestStruct = Some(toStructWithDuplicateFirstTopic(singleRequest)))
 
     // Duplicate Partial with validateOnly
     val doubleRequestValidateOnly = new CreateTopicsRequest.Builder(Map(
       "duplicate-topic" -> new CreateTopicsRequest.TopicDetails(1, 1.toShort),
       "other-topic" -> new CreateTopicsRequest.TopicDetails(1, 1.toShort)).asJava, 1000,
true).build()
-    val duplicateDoubleRequestValidateOnly = duplicateFirstTopic(doubleRequestValidateOnly)
-    assertFalse("Request doesn't have duplicate topics", duplicateDoubleRequestValidateOnly.duplicateTopics.isEmpty)
-    validateErrorCreateTopicsRequests(duplicateDoubleRequestValidateOnly, Map(
+    validateErrorCreateTopicsRequests(doubleRequestValidateOnly, Map(
       "duplicate-topic" -> error(Errors.INVALID_REQUEST),
-      "other-topic" -> error(Errors.NONE)), checkErrorMessage = false)
+      "other-topic" -> error(Errors.NONE)), checkErrorMessage = false,
+      requestStruct = Some(toStructWithDuplicateFirstTopic(doubleRequestValidateOnly)))
 
     // Duplicate Partial
     val doubleRequest = new CreateTopicsRequest.Builder(Map(
       "duplicate-topic" -> new CreateTopicsRequest.TopicDetails(1, 1.toShort),
       "other-topic" -> new CreateTopicsRequest.TopicDetails(1, 1.toShort)).asJava, 1000).build()
-    val duplicateDoubleRequest = duplicateFirstTopic(doubleRequest)
-    assertFalse("Request doesn't have duplicate topics", duplicateDoubleRequest.duplicateTopics.isEmpty)
-    validateErrorCreateTopicsRequests(duplicateDoubleRequest, Map(
+    validateErrorCreateTopicsRequests(doubleRequest, Map(
       "duplicate-topic" -> error(Errors.INVALID_REQUEST),
-      "other-topic" -> error(Errors.NONE)), checkErrorMessage = false)
+      "other-topic" -> error(Errors.NONE)), checkErrorMessage = false,
+      requestStruct = Some(toStructWithDuplicateFirstTopic(doubleRequest)))
 
     // Partitions/ReplicationFactor and ReplicaAssignment
     val assignments = replicaAssignmentToJava(Map(0 -> List(0)))

http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/core/src/test/scala/unit/kafka/server/DeleteTopicsRequestTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/server/DeleteTopicsRequestTest.scala b/core/src/test/scala/unit/kafka/server/DeleteTopicsRequestTest.scala
index 9a092d0..9cd53d8 100644
--- a/core/src/test/scala/unit/kafka/server/DeleteTopicsRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/DeleteTopicsRequestTest.scala
@@ -112,12 +112,12 @@ class DeleteTopicsRequestTest extends BaseRequestTest {
   }
 
   private def sendDeleteTopicsRequest(request: DeleteTopicsRequest, socketServer: SocketServer
= controllerSocketServer): DeleteTopicsResponse = {
-    val response = send(request, ApiKeys.DELETE_TOPICS, socketServer)
+    val response = connectAndSend(request, ApiKeys.DELETE_TOPICS, socketServer)
     DeleteTopicsResponse.parse(response, request.version)
   }
 
   private def sendMetadataRequest(request: MetadataRequest): MetadataResponse = {
-    val response = send(request, ApiKeys.METADATA)
-    MetadataResponse.parse(response)
+    val response = connectAndSend(request, ApiKeys.METADATA)
+    MetadataResponse.parse(response, request.version)
   }
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/core/src/test/scala/unit/kafka/server/EdgeCaseRequestTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/server/EdgeCaseRequestTest.scala b/core/src/test/scala/unit/kafka/server/EdgeCaseRequestTest.scala
index 1f9e18b..2d4a22a 100755
--- a/core/src/test/scala/unit/kafka/server/EdgeCaseRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/EdgeCaseRequestTest.scala
@@ -114,15 +114,15 @@ class EdgeCaseRequestTest extends KafkaServerTestHarness {
     val correlationId = -1
     TestUtils.createTopic(zkUtils, topic, numPartitions = 1, replicationFactor = 1, servers
= servers)
 
+    val version = 2: Short
     val serializedBytes = {
-      val headerBytes = requestHeaderBytes(ApiKeys.PRODUCE.id, 2, null, correlationId)
+      val headerBytes = requestHeaderBytes(ApiKeys.PRODUCE.id, version, null, correlationId)
       val messageBytes = "message".getBytes
       val records = MemoryRecords.readableRecords(ByteBuffer.wrap(messageBytes))
-      val request = new ProduceRequest.Builder(
-          1, 10000, Map(topicPartition -> records).asJava).build()
-      val byteBuffer = ByteBuffer.allocate(headerBytes.length + request.sizeOf)
+      val request = new ProduceRequest.Builder(1, 10000, Map(topicPartition -> records).asJava).build()
+      val byteBuffer = ByteBuffer.allocate(headerBytes.length + request.toStruct.sizeOf)
       byteBuffer.put(headerBytes)
-      request.writeTo(byteBuffer)
+      request.toStruct.writeTo(byteBuffer)
       byteBuffer.array()
     }
 
@@ -130,13 +130,13 @@ class EdgeCaseRequestTest extends KafkaServerTestHarness {
 
     val responseBuffer = ByteBuffer.wrap(response)
     val responseHeader = ResponseHeader.parse(responseBuffer)
-    val produceResponse = ProduceResponse.parse(responseBuffer)
+    val produceResponse = ProduceResponse.parse(responseBuffer, version)
 
-    assertEquals("The response should parse completely", 0, responseBuffer.remaining())
-    assertEquals("The correlationId should match request", correlationId, responseHeader.correlationId())
-    assertEquals("One partition response should be returned", 1, produceResponse.responses().size())
+    assertEquals("The response should parse completely", 0, responseBuffer.remaining)
+    assertEquals("The correlationId should match request", correlationId, responseHeader.correlationId)
+    assertEquals("One partition response should be returned", 1, produceResponse.responses.size)
 
-    val partitionResponse = produceResponse.responses().get(topicPartition)
+    val partitionResponse = produceResponse.responses.get(topicPartition)
     assertNotNull(partitionResponse)
     assertEquals("There should be no error", Errors.NONE, partitionResponse.error)
   }

http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/core/src/test/scala/unit/kafka/server/FetchRequestTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/server/FetchRequestTest.scala b/core/src/test/scala/unit/kafka/server/FetchRequestTest.scala
index 3811360..64be5b3 100644
--- a/core/src/test/scala/unit/kafka/server/FetchRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/FetchRequestTest.scala
@@ -24,7 +24,7 @@ import kafka.utils.TestUtils
 import kafka.utils.TestUtils._
 import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
 import org.apache.kafka.common.TopicPartition
-import org.apache.kafka.common.protocol.{ApiKeys, Errors}
+import org.apache.kafka.common.protocol.{ApiKeys, Errors, ProtoUtils}
 import org.apache.kafka.common.record.LogEntry
 import org.apache.kafka.common.requests.{FetchRequest, FetchResponse}
 import org.apache.kafka.common.serialization.StringSerializer
@@ -56,7 +56,7 @@ class FetchRequestTest extends BaseRequestTest {
 
   private def createFetchRequest(maxResponseBytes: Int, maxPartitionBytes: Int, topicPartitions:
Seq[TopicPartition],
                                  offsetMap: Map[TopicPartition, Long] = Map.empty): FetchRequest
=
-    new FetchRequest.Builder(Int.MaxValue, 0, createPartitionMap(maxPartitionBytes, topicPartitions,
offsetMap))
+    FetchRequest.Builder.forConsumer(Int.MaxValue, 0, createPartitionMap(maxPartitionBytes,
topicPartitions, offsetMap))
       .setMaxBytes(maxResponseBytes).build()
 
   private def createPartitionMap(maxPartitionBytes: Int, topicPartitions: Seq[TopicPartition],
@@ -69,8 +69,8 @@ class FetchRequestTest extends BaseRequestTest {
   }
 
   private def sendFetchRequest(leaderId: Int, request: FetchRequest): FetchResponse = {
-    val response = send(request, ApiKeys.FETCH, destination = brokerSocketServer(leaderId))
-    FetchResponse.parse(response)
+    val response = connectAndSend(request, ApiKeys.FETCH, destination = brokerSocketServer(leaderId))
+    FetchResponse.parse(response, ProtoUtils.latestVersion(ApiKeys.FETCH.id))
   }
 
   @Test
@@ -156,10 +156,9 @@ class FetchRequestTest extends BaseRequestTest {
     val (topicPartition, leaderId) = createTopics(numTopics = 1, numPartitions = 1).head
     producer.send(new ProducerRecord(topicPartition.topic, topicPartition.partition,
       "key", new String(new Array[Byte](maxPartitionBytes + 1)))).get
-    val fetchRequestBuilder = new FetchRequest.Builder(
-      Int.MaxValue, 0, createPartitionMap(maxPartitionBytes, Seq(topicPartition))).
-      setVersion(2)
-    val fetchResponse = sendFetchRequest(leaderId, fetchRequestBuilder.build())
+    val fetchRequest = FetchRequest.Builder.forConsumer(Int.MaxValue, 0,
+      createPartitionMap(maxPartitionBytes, Seq(topicPartition))).build(2)
+    val fetchResponse = sendFetchRequest(leaderId, fetchRequest)
     val partitionData = fetchResponse.responseData.get(topicPartition)
     assertEquals(Errors.NONE, partitionData.error)
     assertTrue(partitionData.highWatermark > 0)

http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/core/src/test/scala/unit/kafka/server/MetadataCacheTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/server/MetadataCacheTest.scala b/core/src/test/scala/unit/kafka/server/MetadataCacheTest.scala
index 99a95ad..1b5007d 100644
--- a/core/src/test/scala/unit/kafka/server/MetadataCacheTest.scala
+++ b/core/src/test/scala/unit/kafka/server/MetadataCacheTest.scala
@@ -22,7 +22,7 @@ import util.Arrays.asList
 import kafka.common.BrokerEndPointNotAvailableException
 import org.apache.kafka.common.TopicPartition
 import org.apache.kafka.common.network.ListenerName
-import org.apache.kafka.common.protocol.{Errors, SecurityProtocol}
+import org.apache.kafka.common.protocol.{ApiKeys, Errors, ProtoUtils, SecurityProtocol}
 import org.apache.kafka.common.requests.{PartitionState, UpdateMetadataRequest}
 import org.apache.kafka.common.requests.UpdateMetadataRequest.{Broker, EndPoint}
 import org.junit.Test
@@ -69,8 +69,9 @@ class MetadataCacheTest {
       new TopicPartition(topic, 1) -> new PartitionState(controllerEpoch, 1, 1, asList(1),
zkVersion, asSet(1)),
       new TopicPartition(topic, 2) -> new PartitionState(controllerEpoch, 2, 2, asList(2),
zkVersion, asSet(2)))
 
-    val updateMetadataRequest = new UpdateMetadataRequest.Builder(
-      controllerId, controllerEpoch, partitionStates.asJava, brokers.asJava).build()
+    val version = ProtoUtils.latestVersion(ApiKeys.UPDATE_METADATA_KEY.id)
+    val updateMetadataRequest = new UpdateMetadataRequest.Builder(version, controllerId,
controllerEpoch,
+      partitionStates.asJava, brokers.asJava).build()
     cache.updateCache(15, updateMetadataRequest)
 
     for (securityProtocol <- Seq(SecurityProtocol.PLAINTEXT, SecurityProtocol.SSL)) {
@@ -120,8 +121,9 @@ class MetadataCacheTest {
     val partitionStates = Map(
       new TopicPartition(topic, 0) -> new PartitionState(controllerEpoch, leader, leaderEpoch,
asList(0), zkVersion, asSet(0)))
 
-    val updateMetadataRequest = new UpdateMetadataRequest.Builder(
-      controllerId, controllerEpoch, partitionStates.asJava, brokers.asJava).build()
+    val version = ProtoUtils.latestVersion(ApiKeys.UPDATE_METADATA_KEY.id)
+    val updateMetadataRequest = new UpdateMetadataRequest.Builder(version, controllerId,
controllerEpoch,
+      partitionStates.asJava, brokers.asJava).build()
     cache.updateCache(15, updateMetadataRequest)
 
     val topicMetadatas = cache.getTopicMetadata(Set(topic), listenerName)
@@ -163,8 +165,9 @@ class MetadataCacheTest {
     val partitionStates = Map(
       new TopicPartition(topic, 0) -> new PartitionState(controllerEpoch, leader, leaderEpoch,
isr, zkVersion, replicas))
 
-    val updateMetadataRequest = new UpdateMetadataRequest.Builder(
-      controllerId, controllerEpoch, partitionStates.asJava, brokers.asJava).build()
+    val version = ProtoUtils.latestVersion(ApiKeys.UPDATE_METADATA_KEY.id)
+    val updateMetadataRequest = new UpdateMetadataRequest.Builder(version, controllerId,
controllerEpoch,
+      partitionStates.asJava, brokers.asJava).build()
     cache.updateCache(15, updateMetadataRequest)
 
     // Validate errorUnavailableEndpoints = false
@@ -222,8 +225,9 @@ class MetadataCacheTest {
     val partitionStates = Map(
       new TopicPartition(topic, 0) -> new PartitionState(controllerEpoch, leader, leaderEpoch,
isr, zkVersion, replicas))
 
-    val updateMetadataRequest = new UpdateMetadataRequest.Builder(
-      controllerId, controllerEpoch, partitionStates.asJava, brokers.asJava).build()
+    val version = ProtoUtils.latestVersion(ApiKeys.UPDATE_METADATA_KEY.id)
+    val updateMetadataRequest = new UpdateMetadataRequest.Builder(version, controllerId,
controllerEpoch,
+      partitionStates.asJava, brokers.asJava).build()
     cache.updateCache(15, updateMetadataRequest)
 
     // Validate errorUnavailableEndpoints = false
@@ -273,8 +277,9 @@ class MetadataCacheTest {
     val isr = asList[Integer](0, 1)
     val partitionStates = Map(
       new TopicPartition(topic, 0) -> new PartitionState(controllerEpoch, leader, leaderEpoch,
isr, 3, replicas))
-    val updateMetadataRequest = new UpdateMetadataRequest.Builder(
-      2, controllerEpoch, partitionStates.asJava, brokers.asJava).build()
+    val version = ProtoUtils.latestVersion(ApiKeys.UPDATE_METADATA_KEY.id)
+    val updateMetadataRequest = new UpdateMetadataRequest.Builder(version, 2, controllerEpoch,
partitionStates.asJava,
+      brokers.asJava).build()
     cache.updateCache(15, updateMetadataRequest)
 
     try {
@@ -305,8 +310,9 @@ class MetadataCacheTest {
       val isr = asList[Integer](0, 1)
       val partitionStates = Map(
         new TopicPartition(topic, 0) -> new PartitionState(controllerEpoch, leader, leaderEpoch,
isr, 3, replicas))
-      val updateMetadataRequest = new UpdateMetadataRequest.Builder(
-        2, controllerEpoch, partitionStates.asJava, brokers.asJava).build()
+      val version = ProtoUtils.latestVersion(ApiKeys.UPDATE_METADATA_KEY.id)
+      val updateMetadataRequest = new UpdateMetadataRequest.Builder(version, 2, controllerEpoch,
partitionStates.asJava,
+        brokers.asJava).build()
       cache.updateCache(15, updateMetadataRequest)
     }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/core/src/test/scala/unit/kafka/server/MetadataRequestTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/server/MetadataRequestTest.scala b/core/src/test/scala/unit/kafka/server/MetadataRequestTest.scala
index f3bb912..ed0e805 100644
--- a/core/src/test/scala/unit/kafka/server/MetadataRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/MetadataRequestTest.scala
@@ -36,14 +36,14 @@ class MetadataRequestTest extends BaseRequestTest {
 
   @Test
   def testClusterIdWithRequestVersion1() {
-    val v1MetadataResponse = sendMetadataRequest(MetadataRequest.allTopics(1.toShort))
+    val v1MetadataResponse = sendMetadataRequest(MetadataRequest.Builder.allTopics.build(1.toShort))
     val v1ClusterId = v1MetadataResponse.clusterId
     assertNull(s"v1 clusterId should be null", v1ClusterId)
   }
 
   @Test
   def testClusterIdIsValid() {
-    val metadataResponse = sendMetadataRequest(MetadataRequest.allTopics(2.toShort))
+    val metadataResponse = sendMetadataRequest(MetadataRequest.Builder.allTopics.build(2.toShort))
     isValidClusterId(metadataResponse.clusterId)
   }
 
@@ -51,7 +51,7 @@ class MetadataRequestTest extends BaseRequestTest {
   def testControllerId() {
     val controllerServer = servers.find(_.kafkaController.isActive).get
     val controllerId = controllerServer.config.brokerId
-    val metadataResponse = sendMetadataRequest(MetadataRequest.allTopics(1.toShort))
+    val metadataResponse = sendMetadataRequest(MetadataRequest.Builder.allTopics.build(1.toShort))
 
     assertEquals("Controller id should match the active controller",
       controllerId, metadataResponse.controller.id)
@@ -64,14 +64,14 @@ class MetadataRequestTest extends BaseRequestTest {
     val controllerId2 = controllerServer2.config.brokerId
     assertNotEquals("Controller id should switch to a new broker", controllerId, controllerId2)
     TestUtils.waitUntilTrue(() => {
-      val metadataResponse2 = sendMetadataRequest(MetadataRequest.allTopics(1.toShort))
+      val metadataResponse2 = sendMetadataRequest(MetadataRequest.Builder.allTopics.build(1.toShort))
       metadataResponse2.controller != null && controllerServer2.apis.brokerId ==
metadataResponse2.controller.id
     }, "Controller id should match the active controller after failover", 5000)
   }
 
   @Test
   def testRack() {
-    val metadataResponse = sendMetadataRequest(MetadataRequest.allTopics(1.toShort))
+    val metadataResponse = sendMetadataRequest(MetadataRequest.Builder.allTopics.build(1.toShort))
     // Validate rack matches what's set in generateConfigs() above
     metadataResponse.brokers.asScala.foreach { broker =>
       assertEquals("Rack information should match config", s"rack/${broker.id}", broker.rack)
@@ -86,7 +86,7 @@ class MetadataRequestTest extends BaseRequestTest {
     TestUtils.createTopic(zkUtils, internalTopic, 3, 2, servers)
     TestUtils.createTopic(zkUtils, notInternalTopic, 3, 2, servers)
 
-    val metadataResponse = sendMetadataRequest(MetadataRequest.allTopics(1.toShort))
+    val metadataResponse = sendMetadataRequest(MetadataRequest.Builder.allTopics.build(1.toShort))
     assertTrue("Response should have no errors", metadataResponse.errors.isEmpty)
 
     val topicMetadata = metadataResponse.topicMetadata.asScala
@@ -124,7 +124,7 @@ class MetadataRequestTest extends BaseRequestTest {
     assertEquals("V0 Response should have 2 (all) topics", 2, metadataResponseV0.topicMetadata.size())
 
     // v1, Null represents all topics
-    val metadataResponseV1 = sendMetadataRequest(MetadataRequest.allTopics(1.toShort))
+    val metadataResponseV1 = sendMetadataRequest(MetadataRequest.Builder.allTopics.build(1.toShort))
     assertTrue("V1 Response should have no errors", metadataResponseV1.errors.isEmpty)
     assertEquals("V1 Response should have 2 (all) topics", 2, metadataResponseV1.topicMetadata.size())
   }
@@ -177,7 +177,7 @@ class MetadataRequestTest extends BaseRequestTest {
   }
 
   private def sendMetadataRequest(request: MetadataRequest): MetadataResponse = {
-    val response = send(request, ApiKeys.METADATA)
+    val response = connectAndSend(request, ApiKeys.METADATA)
     MetadataResponse.parse(response, request.version)
   }
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/core/src/test/scala/unit/kafka/server/ProduceRequestTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/server/ProduceRequestTest.scala b/core/src/test/scala/unit/kafka/server/ProduceRequestTest.scala
index b05be9d..81118fa 100644
--- a/core/src/test/scala/unit/kafka/server/ProduceRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ProduceRequestTest.scala
@@ -88,8 +88,8 @@ class ProduceRequestTest extends BaseRequestTest {
   }
 
   private def sendProduceRequest(leaderId: Int, request: ProduceRequest): ProduceResponse
= {
-    val response = send(request, ApiKeys.PRODUCE, destination = brokerSocketServer(leaderId))
-    ProduceResponse.parse(response)
+    val response = connectAndSend(request, ApiKeys.PRODUCE, destination = brokerSocketServer(leaderId))
+    ProduceResponse.parse(response, request.version)
   }
 
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/core/src/test/scala/unit/kafka/server/SaslApiVersionsRequestTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/server/SaslApiVersionsRequestTest.scala b/core/src/test/scala/unit/kafka/server/SaslApiVersionsRequestTest.scala
index 927ace9..92a518d 100644
--- a/core/src/test/scala/unit/kafka/server/SaslApiVersionsRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/SaslApiVersionsRequestTest.scala
@@ -20,14 +20,13 @@ import java.io.IOException
 import java.net.Socket
 import java.util.Collections
 
-import org.apache.kafka.common.protocol.{ApiKeys, Errors, ProtoUtils, SecurityProtocol}
+import org.apache.kafka.common.protocol.{ApiKeys, Errors, SecurityProtocol}
 import org.apache.kafka.common.requests.{ApiVersionsRequest, ApiVersionsResponse}
 import org.apache.kafka.common.requests.SaslHandshakeRequest
 import org.apache.kafka.common.requests.SaslHandshakeResponse
 import org.junit.Test
 import org.junit.Assert._
 import kafka.api.SaslTestHarness
-import org.apache.kafka.common.protocol.types.Struct
 
 class SaslApiVersionsRequestTest extends BaseRequestTest with SaslTestHarness {
   override protected def securityProtocol = SecurityProtocol.SASL_PLAINTEXT
@@ -42,8 +41,7 @@ class SaslApiVersionsRequestTest extends BaseRequestTest with SaslTestHarness
{
   def testApiVersionsRequestBeforeSaslHandshakeRequest() {
     val plaintextSocket = connect(protocol = securityProtocol)
     try {
-      val apiVersionsResponse = sendApiVersionsRequest(plaintextSocket,
-          new ApiVersionsRequest.Builder().setVersion(0).build())
+      val apiVersionsResponse = sendApiVersionsRequest(plaintextSocket, new ApiVersionsRequest.Builder().build(0))
       ApiVersionsRequestTest.validateApiVersionsResponse(apiVersionsResponse)
       sendSaslHandshakeRequestValidateResponse(plaintextSocket)
     } finally {
@@ -57,8 +55,7 @@ class SaslApiVersionsRequestTest extends BaseRequestTest with SaslTestHarness
{
     try {
       sendSaslHandshakeRequestValidateResponse(plaintextSocket)
       try {
-        sendApiVersionsRequest(plaintextSocket,
-            new ApiVersionsRequest.Builder().setVersion(0).build())
+        sendApiVersionsRequest(plaintextSocket, new ApiVersionsRequest.Builder().build(0))
         fail("Versions Request during Sasl handshake did not fail")
       } catch {
         case _: IOException => // expected exception
@@ -72,12 +69,10 @@ class SaslApiVersionsRequestTest extends BaseRequestTest with SaslTestHarness
{
   def testApiVersionsRequestWithUnsupportedVersion() {
     val plaintextSocket = connect(protocol = securityProtocol)
     try {
-      val apiVersionsRequest = new ApiVersionsRequest(
-        new Struct(ProtoUtils.requestSchema(ApiKeys.API_VERSIONS.id, 0)), Short.MaxValue);
-      val apiVersionsResponse = sendApiVersionsRequest(plaintextSocket, apiVersionsRequest)
+      val apiVersionsRequest = new ApiVersionsRequest(0)
+      val apiVersionsResponse = sendApiVersionsRequest(plaintextSocket, apiVersionsRequest,
Some(Short.MaxValue))
       assertEquals(Errors.UNSUPPORTED_VERSION, apiVersionsResponse.error)
-      val apiVersionsResponse2 = sendApiVersionsRequest(plaintextSocket,
-          new ApiVersionsRequest.Builder().setVersion(0).build())
+      val apiVersionsResponse2 = sendApiVersionsRequest(plaintextSocket, new ApiVersionsRequest.Builder().build(0))
       ApiVersionsRequestTest.validateApiVersionsResponse(apiVersionsResponse2)
       sendSaslHandshakeRequestValidateResponse(plaintextSocket)
     } finally {
@@ -85,15 +80,17 @@ class SaslApiVersionsRequestTest extends BaseRequestTest with SaslTestHarness
{
     }
   }
 
-  private def sendApiVersionsRequest(socket: Socket, request: ApiVersionsRequest): ApiVersionsResponse
= {
-    val response = send(request, ApiKeys.API_VERSIONS, socket)
-    ApiVersionsResponse.parse(response)
+  private def sendApiVersionsRequest(socket: Socket, request: ApiVersionsRequest,
+                                     apiVersion: Option[Short] = None): ApiVersionsResponse
= {
+    val response = send(request, ApiKeys.API_VERSIONS, socket, apiVersion)
+    ApiVersionsResponse.parse(response, request.version)
   }
 
   private def sendSaslHandshakeRequestValidateResponse(socket: Socket) {
-    val response = send(new SaslHandshakeRequest("PLAIN"), ApiKeys.SASL_HANDSHAKE, socket)
-    val handshakeResponse = SaslHandshakeResponse.parse(response)
+    val request = new SaslHandshakeRequest("PLAIN")
+    val response = send(request, ApiKeys.SASL_HANDSHAKE, socket)
+    val handshakeResponse = SaslHandshakeResponse.parse(response, request.version)
     assertEquals(Errors.NONE, handshakeResponse.error)
-    assertEquals(Collections.singletonList("PLAIN"), handshakeResponse.enabledMechanisms())
+    assertEquals(Collections.singletonList("PLAIN"), handshakeResponse.enabledMechanisms)
   }
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopicManagerTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopicManagerTest.java
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopicManagerTest.java
index 69b3c46..5399653 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopicManagerTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopicManagerTest.java
@@ -106,8 +106,8 @@ public class InternalTopicManagerTest {
             MetadataResponse.PartitionMetadata partitionMetadata = new MetadataResponse.PartitionMetadata(Errors.NONE,
1, node, new ArrayList<Node>(), new ArrayList<Node>());
             MetadataResponse.TopicMetadata topicMetadata = new MetadataResponse.TopicMetadata(Errors.NONE,
topic, true, Collections.singletonList(partitionMetadata));
             MetadataResponse response = new MetadataResponse(Collections.<Node>emptyList(),
null, MetadataResponse.NO_CONTROLLER_ID,
-                Collections.singletonList(topicMetadata), 0);
+                Collections.singletonList(topicMetadata));
             return response;
         }
     }
-}
\ No newline at end of file
+}


Mime
View raw message