kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From j...@apache.org
Subject [2/5] kafka git commit: KAFKA-2066; Use client-side FetchRequest/FetchResponse on server
Date Tue, 15 Nov 2016 00:46:28 GMT
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b4c3479/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java
index b7645dd..097b4fc 100644
--- a/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java
@@ -12,31 +12,19 @@
  */
 package org.apache.kafka.clients.producer.internals;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-
-import java.util.Collections;
-import java.util.LinkedHashMap;
-import java.util.Map;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
-
 import org.apache.kafka.clients.Metadata;
 import org.apache.kafka.clients.MockClient;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.Cluster;
-import org.apache.kafka.common.internals.ClusterResourceListeners;
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.internals.ClusterResourceListeners;
 import org.apache.kafka.common.metrics.KafkaMetric;
 import org.apache.kafka.common.metrics.MetricConfig;
 import org.apache.kafka.common.metrics.Metrics;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.Errors;
-import org.apache.kafka.common.protocol.types.Struct;
 import org.apache.kafka.common.record.CompressionType;
 import org.apache.kafka.common.record.Record;
 import org.apache.kafka.common.requests.ProduceResponse;
@@ -46,6 +34,17 @@ import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
 public class SenderTest {
 
     private static final int MAX_REQUEST_SIZE = 1024 * 1024;
@@ -69,7 +68,7 @@ public class SenderTest {
 
     @Before
     public void setup() {
-        Map<String, String> metricTags = new LinkedHashMap<String, String>();
+        Map<String, String> metricTags = new LinkedHashMap<>();
         metricTags.put("client-id", CLIENT_ID);
         MetricConfig metricConfig = new MetricConfig().tags(metricTags);
         metrics = new Metrics(metricConfig, time);
@@ -147,7 +146,7 @@ public class SenderTest {
             Future<RecordMetadata> future = accumulator.append(tp, 0L, "key".getBytes(), "value".getBytes(), null, MAX_BLOCK_TIMEOUT).future;
             sender.run(time.milliseconds()); // connect
             sender.run(time.milliseconds()); // send produce request
-            String id = client.requests().peek().request().destination();
+            String id = client.requests().peek().destination();
             Node node = new Node(Integer.valueOf(id), "localhost", 0);
             assertEquals(1, client.inFlightRequestCount());
             assertTrue("Client ready status should be true", client.isReady(node, 0L));
@@ -168,7 +167,7 @@ public class SenderTest {
             future = accumulator.append(tp, 0L, "key".getBytes(), "value".getBytes(), null, MAX_BLOCK_TIMEOUT).future;
             sender.run(time.milliseconds()); // send produce request
             for (int i = 0; i < maxRetries + 1; i++) {
-                client.disconnect(client.requests().peek().request().destination());
+                client.disconnect(client.requests().peek().destination());
                 sender.run(time.milliseconds()); // receive error
                 sender.run(time.milliseconds()); // reconnect
                 sender.run(time.milliseconds()); // resend
@@ -205,8 +204,8 @@ public class SenderTest {
             accumulator.append(tp2, 0L, "key1".getBytes(), "value1".getBytes(), null, MAX_BLOCK_TIMEOUT);
             sender.run(time.milliseconds()); // connect
             sender.run(time.milliseconds()); // send produce request
-            String id = client.requests().peek().request().destination();
-            assertEquals(ApiKeys.PRODUCE.id, client.requests().peek().request().header().apiKey());
+            String id = client.requests().peek().destination();
+            assertEquals(ApiKeys.PRODUCE.id, client.requests().peek().header().apiKey());
             Node node = new Node(Integer.valueOf(id), "localhost", 0);
             assertEquals(1, client.inFlightRequestCount());
             assertTrue("Client ready status should be true", client.isReady(node, 0L));
@@ -272,11 +271,10 @@ public class SenderTest {
         }
     }
 
-    private Struct produceResponse(TopicPartition tp, long offset, int error, int throttleTimeMs) {
+    private ProduceResponse produceResponse(TopicPartition tp, long offset, int error, int throttleTimeMs) {
         ProduceResponse.PartitionResponse resp = new ProduceResponse.PartitionResponse((short) error, offset, Record.NO_TIMESTAMP);
         Map<TopicPartition, ProduceResponse.PartitionResponse> partResp = Collections.singletonMap(tp, resp);
-        ProduceResponse response = new ProduceResponse(partResp, throttleTimeMs);
-        return response.toStruct();
+        return new ProduceResponse(partResp, throttleTimeMs);
     }
 
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/3b4c3479/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
index 30faac1..a5eb22a 100644
--- a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
@@ -16,15 +16,20 @@ package org.apache.kafka.common.requests;
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.UnknownServerException;
+import org.apache.kafka.common.network.Send;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.protocol.ProtoUtils;
 import org.apache.kafka.common.protocol.SecurityProtocol;
+import org.apache.kafka.common.protocol.types.Struct;
+import org.apache.kafka.common.record.MemoryRecords;
 import org.apache.kafka.common.record.Record;
 import org.junit.Test;
 
+import java.io.IOException;
 import java.lang.reflect.Method;
 import java.nio.ByteBuffer;
+import java.nio.channels.GatheringByteChannel;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -36,6 +41,7 @@ import java.util.Map;
 import java.util.Set;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 
 public class RequestResponseTest {
 
@@ -113,7 +119,6 @@ public class RequestResponseTest {
         for (AbstractRequestResponse req : requestResponseList)
             checkSerialization(req, null);
 
-
         checkOlderFetchVersions();
         checkSerialization(createMetadataResponse(0), 0);
         checkSerialization(createMetadataResponse(1), 1);
@@ -180,7 +185,9 @@ public class RequestResponseTest {
     @Test
     public void fetchResponseVersionTest() {
         Map<TopicPartition, FetchResponse.PartitionData> responseData = new HashMap<>();
-        responseData.put(new TopicPartition("test", 0), new FetchResponse.PartitionData(Errors.NONE.code(), 1000000, ByteBuffer.allocate(10)));
+
+        MemoryRecords records = MemoryRecords.readableRecords(ByteBuffer.allocate(10));
+        responseData.put(new TopicPartition("test", 0), new FetchResponse.PartitionData(Errors.NONE.code(), 1000000, records));
 
         FetchResponse v0Response = new FetchResponse(responseData);
         FetchResponse v1Response = new FetchResponse(responseData, 10);
@@ -193,6 +200,34 @@ public class RequestResponseTest {
     }
 
     @Test
+    public void verifyFetchResponseFullWrite() throws Exception {
+        FetchResponse fetchResponse = createFetchResponse();
+        RequestHeader header = new RequestHeader(ApiKeys.FETCH.id, "client", 15);
+
+        Send send = fetchResponse.toSend("1", header);
+        ByteBufferChannel channel = new ByteBufferChannel(send.size());
+        send.writeTo(channel);
+        channel.close();
+
+        ByteBuffer buf = channel.buf;
+
+        // read the size
+        int size = buf.getInt();
+        assertTrue(size > 0);
+
+        // read the header
+        ResponseHeader responseHeader = ResponseHeader.parse(channel.buf);
+        assertEquals(header.correlationId(), responseHeader.correlationId());
+
+        // read the body
+        Struct responseBody = ProtoUtils.responseSchema(ApiKeys.FETCH.id, header.apiVersion()).read(buf);
+        FetchResponse parsedResponse = new FetchResponse(responseBody);
+        assertEquals(parsedResponse, fetchResponse);
+
+        assertEquals(size, responseHeader.sizeOf() + parsedResponse.sizeOf());
+    }
+
+    @Test
     public void testControlledShutdownResponse() {
         ControlledShutdownResponse response = createControlledShutdownResponse();
         ByteBuffer buffer = ByteBuffer.allocate(response.sizeOf());
@@ -216,24 +251,24 @@ public class RequestResponseTest {
         assertEquals("", deserialized.clientId()); // null is defaulted to ""
     }
 
-    private AbstractRequestResponse createRequestHeader() {
+    private RequestHeader createRequestHeader() {
         return new RequestHeader((short) 10, (short) 1, "", 10);
     }
 
-    private AbstractRequestResponse createResponseHeader() {
+    private ResponseHeader createResponseHeader() {
         return new ResponseHeader(10);
     }
 
-    private AbstractRequest createGroupCoordinatorRequest() {
+    private GroupCoordinatorRequest createGroupCoordinatorRequest() {
         return new GroupCoordinatorRequest("test-group");
     }
 
-    private AbstractRequestResponse createGroupCoordinatorResponse() {
+    private GroupCoordinatorResponse createGroupCoordinatorResponse() {
         return new GroupCoordinatorResponse(Errors.NONE.code(), new Node(10, "host1", 2014));
     }
 
     @SuppressWarnings("deprecation")
-    private AbstractRequest createFetchRequest(int version) {
+    private FetchRequest createFetchRequest(int version) {
         LinkedHashMap<TopicPartition, FetchRequest.PartitionData> fetchData = new LinkedHashMap<>();
         fetchData.put(new TopicPartition("test1", 0), new FetchRequest.PartitionData(100, 1000000));
         fetchData.put(new TopicPartition("test2", 0), new FetchRequest.PartitionData(200, 1000000));
@@ -243,22 +278,23 @@ public class RequestResponseTest {
             return new FetchRequest(100, 1000, 1000000, fetchData);
     }
 
-    private AbstractRequestResponse createFetchResponse() {
+    private FetchResponse createFetchResponse() {
         Map<TopicPartition, FetchResponse.PartitionData> responseData = new HashMap<>();
-        responseData.put(new TopicPartition("test", 0), new FetchResponse.PartitionData(Errors.NONE.code(), 1000000, ByteBuffer.allocate(10)));
-        return new FetchResponse(responseData, 0);
+        MemoryRecords records = MemoryRecords.readableRecords(ByteBuffer.allocate(10));
+        responseData.put(new TopicPartition("test", 0), new FetchResponse.PartitionData(Errors.NONE.code(), 1000000, records));
+        return new FetchResponse(responseData, 25);
     }
 
-    private AbstractRequest createHeartBeatRequest() {
+    private HeartbeatRequest createHeartBeatRequest() {
         return new HeartbeatRequest("group1", 1, "consumer1");
     }
 
-    private AbstractRequestResponse createHeartBeatResponse() {
+    private HeartbeatResponse createHeartBeatResponse() {
         return new HeartbeatResponse(Errors.NONE.code());
     }
 
     @SuppressWarnings("deprecation")
-    private AbstractRequest createJoinGroupRequest(int version) {
+    private JoinGroupRequest createJoinGroupRequest(int version) {
         ByteBuffer metadata = ByteBuffer.wrap(new byte[] {});
         List<JoinGroupRequest.ProtocolMetadata> protocols = new ArrayList<>();
         protocols.add(new JoinGroupRequest.ProtocolMetadata("consumer-range", metadata));
@@ -269,27 +305,27 @@ public class RequestResponseTest {
         }
     }
 
-    private AbstractRequestResponse createJoinGroupResponse() {
+    private JoinGroupResponse createJoinGroupResponse() {
         Map<String, ByteBuffer> members = new HashMap<>();
         members.put("consumer1", ByteBuffer.wrap(new byte[]{}));
         members.put("consumer2", ByteBuffer.wrap(new byte[]{}));
         return new JoinGroupResponse(Errors.NONE.code(), 1, "range", "consumer1", "leader", members);
     }
 
-    private AbstractRequest createListGroupsRequest() {
+    private ListGroupsRequest createListGroupsRequest() {
         return new ListGroupsRequest();
     }
 
-    private AbstractRequestResponse createListGroupsResponse() {
+    private ListGroupsResponse createListGroupsResponse() {
         List<ListGroupsResponse.Group> groups = Arrays.asList(new ListGroupsResponse.Group("test-group", "consumer"));
         return new ListGroupsResponse(Errors.NONE.code(), groups);
     }
 
-    private AbstractRequest createDescribeGroupRequest() {
+    private DescribeGroupsRequest createDescribeGroupRequest() {
         return new DescribeGroupsRequest(Collections.singletonList("test-group"));
     }
 
-    private AbstractRequestResponse createDescribeGroupResponse() {
+    private DescribeGroupsResponse createDescribeGroupResponse() {
         String clientId = "consumer-1";
         String clientHost = "localhost";
         ByteBuffer empty = ByteBuffer.allocate(0);
@@ -300,16 +336,16 @@ public class RequestResponseTest {
         return new DescribeGroupsResponse(Collections.singletonMap("test-group", metadata));
     }
 
-    private AbstractRequest createLeaveGroupRequest() {
+    private LeaveGroupRequest createLeaveGroupRequest() {
         return new LeaveGroupRequest("group1", "consumer1");
     }
 
-    private AbstractRequestResponse createLeaveGroupResponse() {
+    private LeaveGroupResponse createLeaveGroupResponse() {
         return new LeaveGroupResponse(Errors.NONE.code());
     }
 
     @SuppressWarnings("deprecation")
-    private AbstractRequest createListOffsetRequest(int version) {
+    private ListOffsetRequest createListOffsetRequest(int version) {
         if (version == 0) {
             Map<TopicPartition, ListOffsetRequest.PartitionData> offsetData = new HashMap<>();
             offsetData.put(new TopicPartition("test", 0), new ListOffsetRequest.PartitionData(1000000L, 10));
@@ -324,7 +360,7 @@ public class RequestResponseTest {
     }
 
     @SuppressWarnings("deprecation")
-    private AbstractRequestResponse createListOffsetResponse(int version) {
+    private ListOffsetResponse createListOffsetResponse(int version) {
         if (version == 0) {
             Map<TopicPartition, ListOffsetResponse.PartitionData> responseData = new HashMap<>();
             responseData.put(new TopicPartition("test", 0), new ListOffsetResponse.PartitionData(Errors.NONE.code(), Arrays.asList(100L)));
@@ -338,11 +374,11 @@ public class RequestResponseTest {
         }
     }
 
-    private AbstractRequest createMetadataRequest(List<String> topics) {
+    private MetadataRequest createMetadataRequest(List<String> topics) {
         return new MetadataRequest(topics);
     }
 
-    private AbstractRequestResponse createMetadataResponse(int version) {
+    private MetadataResponse createMetadataResponse(int version) {
         Node node = new Node(1, "host1", 1001);
         List<Node> replicas = Arrays.asList(node);
         List<Node> isr = Arrays.asList(node);
@@ -357,7 +393,7 @@ public class RequestResponseTest {
     }
 
     @SuppressWarnings("deprecation")
-    private AbstractRequest createOffsetCommitRequest(int version) {
+    private OffsetCommitRequest createOffsetCommitRequest(int version) {
         Map<TopicPartition, OffsetCommitRequest.PartitionData> commitData = new HashMap<>();
         commitData.put(new TopicPartition("test", 0), new OffsetCommitRequest.PartitionData(100, ""));
         commitData.put(new TopicPartition("test", 1), new OffsetCommitRequest.PartitionData(200, null));
@@ -371,47 +407,47 @@ public class RequestResponseTest {
         throw new IllegalArgumentException("Unknown offset commit request version " + version);
     }
 
-    private AbstractRequestResponse createOffsetCommitResponse() {
+    private OffsetCommitResponse createOffsetCommitResponse() {
         Map<TopicPartition, Short> responseData = new HashMap<>();
         responseData.put(new TopicPartition("test", 0), Errors.NONE.code());
         return new OffsetCommitResponse(responseData);
     }
 
-    private AbstractRequest createOffsetFetchRequest() {
+    private OffsetFetchRequest createOffsetFetchRequest() {
         return new OffsetFetchRequest("group1", Arrays.asList(new TopicPartition("test11", 1)));
     }
 
-    private AbstractRequestResponse createOffsetFetchResponse() {
+    private OffsetFetchResponse createOffsetFetchResponse() {
         Map<TopicPartition, OffsetFetchResponse.PartitionData> responseData = new HashMap<>();
         responseData.put(new TopicPartition("test", 0), new OffsetFetchResponse.PartitionData(100L, "", Errors.NONE.code()));
         responseData.put(new TopicPartition("test", 1), new OffsetFetchResponse.PartitionData(100L, null, Errors.NONE.code()));
         return new OffsetFetchResponse(responseData);
     }
 
-    private AbstractRequest createProduceRequest() {
-        Map<TopicPartition, ByteBuffer> produceData = new HashMap<>();
-        produceData.put(new TopicPartition("test", 0), ByteBuffer.allocate(10));
+    private ProduceRequest createProduceRequest() {
+        Map<TopicPartition, MemoryRecords> produceData = new HashMap<>();
+        produceData.put(new TopicPartition("test", 0), MemoryRecords.readableRecords(ByteBuffer.allocate(10)));
         return new ProduceRequest((short) 1, 5000, produceData);
     }
 
-    private AbstractRequestResponse createProduceResponse() {
+    private ProduceResponse createProduceResponse() {
         Map<TopicPartition, ProduceResponse.PartitionResponse> responseData = new HashMap<>();
         responseData.put(new TopicPartition("test", 0), new ProduceResponse.PartitionResponse(Errors.NONE.code(), 10000, Record.NO_TIMESTAMP));
         return new ProduceResponse(responseData, 0);
     }
 
-    private AbstractRequest createStopReplicaRequest(boolean deletePartitions) {
+    private StopReplicaRequest createStopReplicaRequest(boolean deletePartitions) {
         Set<TopicPartition> partitions = new HashSet<>(Arrays.asList(new TopicPartition("test", 0)));
         return new StopReplicaRequest(0, 1, deletePartitions, partitions);
     }
 
-    private AbstractRequestResponse createStopReplicaResponse() {
+    private StopReplicaResponse createStopReplicaResponse() {
         Map<TopicPartition, Short> responses = new HashMap<>();
         responses.put(new TopicPartition("test", 0), Errors.NONE.code());
         return new StopReplicaResponse(Errors.NONE.code(), responses);
     }
 
-    private AbstractRequest createControlledShutdownRequest() {
+    private ControlledShutdownRequest createControlledShutdownRequest() {
         return new ControlledShutdownRequest(10);
     }
 
@@ -423,7 +459,7 @@ public class RequestResponseTest {
         return new ControlledShutdownResponse(Errors.NONE.code(), topicPartitions);
     }
 
-    private AbstractRequest createLeaderAndIsrRequest() {
+    private LeaderAndIsrRequest createLeaderAndIsrRequest() {
         Map<TopicPartition, PartitionState> partitionStates = new HashMap<>();
         List<Integer> isr = Arrays.asList(1, 2);
         List<Integer> replicas = Arrays.asList(1, 2, 3, 4);
@@ -442,14 +478,14 @@ public class RequestResponseTest {
         return new LeaderAndIsrRequest(1, 10, partitionStates, leaders);
     }
 
-    private AbstractRequestResponse createLeaderAndIsrResponse() {
+    private LeaderAndIsrResponse createLeaderAndIsrResponse() {
         Map<TopicPartition, Short> responses = new HashMap<>();
         responses.put(new TopicPartition("test", 0), Errors.NONE.code());
         return new LeaderAndIsrResponse(Errors.NONE.code(), responses);
     }
 
     @SuppressWarnings("deprecation")
-    private AbstractRequest createUpdateMetadataRequest(int version, String rack) {
+    private UpdateMetadataRequest createUpdateMetadataRequest(int version, String rack) {
         Map<TopicPartition, PartitionState> partitionStates = new HashMap<>();
         List<Integer> isr = Arrays.asList(1, 2);
         List<Integer> replicas = Arrays.asList(1, 2, 3, 4);
@@ -482,28 +518,28 @@ public class RequestResponseTest {
         }
     }
 
-    private AbstractRequestResponse createUpdateMetadataResponse() {
+    private UpdateMetadataResponse createUpdateMetadataResponse() {
         return new UpdateMetadataResponse(Errors.NONE.code());
     }
 
-    private AbstractRequest createSaslHandshakeRequest() {
+    private SaslHandshakeRequest createSaslHandshakeRequest() {
         return new SaslHandshakeRequest("PLAIN");
     }
 
-    private AbstractRequestResponse createSaslHandshakeResponse() {
+    private SaslHandshakeResponse createSaslHandshakeResponse() {
         return new SaslHandshakeResponse(Errors.NONE.code(), Collections.singletonList("GSSAPI"));
     }
 
-    private AbstractRequest createApiVersionRequest() {
+    private ApiVersionsRequest createApiVersionRequest() {
         return new ApiVersionsRequest();
     }
 
-    private AbstractRequestResponse createApiVersionResponse() {
+    private ApiVersionsResponse createApiVersionResponse() {
         List<ApiVersionsResponse.ApiVersion> apiVersions = Arrays.asList(new ApiVersionsResponse.ApiVersion((short) 0, (short) 0, (short) 2));
         return new ApiVersionsResponse(Errors.NONE.code(), apiVersions);
     }
 
-    private AbstractRequest createCreateTopicRequest() {
+    private CreateTopicsRequest createCreateTopicRequest() {
         CreateTopicsRequest.TopicDetails request1 = new CreateTopicsRequest.TopicDetails(3, (short) 5);
 
         Map<Integer, List<Integer>> replicaAssignments = new HashMap<>();
@@ -521,21 +557,65 @@ public class RequestResponseTest {
         return new CreateTopicsRequest(request, 0);
     }
 
-    private AbstractRequestResponse createCreateTopicResponse() {
+    private CreateTopicsResponse createCreateTopicResponse() {
         Map<String, Errors> errors = new HashMap<>();
         errors.put("t1", Errors.INVALID_TOPIC_EXCEPTION);
         errors.put("t2", Errors.LEADER_NOT_AVAILABLE);
         return new CreateTopicsResponse(errors);
     }
 
-    private AbstractRequest createDeleteTopicsRequest() {
+    private DeleteTopicsRequest createDeleteTopicsRequest() {
         return new DeleteTopicsRequest(new HashSet<>(Arrays.asList("my_t1", "my_t2")), 10000);
     }
 
-    private AbstractRequestResponse createDeleteTopicsResponse() {
+    private DeleteTopicsResponse createDeleteTopicsResponse() {
         Map<String, Errors> errors = new HashMap<>();
         errors.put("t1", Errors.INVALID_TOPIC_EXCEPTION);
         errors.put("t2", Errors.TOPIC_AUTHORIZATION_FAILED);
         return new DeleteTopicsResponse(errors);
     }
+
+    private static class ByteBufferChannel implements GatheringByteChannel {
+        private final ByteBuffer buf;
+        private boolean closed = false;
+
+        private ByteBufferChannel(long size) {
+            this.buf = ByteBuffer.allocate(Long.valueOf(size).intValue());
+        }
+
+        @Override
+        public long write(ByteBuffer[] srcs, int offset, int length) throws IOException {
+            int position = buf.position();
+            for (int i = 0; i < length; i++) {
+                ByteBuffer src = srcs[i].duplicate();
+                if (i == 0)
+                    src.position(offset);
+                buf.put(src);
+            }
+            return buf.position() - position;
+        }
+
+        @Override
+        public long write(ByteBuffer[] srcs) throws IOException {
+            return write(srcs, 0, srcs.length);
+        }
+
+        @Override
+        public int write(ByteBuffer src) throws IOException {
+            int position = buf.position();
+            buf.put(src);
+            return buf.position() - position;
+        }
+
+        @Override
+        public boolean isOpen() {
+            return !closed;
+        }
+
+        @Override
+        public void close() throws IOException {
+            buf.flip();
+            closed = true;
+        }
+    }
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/3b4c3479/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java b/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java
index 97fe3d8..27c5695 100644
--- a/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java
@@ -12,18 +12,6 @@
  */
 package org.apache.kafka.common.security.authenticator;
 
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.Random;
-import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.nio.ByteBuffer;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
-
 import org.apache.kafka.clients.NetworkClient;
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.config.SaslConfigs;
@@ -36,17 +24,17 @@ import org.apache.kafka.common.network.NetworkSend;
 import org.apache.kafka.common.network.NetworkTestUtils;
 import org.apache.kafka.common.network.NioEchoServer;
 import org.apache.kafka.common.network.Selector;
+import org.apache.kafka.common.network.Send;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.protocol.Protocol;
 import org.apache.kafka.common.protocol.SecurityProtocol;
-import org.apache.kafka.common.protocol.types.Struct;
-import org.apache.kafka.common.requests.AbstractRequestResponse;
+import org.apache.kafka.common.requests.AbstractRequest;
+import org.apache.kafka.common.requests.AbstractResponse;
 import org.apache.kafka.common.requests.ApiVersionsRequest;
 import org.apache.kafka.common.requests.ApiVersionsResponse;
 import org.apache.kafka.common.requests.MetadataRequest;
 import org.apache.kafka.common.requests.RequestHeader;
-import org.apache.kafka.common.requests.RequestSend;
 import org.apache.kafka.common.requests.ResponseHeader;
 import org.apache.kafka.common.requests.SaslHandshakeRequest;
 import org.apache.kafka.common.requests.SaslHandshakeResponse;
@@ -55,6 +43,18 @@ import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
 /**
  * Tests for the Sasl authenticator. These use a test harness that runs a simple socket server that echos back responses.
  */
@@ -261,7 +261,8 @@ public class SaslAuthenticatorTest {
         String node = "1";
         createClientConnection(SecurityProtocol.PLAINTEXT, node);
         RequestHeader header = new RequestHeader(ApiKeys.API_VERSIONS.id, Short.MAX_VALUE, "someclient", 1);
-        selector.send(new NetworkSend(node, RequestSend.serialize(header, new ApiVersionsRequest().toStruct())));
+        ApiVersionsRequest request = new ApiVersionsRequest();
+        selector.send(request.toSend(node, header));
         ByteBuffer responseBuffer = waitForResponse();
         ResponseHeader.parse(responseBuffer);
         ApiVersionsResponse response = ApiVersionsResponse.parse(responseBuffer);
@@ -291,8 +292,9 @@ public class SaslAuthenticatorTest {
         // Send ApiVersionsRequest and validate error response.
         String node1 = "invalid1";
         createClientConnection(SecurityProtocol.PLAINTEXT, node1);
+        SaslHandshakeRequest request = new SaslHandshakeRequest("PLAIN");
         RequestHeader header = new RequestHeader(ApiKeys.SASL_HANDSHAKE.id, Short.MAX_VALUE, "someclient", 2);
-        selector.send(new NetworkSend(node1, RequestSend.serialize(header, new SaslHandshakeRequest("PLAIN").toStruct())));
+        selector.send(request.toSend(node1, header));
         NetworkTestUtils.waitForChannelClose(selector, node1);
         selector.close();
 
@@ -355,8 +357,9 @@ public class SaslAuthenticatorTest {
         createClientConnection(SecurityProtocol.PLAINTEXT, node1);
         sendHandshakeRequestReceiveResponse(node1);
 
+        ApiVersionsRequest request = new ApiVersionsRequest();
         RequestHeader versionsHeader = new RequestHeader(ApiKeys.API_VERSIONS.id, "someclient", 2);
-        selector.send(new NetworkSend(node1, RequestSend.serialize(versionsHeader, new ApiVersionsRequest().toStruct())));
+        selector.send(request.toSend(node1, versionsHeader));
         NetworkTestUtils.waitForChannelClose(selector, node1);
         selector.close();
 
@@ -420,7 +423,7 @@ public class SaslAuthenticatorTest {
         createClientConnection(SecurityProtocol.PLAINTEXT, node1);
         RequestHeader metadataRequestHeader1 = new RequestHeader(ApiKeys.METADATA.id, "someclient", 1);
         MetadataRequest metadataRequest1 = new MetadataRequest(Collections.singletonList("sometopic"));
-        selector.send(new NetworkSend(node1, RequestSend.serialize(metadataRequestHeader1, metadataRequest1.toStruct())));
+        selector.send(metadataRequest1.toSend(node1, metadataRequestHeader1));
         NetworkTestUtils.waitForChannelClose(selector, node1);
         selector.close();
 
@@ -433,7 +436,7 @@ public class SaslAuthenticatorTest {
         sendHandshakeRequestReceiveResponse(node2);
         RequestHeader metadataRequestHeader2 = new RequestHeader(ApiKeys.METADATA.id, "someclient", 2);
         MetadataRequest metadataRequest2 = new MetadataRequest(Collections.singletonList("sometopic"));
-        selector.send(new NetworkSend(node2, RequestSend.serialize(metadataRequestHeader2, metadataRequest2.toStruct())));
+        selector.send(metadataRequest2.toSend(node2, metadataRequestHeader2));
         NetworkTestUtils.waitForChannelClose(selector, node2);
         selector.close();
 
@@ -581,25 +584,24 @@ public class SaslAuthenticatorTest {
         selector = null;
     }
 
-    private Struct sendKafkaRequestReceiveResponse(String node, ApiKeys apiKey, AbstractRequestResponse request) throws IOException {
+    private AbstractResponse sendKafkaRequestReceiveResponse(String node, ApiKeys apiKey, AbstractRequest request) throws IOException {
         RequestHeader header = new RequestHeader(apiKey.id, "someclient", 1);
-        selector.send(new NetworkSend(node, RequestSend.serialize(header, request.toStruct())));
+        Send send = request.toSend(node, header);
+        selector.send(send);
         ByteBuffer responseBuffer = waitForResponse();
         return NetworkClient.parseResponse(responseBuffer, header);
     }
 
     private SaslHandshakeResponse sendHandshakeRequestReceiveResponse(String node) throws Exception {
         SaslHandshakeRequest handshakeRequest = new SaslHandshakeRequest("PLAIN");
-        Struct responseStruct = sendKafkaRequestReceiveResponse(node, ApiKeys.SASL_HANDSHAKE, handshakeRequest);
-        SaslHandshakeResponse response = new SaslHandshakeResponse(responseStruct);
+        SaslHandshakeResponse response = (SaslHandshakeResponse) sendKafkaRequestReceiveResponse(node, ApiKeys.SASL_HANDSHAKE, handshakeRequest);
         assertEquals(Errors.NONE.code(), response.errorCode());
         return response;
     }
 
     private ApiVersionsResponse sendVersionRequestReceiveResponse(String node) throws Exception {
         ApiVersionsRequest handshakeRequest = new ApiVersionsRequest();
-        Struct responseStruct = sendKafkaRequestReceiveResponse(node, ApiKeys.API_VERSIONS, handshakeRequest);
-        ApiVersionsResponse response = new ApiVersionsResponse(responseStruct);
+        ApiVersionsResponse response =  (ApiVersionsResponse) sendKafkaRequestReceiveResponse(node, ApiKeys.API_VERSIONS, handshakeRequest);
         assertEquals(Errors.NONE.code(), response.errorCode());
         return response;
     }

http://git-wip-us.apache.org/repos/asf/kafka/blob/3b4c3479/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinatorTest.java
----------------------------------------------------------------------
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinatorTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinatorTest.java
index 3bfa83f..3aff8f2 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinatorTest.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinatorTest.java
@@ -16,7 +16,6 @@
  **/
 package org.apache.kafka.connect.runtime.distributed;
 
-import org.apache.kafka.clients.ClientRequest;
 import org.apache.kafka.clients.Metadata;
 import org.apache.kafka.clients.MockClient;
 import org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient;
@@ -24,7 +23,7 @@ import org.apache.kafka.common.Cluster;
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.metrics.Metrics;
 import org.apache.kafka.common.protocol.Errors;
-import org.apache.kafka.common.protocol.types.Struct;
+import org.apache.kafka.common.requests.AbstractRequest;
 import org.apache.kafka.common.requests.GroupCoordinatorResponse;
 import org.apache.kafka.common.requests.JoinGroupRequest.ProtocolMetadata;
 import org.apache.kafka.common.requests.JoinGroupResponse;
@@ -172,7 +171,7 @@ public class WorkerCoordinatorTest {
 
         final String consumerId = "leader";
 
-        client.prepareResponse(groupMetadataResponse(node, Errors.NONE.code()));
+        client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code()));
         coordinator.ensureCoordinatorReady();
 
         // normal join group
@@ -182,8 +181,8 @@ public class WorkerCoordinatorTest {
         client.prepareResponse(joinGroupLeaderResponse(1, consumerId, memberConfigOffsets, Errors.NONE.code()));
         client.prepareResponse(new MockClient.RequestMatcher() {
             @Override
-            public boolean matches(ClientRequest request) {
-                SyncGroupRequest sync = new SyncGroupRequest(request.request().body());
+            public boolean matches(AbstractRequest body) {
+                SyncGroupRequest sync = (SyncGroupRequest) body;
                 return sync.memberId().equals(consumerId) &&
                         sync.generationId() == 1 &&
                         sync.groupAssignment().containsKey(consumerId);
@@ -212,15 +211,15 @@ public class WorkerCoordinatorTest {
 
         final String memberId = "member";
 
-        client.prepareResponse(groupMetadataResponse(node, Errors.NONE.code()));
+        client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code()));
         coordinator.ensureCoordinatorReady();
 
         // normal join group
         client.prepareResponse(joinGroupFollowerResponse(1, memberId, "leader", Errors.NONE.code()));
         client.prepareResponse(new MockClient.RequestMatcher() {
             @Override
-            public boolean matches(ClientRequest request) {
-                SyncGroupRequest sync = new SyncGroupRequest(request.request().body());
+            public boolean matches(AbstractRequest body) {
+                SyncGroupRequest sync = (SyncGroupRequest) body;
                 return sync.memberId().equals(memberId) &&
                         sync.generationId() == 1 &&
                         sync.groupAssignment().isEmpty();
@@ -253,15 +252,15 @@ public class WorkerCoordinatorTest {
 
         final String memberId = "member";
 
-        client.prepareResponse(groupMetadataResponse(node, Errors.NONE.code()));
+        client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code()));
         coordinator.ensureCoordinatorReady();
 
         // config mismatch results in assignment error
         client.prepareResponse(joinGroupFollowerResponse(1, memberId, "leader", Errors.NONE.code()));
         MockClient.RequestMatcher matcher = new MockClient.RequestMatcher() {
             @Override
-            public boolean matches(ClientRequest request) {
-                SyncGroupRequest sync = new SyncGroupRequest(request.request().body());
+            public boolean matches(AbstractRequest body) {
+                SyncGroupRequest sync = (SyncGroupRequest) body;
                 return sync.memberId().equals(memberId) &&
                         sync.generationId() == 1 &&
                         sync.groupAssignment().isEmpty();
@@ -284,7 +283,7 @@ public class WorkerCoordinatorTest {
 
         PowerMock.replayAll();
 
-        client.prepareResponse(groupMetadataResponse(node, Errors.NONE.code()));
+        client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code()));
         coordinator.ensureCoordinatorReady();
 
         // join the group once
@@ -392,12 +391,11 @@ public class WorkerCoordinatorTest {
     }
 
 
-    private Struct groupMetadataResponse(Node node, short error) {
-        GroupCoordinatorResponse response = new GroupCoordinatorResponse(error, node);
-        return response.toStruct();
+    private GroupCoordinatorResponse groupCoordinatorResponse(Node node, short error) {
+        return new GroupCoordinatorResponse(error, node);
     }
 
-    private Struct joinGroupLeaderResponse(int generationId, String memberId,
+    private JoinGroupResponse joinGroupLeaderResponse(int generationId, String memberId,
                                            Map<String, Long> configOffsets, short error) {
         Map<String, ByteBuffer> metadata = new HashMap<>();
         for (Map.Entry<String, Long> configStateEntry : configOffsets.entrySet()) {
@@ -407,22 +405,21 @@ public class WorkerCoordinatorTest {
             ByteBuffer buf = ConnectProtocol.serializeMetadata(new ConnectProtocol.WorkerState(memberUrl, configOffset));
             metadata.put(configStateEntry.getKey(), buf);
         }
-        return new JoinGroupResponse(error, generationId, WorkerCoordinator.DEFAULT_SUBPROTOCOL, memberId, memberId, metadata).toStruct();
+        return new JoinGroupResponse(error, generationId, WorkerCoordinator.DEFAULT_SUBPROTOCOL, memberId, memberId, metadata);
     }
 
-    private Struct joinGroupFollowerResponse(int generationId, String memberId, String leaderId, short error) {
+    private JoinGroupResponse joinGroupFollowerResponse(int generationId, String memberId, String leaderId, short error) {
         return new JoinGroupResponse(error, generationId, WorkerCoordinator.DEFAULT_SUBPROTOCOL, memberId, leaderId,
-                Collections.<String, ByteBuffer>emptyMap()).toStruct();
+                Collections.<String, ByteBuffer>emptyMap());
     }
 
-    private Struct syncGroupResponse(short assignmentError, String leader, long configOffset, List<String> connectorIds,
+    private SyncGroupResponse syncGroupResponse(short assignmentError, String leader, long configOffset, List<String> connectorIds,
                                      List<ConnectorTaskId> taskIds, short error) {
         ConnectProtocol.Assignment assignment = new ConnectProtocol.Assignment(assignmentError, leader, LEADER_URL, configOffset, connectorIds, taskIds);
         ByteBuffer buf = ConnectProtocol.serializeAssignment(assignment);
-        return new SyncGroupResponse(error, buf).toStruct();
+        return new SyncGroupResponse(error, buf);
     }
 
-
     private static class MockRebalanceListener implements WorkerRebalanceListener {
         public ConnectProtocol.Assignment assignment = null;
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/3b4c3479/core/src/main/scala/kafka/admin/AdminClient.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/admin/AdminClient.scala b/core/src/main/scala/kafka/admin/AdminClient.scala
index 1179557..592fecf 100644
--- a/core/src/main/scala/kafka/admin/AdminClient.scala
+++ b/core/src/main/scala/kafka/admin/AdminClient.scala
@@ -42,7 +42,7 @@ class AdminClient(val time: Time,
 
   private def send(target: Node,
                    api: ApiKeys,
-                   request: AbstractRequest): Struct = {
+                   request: AbstractRequest): AbstractResponse = {
     var future: RequestFuture[ClientResponse] = null
 
     future = client.send(target, api, request)
@@ -54,9 +54,8 @@ class AdminClient(val time: Time,
       throw future.exception()
   }
 
-  private def sendAnyNode(api: ApiKeys, request: AbstractRequest): Struct = {
-    bootstrapBrokers.foreach {
-      case broker =>
+  private def sendAnyNode(api: ApiKeys, request: AbstractRequest): AbstractResponse = {
+    bootstrapBrokers.foreach { broker =>
         try {
           return send(broker, api, request)
         } catch {
@@ -69,23 +68,20 @@ class AdminClient(val time: Time,
 
   private def findCoordinator(groupId: String): Node = {
     val request = new GroupCoordinatorRequest(groupId)
-    val responseBody = sendAnyNode(ApiKeys.GROUP_COORDINATOR, request)
-    val response = new GroupCoordinatorResponse(responseBody)
+    val response = sendAnyNode(ApiKeys.GROUP_COORDINATOR, request).asInstanceOf[GroupCoordinatorResponse]
     Errors.forCode(response.errorCode()).maybeThrow()
     response.node()
   }
 
   def listGroups(node: Node): List[GroupOverview] = {
-    val responseBody = send(node, ApiKeys.LIST_GROUPS, new ListGroupsRequest())
-    val response = new ListGroupsResponse(responseBody)
+    val response = send(node, ApiKeys.LIST_GROUPS, new ListGroupsRequest()).asInstanceOf[ListGroupsResponse]
     Errors.forCode(response.errorCode()).maybeThrow()
     response.groups().map(group => GroupOverview(group.groupId(), group.protocolType())).toList
   }
 
   private def findAllBrokers(): List[Node] = {
     val request = new MetadataRequest(List[String]())
-    val responseBody = sendAnyNode(ApiKeys.METADATA, request)
-    val response = new MetadataResponse(responseBody)
+    val response = sendAnyNode(ApiKeys.METADATA, request).asInstanceOf[MetadataResponse]
     val errors = response.errors()
     if (!errors.isEmpty)
       debug(s"Metadata request contained errors: $errors")
@@ -140,7 +136,7 @@ class AdminClient(val time: Time,
   def describeConsumerGroup(groupId: String): ConsumerGroupSummary = {
     val coordinator = findCoordinator(groupId)
     val responseBody = send(coordinator, ApiKeys.DESCRIBE_GROUPS, new DescribeGroupsRequest(List(groupId).asJava))
-    val response = new DescribeGroupsResponse(responseBody)
+    val response = responseBody.asInstanceOf[DescribeGroupsResponse]
     val metadata = response.groups.get(groupId)
     if (metadata == null)
       throw new KafkaException(s"Response from broker contained no metadata for group $groupId")

http://git-wip-us.apache.org/repos/asf/kafka/blob/3b4c3479/core/src/main/scala/kafka/api/FetchRequest.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/api/FetchRequest.scala b/core/src/main/scala/kafka/api/FetchRequest.scala
index 57e99c1..00897db 100644
--- a/core/src/main/scala/kafka/api/FetchRequest.scala
+++ b/core/src/main/scala/kafka/api/FetchRequest.scala
@@ -199,8 +199,7 @@ case class FetchRequest(versionId: Short = FetchRequest.CurrentVersion,
     val fetchResponsePartitionData = requestInfo.map { case (topicAndPartition, _) =>
       (topicAndPartition, FetchResponsePartitionData(Errors.forException(e).code, -1, MessageSet.Empty))
     }
-    val fetchRequest = request.requestObj.asInstanceOf[FetchRequest]
-    val errorResponse = FetchResponse(correlationId, fetchResponsePartitionData, fetchRequest.versionId)
+    val errorResponse = FetchResponse(correlationId, fetchResponsePartitionData, request.header.apiVersion)
     // Magic value does not matter here because the message set is empty
     requestChannel.sendResponse(new RequestChannel.Response(request, new FetchResponseSend(request.connectionId, errorResponse)))
   }

http://git-wip-us.apache.org/repos/asf/kafka/blob/3b4c3479/core/src/main/scala/kafka/api/FetchResponse.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/api/FetchResponse.scala b/core/src/main/scala/kafka/api/FetchResponse.scala
index d99bbcd..d31d4ba 100644
--- a/core/src/main/scala/kafka/api/FetchResponse.scala
+++ b/core/src/main/scala/kafka/api/FetchResponse.scala
@@ -76,7 +76,8 @@ class PartitionDataSend(val partitionId: Int,
       written += channel.write(buffer)
     if (!buffer.hasRemaining) {
       if (messagesSentSize < messageSize) {
-        val bytesSent = partitionData.messages.writeTo(channel, messagesSentSize, messageSize - messagesSentSize)
+        val records = partitionData.messages.asRecords
+        val bytesSent = records.writeTo(channel, messagesSentSize, messageSize - messagesSentSize).toInt
         messagesSentSize += bytesSent
         written += bytesSent
       }

http://git-wip-us.apache.org/repos/asf/kafka/blob/3b4c3479/core/src/main/scala/kafka/api/GenericRequestAndHeader.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/api/GenericRequestAndHeader.scala b/core/src/main/scala/kafka/api/GenericRequestAndHeader.scala
index cb5b95e..3783c29 100644
--- a/core/src/main/scala/kafka/api/GenericRequestAndHeader.scala
+++ b/core/src/main/scala/kafka/api/GenericRequestAndHeader.scala
@@ -14,13 +14,14 @@
 package kafka.api
 
 import java.nio.ByteBuffer
-import org.apache.kafka.common.requests.AbstractRequestResponse
+
 import kafka.api.ApiUtils._
+import org.apache.kafka.common.requests.AbstractResponse
 
 private[kafka] abstract class GenericRequestAndHeader(val versionId: Short,
                                                       val correlationId: Int,
                                                       val clientId: String,
-                                                      val body: AbstractRequestResponse,
+                                                      val body: AbstractResponse,
                                                       val name: String,
                                                       override val requestId: Option[Short] = None)
   extends RequestOrResponse(requestId) {

http://git-wip-us.apache.org/repos/asf/kafka/blob/3b4c3479/core/src/main/scala/kafka/api/GenericResponseAndHeader.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/api/GenericResponseAndHeader.scala b/core/src/main/scala/kafka/api/GenericResponseAndHeader.scala
index 2835fb6..be0c080 100644
--- a/core/src/main/scala/kafka/api/GenericResponseAndHeader.scala
+++ b/core/src/main/scala/kafka/api/GenericResponseAndHeader.scala
@@ -14,10 +14,11 @@
 package kafka.api
 
 import java.nio.ByteBuffer
-import org.apache.kafka.common.requests.AbstractRequestResponse
+
+import org.apache.kafka.common.requests.AbstractResponse
 
 private[kafka] abstract class GenericResponseAndHeader(val correlationId: Int,
-                                                       val body: AbstractRequestResponse,
+                                                       val body: AbstractResponse,
                                                        val name: String,
                                                        override val requestId: Option[Short] = None)
   extends RequestOrResponse(requestId) {

http://git-wip-us.apache.org/repos/asf/kafka/blob/3b4c3479/core/src/main/scala/kafka/api/ProducerRequest.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/api/ProducerRequest.scala b/core/src/main/scala/kafka/api/ProducerRequest.scala
index 3ca7bd7..f6e4475 100644
--- a/core/src/main/scala/kafka/api/ProducerRequest.scala
+++ b/core/src/main/scala/kafka/api/ProducerRequest.scala
@@ -128,8 +128,8 @@ case class ProducerRequest(versionId: Short = ProducerRequest.CurrentVersion,
     describe(true)
   }
 
-  override  def handleError(e: Throwable, requestChannel: RequestChannel, request: RequestChannel.Request): Unit = {
-    if(request.requestObj.asInstanceOf[ProducerRequest].requiredAcks == 0) {
+  override def handleError(e: Throwable, requestChannel: RequestChannel, request: RequestChannel.Request): Unit = {
+    if(request.body.asInstanceOf[org.apache.kafka.common.requests.ProduceRequest].acks == 0) {
         requestChannel.closeConnection(request.processor, request)
     }
     else {

http://git-wip-us.apache.org/repos/asf/kafka/blob/3b4c3479/core/src/main/scala/kafka/api/RequestOrResponse.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/api/RequestOrResponse.scala b/core/src/main/scala/kafka/api/RequestOrResponse.scala
index 65b37fd..d013047 100644
--- a/core/src/main/scala/kafka/api/RequestOrResponse.scala
+++ b/core/src/main/scala/kafka/api/RequestOrResponse.scala
@@ -42,6 +42,6 @@ abstract class RequestOrResponse(val requestId: Option[Short] = None) extends Lo
   *  This API has no meaning for a Response object.
    * @param details If this is false, omit the parts of the request description that are proportional to the number of
    *                topics or partitions. This is mainly to control the amount of request logging. */
-  def describe(details: Boolean):String
+  def describe(details: Boolean): String
 }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/3b4c3479/core/src/main/scala/kafka/controller/ControllerChannelManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/controller/ControllerChannelManager.scala b/core/src/main/scala/kafka/controller/ControllerChannelManager.scala
index 03cd98c..1935ea2 100755
--- a/core/src/main/scala/kafka/controller/ControllerChannelManager.scala
+++ b/core/src/main/scala/kafka/controller/ControllerChannelManager.scala
@@ -42,7 +42,7 @@ class ControllerChannelManager(controllerContext: ControllerContext, config: Kaf
   private val brokerLock = new Object
   this.logIdent = "[Channel manager on controller " + config.brokerId + "]: "
 
-  controllerContext.liveBrokers.foreach(addNewBroker(_))
+  controllerContext.liveBrokers.foreach(addNewBroker)
 
   def startup() = {
     brokerLock synchronized {
@@ -56,7 +56,7 @@ class ControllerChannelManager(controllerContext: ControllerContext, config: Kaf
     }
   }
 
-  def sendRequest(brokerId: Int, apiKey: ApiKeys, apiVersion: Option[Short], request: AbstractRequest, callback: AbstractRequestResponse => Unit = null) {
+  def sendRequest(brokerId: Int, apiKey: ApiKeys, apiVersion: Option[Short], request: AbstractRequest, callback: AbstractResponse => Unit = null) {
     brokerLock synchronized {
       val stateInfoOpt = brokerStateInfo.get(brokerId)
       stateInfoOpt match {
@@ -149,7 +149,7 @@ class ControllerChannelManager(controllerContext: ControllerContext, config: Kaf
   }
 }
 
-case class QueueItem(apiKey: ApiKeys, apiVersion: Option[Short], request: AbstractRequest, callback: AbstractRequestResponse => Unit)
+case class QueueItem(apiKey: ApiKeys, apiVersion: Option[Short], request: AbstractRequest, callback: AbstractResponse => Unit)
 
 class RequestSendThread(val controllerId: Int,
                         val controllerContext: ControllerContext,
@@ -185,28 +185,27 @@ class RequestSendThread(val controllerId: Int,
             }
             else {
               val requestHeader = apiVersion.fold(networkClient.nextRequestHeader(apiKey))(networkClient.nextRequestHeader(apiKey, _))
-              val send = new RequestSend(brokerNode.idString, requestHeader, request.toStruct)
-              val clientRequest = new ClientRequest(time.milliseconds(), true, send, null)
-              clientResponse = networkClient.blockingSendAndReceive(clientRequest)(time)
+              val clientRequest = new ClientRequest(brokerNode.idString, time.milliseconds(), true, requestHeader, request, null)
+              clientResponse = networkClient.blockingSendAndReceive(clientRequest, request)(time)
               isSendSuccessful = true
             }
           } catch {
             case e: Throwable => // if the send was not successful, reconnect to broker and resend the message
               warn(("Controller %d epoch %d fails to send request %s to broker %s. " +
                 "Reconnecting to broker.").format(controllerId, controllerContext.epoch,
-                  request.toString, brokerNode.toString()), e)
+                  request.toString, brokerNode.toString), e)
               networkClient.close(brokerNode.idString)
               isSendSuccessful = false
               backoff()
           }
         }
         if (clientResponse != null) {
-          val response = ApiKeys.forId(clientResponse.request.request.header.apiKey) match {
-            case ApiKeys.LEADER_AND_ISR => new LeaderAndIsrResponse(clientResponse.responseBody)
-            case ApiKeys.STOP_REPLICA => new StopReplicaResponse(clientResponse.responseBody)
-            case ApiKeys.UPDATE_METADATA_KEY => new UpdateMetadataResponse(clientResponse.responseBody)
-            case apiKey => throw new KafkaException(s"Unexpected apiKey received: $apiKey")
-          }
+          val api = ApiKeys.forId(clientResponse.requestHeader.apiKey)
+          if (api != ApiKeys.LEADER_AND_ISR && api != ApiKeys.STOP_REPLICA && api != ApiKeys.UPDATE_METADATA_KEY)
+            throw new KafkaException(s"Unexpected apiKey received: $apiKey")
+
+          val response = clientResponse.responseBody
+
           stateChangeLogger.trace("Controller %d epoch %d received response %s for a request sent to broker %s"
             .format(controllerId, controllerContext.epoch, response.toString, brokerNode.toString))
 
@@ -217,7 +216,7 @@ class RequestSendThread(val controllerId: Int,
       }
     } catch {
       case e: Throwable =>
-        error("Controller %d fails to send a request to broker %s".format(controllerId, brokerNode.toString()), e)
+        error("Controller %d fails to send a request to broker %s".format(controllerId, brokerNode.toString), e)
         // If there is any socket error (eg, socket timeout), the connection is no longer usable and needs to be recreated.
         networkClient.close(brokerNode.idString)
     }
@@ -230,13 +229,13 @@ class RequestSendThread(val controllerId: Int,
         if (!networkClient.blockingReady(brokerNode, socketTimeoutMs)(time))
           throw new SocketTimeoutException(s"Failed to connect within $socketTimeoutMs ms")
 
-        info("Controller %d connected to %s for sending state change requests".format(controllerId, brokerNode.toString()))
+        info("Controller %d connected to %s for sending state change requests".format(controllerId, brokerNode.toString))
       }
 
       true
     } catch {
       case e: Throwable =>
-        warn("Controller %d's connection to broker %s was unsuccessful".format(controllerId, brokerNode.toString()), e)
+        warn("Controller %d's connection to broker %s was unsuccessful".format(controllerId, brokerNode.toString), e)
         networkClient.close(brokerNode.idString)
         false
     }
@@ -273,7 +272,7 @@ class ControllerBrokerRequestBatch(controller: KafkaController) extends  Logging
 
   def addLeaderAndIsrRequestForBrokers(brokerIds: Seq[Int], topic: String, partition: Int,
                                        leaderIsrAndControllerEpoch: LeaderIsrAndControllerEpoch,
-                                       replicas: Seq[Int], callback: AbstractRequestResponse => Unit = null) {
+                                       replicas: Seq[Int], callback: AbstractResponse => Unit = null) {
     val topicPartition = new TopicPartition(topic, partition)
 
     brokerIds.filter(_ >= 0).foreach { brokerId =>
@@ -286,13 +285,13 @@ class ControllerBrokerRequestBatch(controller: KafkaController) extends  Logging
   }
 
   def addStopReplicaRequestForBrokers(brokerIds: Seq[Int], topic: String, partition: Int, deletePartition: Boolean,
-                                      callback: (AbstractRequestResponse, Int) => Unit = null) {
+                                      callback: (AbstractResponse, Int) => Unit = null) {
     brokerIds.filter(b => b >= 0).foreach { brokerId =>
       stopReplicaRequestMap.getOrElseUpdate(brokerId, Seq.empty[StopReplicaRequestInfo])
       val v = stopReplicaRequestMap(brokerId)
       if(callback != null)
         stopReplicaRequestMap(brokerId) = v :+ StopReplicaRequestInfo(PartitionAndReplica(topic, partition, brokerId),
-          deletePartition, (r: AbstractRequestResponse) => callback(r, brokerId))
+          deletePartition, (r: AbstractResponse) => callback(r, brokerId))
       else
         stopReplicaRequestMap(brokerId) = v :+ StopReplicaRequestInfo(PartitionAndReplica(topic, partition, brokerId),
           deletePartition)
@@ -302,7 +301,7 @@ class ControllerBrokerRequestBatch(controller: KafkaController) extends  Logging
   /** Send UpdateMetadataRequest to the given brokers for the given partitions and partitions that are being deleted */
   def addUpdateMetadataRequestForBrokers(brokerIds: Seq[Int],
                                          partitions: collection.Set[TopicAndPartition] = Set.empty[TopicAndPartition],
-                                         callback: AbstractRequestResponse => Unit = null) {
+                                         callback: AbstractResponse => Unit = null) {
     def updateMetadataRequestMapFor(partition: TopicAndPartition, beingDeleted: Boolean) {
       val leaderIsrAndControllerEpochOpt = controllerContext.partitionLeadershipInfo.get(partition)
       leaderIsrAndControllerEpochOpt match {
@@ -444,29 +443,29 @@ case class ControllerBrokerStateInfo(networkClient: NetworkClient,
                                      messageQueue: BlockingQueue[QueueItem],
                                      requestSendThread: RequestSendThread)
 
-case class StopReplicaRequestInfo(replica: PartitionAndReplica, deletePartition: Boolean, callback: AbstractRequestResponse => Unit = null)
+case class StopReplicaRequestInfo(replica: PartitionAndReplica, deletePartition: Boolean, callback: AbstractResponse => Unit = null)
 
-class Callbacks private (var leaderAndIsrResponseCallback: AbstractRequestResponse => Unit = null,
-                         var updateMetadataResponseCallback: AbstractRequestResponse => Unit = null,
-                         var stopReplicaResponseCallback: (AbstractRequestResponse, Int) => Unit = null)
+class Callbacks private (var leaderAndIsrResponseCallback: AbstractResponse => Unit = null,
+                         var updateMetadataResponseCallback: AbstractResponse => Unit = null,
+                         var stopReplicaResponseCallback: (AbstractResponse, Int) => Unit = null)
 
 object Callbacks {
   class CallbackBuilder {
-    var leaderAndIsrResponseCbk: AbstractRequestResponse => Unit = null
-    var updateMetadataResponseCbk: AbstractRequestResponse => Unit = null
-    var stopReplicaResponseCbk: (AbstractRequestResponse, Int) => Unit = null
+    var leaderAndIsrResponseCbk: AbstractResponse => Unit = null
+    var updateMetadataResponseCbk: AbstractResponse => Unit = null
+    var stopReplicaResponseCbk: (AbstractResponse, Int) => Unit = null
 
-    def leaderAndIsrCallback(cbk: AbstractRequestResponse => Unit): CallbackBuilder = {
+    def leaderAndIsrCallback(cbk: AbstractResponse => Unit): CallbackBuilder = {
       leaderAndIsrResponseCbk = cbk
       this
     }
 
-    def updateMetadataCallback(cbk: AbstractRequestResponse => Unit): CallbackBuilder = {
+    def updateMetadataCallback(cbk: AbstractResponse => Unit): CallbackBuilder = {
       updateMetadataResponseCbk = cbk
       this
     }
 
-    def stopReplicaCallback(cbk: (AbstractRequestResponse, Int) => Unit): CallbackBuilder = {
+    def stopReplicaCallback(cbk: (AbstractResponse, Int) => Unit): CallbackBuilder = {
       stopReplicaResponseCbk = cbk
       this
     }

http://git-wip-us.apache.org/repos/asf/kafka/blob/3b4c3479/core/src/main/scala/kafka/controller/KafkaController.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/controller/KafkaController.scala b/core/src/main/scala/kafka/controller/KafkaController.scala
index 4a94aad..730f07c 100755
--- a/core/src/main/scala/kafka/controller/KafkaController.scala
+++ b/core/src/main/scala/kafka/controller/KafkaController.scala
@@ -20,7 +20,7 @@ import java.util
 
 import org.apache.kafka.common.errors.{BrokerNotAvailableException, ControllerMovedException}
 import org.apache.kafka.common.protocol.ApiKeys
-import org.apache.kafka.common.requests.{AbstractRequest, AbstractRequestResponse}
+import org.apache.kafka.common.requests.{AbstractRequest, AbstractResponse}
 
 import scala.collection._
 import com.yammer.metrics.core.{Gauge, Meter}
@@ -39,7 +39,7 @@ import kafka.utils.CoreUtils._
 import org.apache.zookeeper.Watcher.Event.KeeperState
 import org.apache.kafka.common.metrics.Metrics
 import org.apache.kafka.common.utils.Time
-import org.I0Itec.zkclient.{IZkChildListener, IZkDataListener, IZkStateListener, ZkClient, ZkConnection}
+import org.I0Itec.zkclient.{IZkChildListener, IZkDataListener, IZkStateListener, ZkClient}
 import org.I0Itec.zkclient.exception.{ZkNoNodeException, ZkNodeExistsException}
 import java.util.concurrent.locks.ReentrantLock
 
@@ -136,7 +136,7 @@ object KafkaController extends Logging {
       Json.parseFull(controllerInfoString) match {
         case Some(m) =>
           val controllerInfo = m.asInstanceOf[Map[String, Any]]
-          controllerInfo.get("brokerid").get.asInstanceOf[Int]
+          controllerInfo("brokerid").asInstanceOf[Int]
         case None => throw new KafkaException("Failed to parse the controller info json [%s].".format(controllerInfoString))
       }
     } catch {
@@ -686,7 +686,7 @@ class KafkaController(val config : KafkaConfig, zkUtils: ZkUtils, val brokerStat
     onControllerResignation()
   }
 
-  def sendRequest(brokerId: Int, apiKey: ApiKeys, apiVersion: Option[Short], request: AbstractRequest, callback: AbstractRequestResponse => Unit = null) = {
+  def sendRequest(brokerId: Int, apiKey: ApiKeys, apiVersion: Option[Short], request: AbstractRequest, callback: AbstractResponse => Unit = null) = {
     controllerContext.controllerChannelManager.sendRequest(brokerId, apiKey, apiVersion, request, callback)
   }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/3b4c3479/core/src/main/scala/kafka/controller/TopicDeletionManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/controller/TopicDeletionManager.scala b/core/src/main/scala/kafka/controller/TopicDeletionManager.scala
index 8e5f3a1..823f9b4 100755
--- a/core/src/main/scala/kafka/controller/TopicDeletionManager.scala
+++ b/core/src/main/scala/kafka/controller/TopicDeletionManager.scala
@@ -19,13 +19,14 @@ package kafka.controller
 
 import kafka.server.ConfigType
 import org.apache.kafka.common.protocol.Errors
-import org.apache.kafka.common.requests.{StopReplicaResponse, AbstractRequestResponse}
+import org.apache.kafka.common.requests.{AbstractResponse, StopReplicaResponse}
 
 import collection.mutable
 import collection.JavaConverters._
-import kafka.utils.{ShutdownableThread, Logging}
+import kafka.utils.{Logging, ShutdownableThread}
 import kafka.utils.CoreUtils._
 import kafka.utils.ZkUtils._
+
 import collection.Set
 import kafka.common.TopicAndPartition
 import java.util.concurrent.locks.ReentrantLock
@@ -378,7 +379,7 @@ class TopicDeletionManager(controller: KafkaController,
     startReplicaDeletion(replicasPerPartition)
   }
 
-  private def deleteTopicStopReplicaCallback(stopReplicaResponseObj: AbstractRequestResponse, replicaId: Int) {
+  private def deleteTopicStopReplicaCallback(stopReplicaResponseObj: AbstractResponse, replicaId: Int) {
     val stopReplicaResponse = stopReplicaResponseObj.asInstanceOf[StopReplicaResponse]
     debug("Delete topic callback invoked for %s".format(stopReplicaResponse))
     val responseMap = stopReplicaResponse.responses.asScala

http://git-wip-us.apache.org/repos/asf/kafka/blob/3b4c3479/core/src/main/scala/kafka/log/FileMessageSet.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/log/FileMessageSet.scala b/core/src/main/scala/kafka/log/FileMessageSet.scala
index fd9ec5f..c33e376 100755
--- a/core/src/main/scala/kafka/log/FileMessageSet.scala
+++ b/core/src/main/scala/kafka/log/FileMessageSet.scala
@@ -26,9 +26,10 @@ import kafka.utils._
 import kafka.message._
 import kafka.common.KafkaException
 import java.util.concurrent.TimeUnit
-import kafka.metrics.{KafkaTimer, KafkaMetricsGroup}
+
+import kafka.metrics.{KafkaMetricsGroup, KafkaTimer}
 import org.apache.kafka.common.errors.CorruptRecordException
-import org.apache.kafka.common.network.TransportLayer
+import org.apache.kafka.common.record.FileRecords
 import org.apache.kafka.common.utils.Utils
 
 import scala.collection.mutable.ArrayBuffer
@@ -48,7 +49,6 @@ class FileMessageSet private[kafka](@volatile var file: File,
                                     private[log] val start: Int,
                                     private[log] val end: Int,
                                     isSlice: Boolean) extends MessageSet {
-  import FileMessageSet._
   /* the size of the message set in bytes */
   private val _size =
     if(isSlice)
@@ -126,6 +126,8 @@ class FileMessageSet private[kafka](@volatile var file: File,
                        })
   }
 
+  override def asRecords: FileRecords = new FileRecords(file, channel, start, end, isSlice)
+
   /**
    * Search forward for the file position of the last offset that is greater than or equal to the target offset
    * and return its physical position and the size of the message (including log overhead) at the returned offset. If
@@ -206,31 +208,6 @@ class FileMessageSet private[kafka](@volatile var file: File,
   }
 
   /**
-   * Write some of this set to the given channel.
-   * @param destChannel The channel to write to.
-   * @param writePosition The position in the message set to begin writing from.
-   * @param size The maximum number of bytes to write
-   * @return The number of bytes actually written.
-   */
-  def writeTo(destChannel: GatheringByteChannel, writePosition: Long, size: Int): Int = {
-    // Ensure that the underlying size has not changed.
-    val newSize = math.min(channel.size.toInt, end) - start
-    if (newSize < _size.get()) {
-      throw new KafkaException("Size of FileMessageSet %s has been truncated during write: old size %d, new size %d"
-        .format(file.getAbsolutePath, _size.get(), newSize))
-    }
-    val position = start + writePosition
-    val count = math.min(size, sizeInBytes)
-    val bytesTransferred = (destChannel match {
-      case tl: TransportLayer => tl.transferFrom(channel, position, count)
-      case dc => channel.transferTo(position, count, dc)
-    }).toInt
-    trace("FileMessageSet " + file.getAbsolutePath + " : bytes transferred : " + bytesTransferred
-      + " bytes requested for transfer : " + math.min(size, sizeInBytes))
-    bytesTransferred
-  }
-
-  /**
     * This method is called before we write messages to the socket using zero-copy transfer. We need to
     * make sure all the messages in the message set have the expected magic value.
     *
@@ -337,7 +314,7 @@ class FileMessageSet private[kafka](@volatile var file: File,
 
         // increment the location and return the item
         location += size + sizeOffsetLength
-        new MessageAndOffset(new Message(buffer), offset)
+        MessageAndOffset(new Message(buffer), offset)
       }
     }
   }

http://git-wip-us.apache.org/repos/asf/kafka/blob/3b4c3479/core/src/main/scala/kafka/message/ByteBufferMessageSet.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/message/ByteBufferMessageSet.scala b/core/src/main/scala/kafka/message/ByteBufferMessageSet.scala
index a33bc4b..096344d 100644
--- a/core/src/main/scala/kafka/message/ByteBufferMessageSet.scala
+++ b/core/src/main/scala/kafka/message/ByteBufferMessageSet.scala
@@ -26,7 +26,7 @@ import java.util.ArrayDeque
 
 import kafka.message.ByteBufferMessageSet.FilterResult
 import org.apache.kafka.common.errors.InvalidTimestampException
-import org.apache.kafka.common.record.TimestampType
+import org.apache.kafka.common.record.{MemoryRecords, TimestampType}
 import org.apache.kafka.common.utils.Utils
 
 import scala.collection.mutable
@@ -352,6 +352,8 @@ class ByteBufferMessageSet(val buffer: ByteBuffer) extends MessageSet with Loggi
 
   def getBuffer = buffer
 
+  override def asRecords: MemoryRecords = MemoryRecords.readableRecords(buffer.duplicate())
+
   private def shallowValidBytes: Int = {
     if (shallowValidByteCount < 0) {
       this.shallowValidByteCount = this.internalIterator(isShallow = true).map { messageAndOffset =>
@@ -371,19 +373,6 @@ class ByteBufferMessageSet(val buffer: ByteBuffer) extends MessageSet with Loggi
     written
   }
 
-  /** Write the messages in this set to the given channel starting at the given offset byte.
-    * Less than the complete amount may be written, but no more than maxSize can be. The number
-    * of bytes written is returned */
-  def writeTo(channel: GatheringByteChannel, offset: Long, maxSize: Int): Int = {
-    if (offset > Int.MaxValue)
-      throw new IllegalArgumentException(s"offset should not be larger than Int.MaxValue: $offset")
-    val dup = buffer.duplicate()
-    val position = offset.toInt
-    dup.position(position)
-    dup.limit(math.min(buffer.limit, position + maxSize))
-    channel.write(dup)
-  }
-
   override def isMagicValueInAllWrapperMessages(expectedMagicValue: Byte): Boolean = {
     for (messageAndOffset <- shallowIterator) {
       if (messageAndOffset.message.magic != expectedMagicValue)

http://git-wip-us.apache.org/repos/asf/kafka/blob/3b4c3479/core/src/main/scala/kafka/message/MessageSet.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/message/MessageSet.scala b/core/src/main/scala/kafka/message/MessageSet.scala
index 14c455c..ffa27fa 100644
--- a/core/src/main/scala/kafka/message/MessageSet.scala
+++ b/core/src/main/scala/kafka/message/MessageSet.scala
@@ -20,6 +20,8 @@ package kafka.message
 import java.nio._
 import java.nio.channels._
 
+import org.apache.kafka.common.record.Records
+
 /**
  * Message set helper functions
  */
@@ -70,11 +72,6 @@ case class MagicAndTimestamp(magic: Byte, timestamp: Long)
  */
 abstract class MessageSet extends Iterable[MessageAndOffset] {
 
-  /** Write the messages in this set to the given channel starting at the given offset byte. 
-    * Less than the complete amount may be written, but no more than maxSize can be. The number
-    * of bytes written is returned */
-  def writeTo(channel: GatheringByteChannel, offset: Long, maxSize: Int): Int
-
   /**
    * Check if all the wrapper messages in the message set have the expected magic value
    */
@@ -91,6 +88,11 @@ abstract class MessageSet extends Iterable[MessageAndOffset] {
   def sizeInBytes: Int
 
   /**
+   * Get the client representation of the message set
+   */
+  def asRecords: Records
+
+  /**
    * Print this message set's contents. If the message set has more than 100 messages, just
    * print the first 100.
    */

http://git-wip-us.apache.org/repos/asf/kafka/blob/3b4c3479/core/src/main/scala/kafka/network/RequestChannel.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/network/RequestChannel.scala b/core/src/main/scala/kafka/network/RequestChannel.scala
index dace782..0cece68 100644
--- a/core/src/main/scala/kafka/network/RequestChannel.scala
+++ b/core/src/main/scala/kafka/network/RequestChannel.scala
@@ -23,7 +23,7 @@ import java.util.HashMap
 import java.util.concurrent._
 
 import com.yammer.metrics.core.Gauge
-import kafka.api._
+import kafka.api.{ControlledShutdownRequest, RequestOrResponse}
 import kafka.metrics.KafkaMetricsGroup
 import kafka.server.QuotaId
 import kafka.utils.{Logging, SystemTime}
@@ -31,19 +31,19 @@ import org.apache.kafka.common.TopicPartition
 import org.apache.kafka.common.errors.InvalidRequestException
 import org.apache.kafka.common.network.Send
 import org.apache.kafka.common.protocol.{ApiKeys, Protocol, SecurityProtocol}
-import org.apache.kafka.common.requests.{AbstractRequest, ApiVersionsRequest, ProduceRequest, RequestHeader, RequestSend}
+import org.apache.kafka.common.record.MemoryRecords
+import org.apache.kafka.common.requests._
 import org.apache.kafka.common.security.auth.KafkaPrincipal
 import org.apache.log4j.Logger
 
-
 object RequestChannel extends Logging {
-  val AllDone = new Request(processor = 1, connectionId = "2", new Session(KafkaPrincipal.ANONYMOUS, InetAddress.getLocalHost()), buffer = getShutdownReceive(), startTimeMs = 0, securityProtocol = SecurityProtocol.PLAINTEXT)
+  val AllDone = Request(processor = 1, connectionId = "2", Session(KafkaPrincipal.ANONYMOUS, InetAddress.getLocalHost), buffer = getShutdownReceive(), startTimeMs = 0, securityProtocol = SecurityProtocol.PLAINTEXT)
   private val requestLogger = Logger.getLogger("kafka.request.logger")
 
   def getShutdownReceive() = {
     val emptyRequestHeader = new RequestHeader(ApiKeys.PRODUCE.id, "", 0)
-    val emptyProduceRequest = new ProduceRequest(0, 0, new HashMap[TopicPartition, ByteBuffer]())
-    RequestSend.serialize(emptyRequestHeader, emptyProduceRequest.toStruct)
+    val emptyProduceRequest = new ProduceRequest(0, 0, new HashMap[TopicPartition, MemoryRecords]())
+    AbstractRequestResponse.serialize(emptyRequestHeader, emptyProduceRequest)
   }
 
   case class Session(principal: KafkaPrincipal, clientAddress: InetAddress) {
@@ -61,19 +61,12 @@ object RequestChannel extends Logging {
 
     val requestId = buffer.getShort()
 
-    // TODO: this will be removed once we migrated to client-side format
-    // for server-side request / response format
-    // NOTE: this map only includes the server-side request/response handlers. Newer
-    // request types should only use the client-side versions which are parsed with
-    // o.a.k.common.requests.AbstractRequest.getRequest()
-    private val keyToNameAndDeserializerMap: Map[Short, (ByteBuffer) => RequestOrResponse]=
-      Map(ApiKeys.FETCH.id -> FetchRequest.readFrom,
-        ApiKeys.CONTROLLED_SHUTDOWN_KEY.id -> ControlledShutdownRequest.readFrom
-      )
-
-    // TODO: this will be removed once we migrated to client-side format
-    val requestObj =
-      keyToNameAndDeserializerMap.get(requestId).map(readFrom => readFrom(buffer)).orNull
+    // TODO: this will be removed once we remove support for v0 of ControlledShutdownRequest (which
+    // depends on a non-standard request header)
+    val requestObj: RequestOrResponse = if (requestId == ApiKeys.CONTROLLED_SHUTDOWN_KEY.id)
+      ControlledShutdownRequest.readFrom(buffer)
+    else
+      null
 
     // if we failed to find a server-side mapping, then try using the
     // client-side request / response format
@@ -108,7 +101,7 @@ object RequestChannel extends Logging {
       if (requestObj != null)
         requestObj.describe(details)
       else
-        header.toString + " -- " + body.toString
+        s"$header -- $body"
     }
 
     trace("Processor %d received request : %s".format(processor, requestDesc(true)))
@@ -135,7 +128,7 @@ object RequestChannel extends Logging {
       val totalTime = endTimeMs - startTimeMs
       val fetchMetricNames =
         if (requestId == ApiKeys.FETCH.id) {
-          val isFromFollower = requestObj.asInstanceOf[FetchRequest].isFromFollower
+          val isFromFollower = body.asInstanceOf[FetchRequest].isFromFollower
           Seq(
             if (isFromFollower) RequestMetrics.followFetchMetricName
             else RequestMetrics.consumerFetchMetricName
@@ -172,6 +165,9 @@ object RequestChannel extends Logging {
 
     def this(request: Request, send: Send) =
       this(request.processor, request, send)
+
+    def this(request: Request, response: AbstractResponse) =
+      this(request, response.toSend(request.connectionId, request.header))
   }
 
   trait ResponseAction
@@ -221,14 +217,14 @@ class RequestChannel(val numProcessors: Int, val queueSize: Int) extends KafkaMe
 
   /** No operation to take for the request, need to read more over the network */
   def noOperation(processor: Int, request: RequestChannel.Request) {
-    responseQueues(processor).put(new RequestChannel.Response(processor, request, null, RequestChannel.NoOpAction))
+    responseQueues(processor).put(RequestChannel.Response(processor, request, null, RequestChannel.NoOpAction))
     for(onResponse <- responseListeners)
       onResponse(processor)
   }
 
   /** Close the connection for the request */
   def closeConnection(processor: Int, request: RequestChannel.Request) {
-    responseQueues(processor).put(new RequestChannel.Response(processor, request, null, RequestChannel.CloseConnectionAction))
+    responseQueues(processor).put(RequestChannel.Response(processor, request, null, RequestChannel.CloseConnectionAction))
     for(onResponse <- responseListeners)
       onResponse(processor)
   }
@@ -254,7 +250,7 @@ class RequestChannel(val numProcessors: Int, val queueSize: Int) extends KafkaMe
   }
 
   def shutdown() {
-    requestQueue.clear
+    requestQueue.clear()
   }
 }
 
@@ -283,4 +279,3 @@ class RequestMetrics(name: String) extends KafkaMetricsGroup {
   val responseSendTimeHist = newHistogram("ResponseSendTimeMs", biased = true, tags)
   val totalTimeHist = newHistogram("TotalTimeMs", biased = true, tags)
 }
-

http://git-wip-us.apache.org/repos/asf/kafka/blob/3b4c3479/core/src/main/scala/kafka/server/DelayedFetch.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/DelayedFetch.scala b/core/src/main/scala/kafka/server/DelayedFetch.scala
index 4bf04e6..001051f 100644
--- a/core/src/main/scala/kafka/server/DelayedFetch.scala
+++ b/core/src/main/scala/kafka/server/DelayedFetch.scala
@@ -20,14 +20,15 @@ package kafka.server
 import java.util.concurrent.TimeUnit
 
 import kafka.api.FetchResponsePartitionData
-import kafka.api.PartitionFetchInfo
 import kafka.common.TopicAndPartition
 import kafka.metrics.KafkaMetricsGroup
+import org.apache.kafka.common.TopicPartition
 import org.apache.kafka.common.errors.{NotLeaderForPartitionException, UnknownTopicOrPartitionException}
+import org.apache.kafka.common.requests.FetchRequest.PartitionData
 
 import scala.collection._
 
-case class FetchPartitionStatus(startOffsetMetadata: LogOffsetMetadata, fetchInfo: PartitionFetchInfo) {
+case class FetchPartitionStatus(startOffsetMetadata: LogOffsetMetadata, fetchInfo: PartitionData) {
 
   override def toString = "[startOffsetMetadata: " + startOffsetMetadata + ", " +
                           "fetchInfo: " + fetchInfo + "]"
@@ -103,7 +104,7 @@ class DelayedFetch(delayMs: Long,
                   return forceComplete()
               } else if (fetchOffset.messageOffset < endOffset.messageOffset) {
                 // we take the partition fetch size as upper bound when accumulating the bytes (skip if a throttled partition)
-                val bytesAvailable = math.min(endOffset.positionDiff(fetchOffset), fetchStatus.fetchInfo.fetchSize)
+                val bytesAvailable = math.min(endOffset.positionDiff(fetchOffset), fetchStatus.fetchInfo.maxBytes)
                 if (quota.isThrottled(topicAndPartition))
                   accumulatedThrottledSize += bytesAvailable
                 else
@@ -146,7 +147,7 @@ class DelayedFetch(delayMs: Long,
       readOnlyCommitted = fetchMetadata.fetchOnlyCommitted,
       fetchMaxBytes = fetchMetadata.fetchMaxBytes,
       hardMaxBytesLimit = fetchMetadata.hardMaxBytesLimit,
-      readPartitionInfo = fetchMetadata.fetchPartitionStatus.map { case (tp, status) => tp -> status.fetchInfo },
+      readPartitionInfo = fetchMetadata.fetchPartitionStatus.map { case (tp, status) => new TopicPartition(tp.topic, tp.partition) -> status.fetchInfo },
       quota = quota
     )
 


Mime
View raw message