http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java ---------------------------------------------------------------------- diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java index fec9251..6a17da8 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java @@ -886,7 +886,7 @@ public class FetcherTest { ListOffsetResponse.PartitionData partitionData = new ListOffsetResponse.PartitionData(error, timestamp, offset); Map<TopicPartition, ListOffsetResponse.PartitionData> allPartitionData = new HashMap<>(); allPartitionData.put(tp, partitionData); - return new ListOffsetResponse(allPartitionData, 1); + return new ListOffsetResponse(allPartitionData); } private FetchResponse fetchResponse(MemoryRecords records, Errors error, long hw, int throttleTime) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/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 699f6e2..d0b9639 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 @@ -30,6 +30,7 @@ import org.apache.kafka.common.record.Record; import org.junit.Test; import java.io.IOException; +import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; import java.nio.ByteBuffer; import java.nio.channels.GatheringByteChannel; @@ -51,133 +52,161 @@ public class RequestResponseTest { @Test public void testSerialization() throws Exception { - checkSerialization(createRequestHeader(), null); - checkSerialization(createResponseHeader(), null); - checkSerialization(createGroupCoordinatorRequest()); - checkSerialization(createGroupCoordinatorRequest().getErrorResponse(new UnknownServerException()), null); - checkSerialization(createGroupCoordinatorResponse(), null); - checkSerialization(createControlledShutdownRequest()); - checkSerialization(createControlledShutdownResponse(), null); - checkSerialization(createControlledShutdownRequest().getErrorResponse(new UnknownServerException()), null); - checkSerialization(createFetchRequest(3), 3); - checkSerialization(createFetchRequest(3).getErrorResponse(new UnknownServerException()), 3); - checkSerialization(createFetchResponse(), null); - checkSerialization(createHeartBeatRequest()); - checkSerialization(createHeartBeatRequest().getErrorResponse(new UnknownServerException()), null); - checkSerialization(createHeartBeatResponse(), null); - checkSerialization(createJoinGroupRequest(1), 1); - checkSerialization(createJoinGroupRequest(0).getErrorResponse(new UnknownServerException()), 0); - checkSerialization(createJoinGroupRequest(1).getErrorResponse(new UnknownServerException()), 1); - checkSerialization(createJoinGroupResponse(), null); - checkSerialization(createLeaveGroupRequest()); - checkSerialization(createLeaveGroupRequest().getErrorResponse(new UnknownServerException()), null); - checkSerialization(createLeaveGroupResponse(), null); - checkSerialization(createListGroupsRequest()); - checkSerialization(createListGroupsRequest().getErrorResponse(new UnknownServerException()), null); - checkSerialization(createListGroupsResponse(), null); - checkSerialization(createDescribeGroupRequest()); - checkSerialization(createDescribeGroupRequest().getErrorResponse(new UnknownServerException()), null); - checkSerialization(createDescribeGroupResponse(), null); - checkSerialization(createListOffsetRequest(1), 1); - checkSerialization(createListOffsetRequest(1).getErrorResponse(new UnknownServerException()), 1); - checkSerialization(createListOffsetResponse(1), 1); - checkSerialization(MetadataRequest.allTopics((short) 2), 2); - checkSerialization(createMetadataRequest(1, Arrays.asList("topic1")), 1); - checkSerialization(createMetadataRequest(1, Arrays.asList("topic1")).getErrorResponse(new UnknownServerException()), 1); - checkSerialization(createMetadataResponse(2), 2); - checkSerialization(createMetadataRequest(2, Arrays.asList("topic1")).getErrorResponse(new UnknownServerException()), 2); - checkSerialization(createOffsetCommitRequest(2), 2); - checkSerialization(createOffsetCommitRequest(2).getErrorResponse(new UnknownServerException()), 2); - checkSerialization(createOffsetCommitResponse(), null); - checkSerialization(OffsetFetchRequest.forAllPartitions("group1")); - checkSerialization(OffsetFetchRequest.forAllPartitions("group1").getErrorResponse(new NotCoordinatorForGroupException()), 2); - checkSerialization(createOffsetFetchRequest(0)); - checkSerialization(createOffsetFetchRequest(1)); - checkSerialization(createOffsetFetchRequest(2)); - checkSerialization(OffsetFetchRequest.forAllPartitions("group1")); - checkSerialization(createOffsetFetchRequest(0).getErrorResponse(new UnknownServerException()), 0); - checkSerialization(createOffsetFetchRequest(1).getErrorResponse(new UnknownServerException()), 1); - checkSerialization(createOffsetFetchRequest(2).getErrorResponse(new UnknownServerException()), 2); - checkSerialization(createOffsetFetchResponse(), null); - checkSerialization(createProduceRequest()); - checkSerialization(createProduceRequest().getErrorResponse(new UnknownServerException()), null); - checkSerialization(createProduceResponse(), null); - checkSerialization(createStopReplicaRequest(true)); - checkSerialization(createStopReplicaRequest(false)); - checkSerialization(createStopReplicaRequest(true).getErrorResponse(new UnknownServerException()), null); - checkSerialization(createStopReplicaResponse(), null); - checkSerialization(createLeaderAndIsrRequest()); - checkSerialization(createLeaderAndIsrRequest().getErrorResponse(new UnknownServerException()), null); - checkSerialization(createLeaderAndIsrResponse(), null); - checkSerialization(createSaslHandshakeRequest()); - checkSerialization(createSaslHandshakeRequest().getErrorResponse(new UnknownServerException()), null); - checkSerialization(createSaslHandshakeResponse(), null); - checkSerialization(createApiVersionRequest()); - checkSerialization(createApiVersionRequest().getErrorResponse(new UnknownServerException()), null); - checkSerialization(createApiVersionResponse(), null); - checkSerialization(createCreateTopicRequest(0), 0); - checkSerialization(createCreateTopicRequest(0).getErrorResponse(new UnknownServerException()), 0); - checkSerialization(createCreateTopicResponse(0), 0); - checkSerialization(createCreateTopicRequest(1), 1); - checkSerialization(createCreateTopicRequest(1).getErrorResponse(new UnknownServerException()), 1); - checkSerialization(createCreateTopicResponse(1), 1); - checkSerialization(createDeleteTopicsRequest()); - checkSerialization(createDeleteTopicsRequest().getErrorResponse(new UnknownServerException()), null); - checkSerialization(createDeleteTopicsResponse(), null); + checkRequest(createGroupCoordinatorRequest()); + checkErrorResponse(createGroupCoordinatorRequest(), new UnknownServerException()); + checkResponse(createGroupCoordinatorResponse(), 0); + checkRequest(createControlledShutdownRequest()); + checkResponse(createControlledShutdownResponse(), 1); + checkErrorResponse(createControlledShutdownRequest(), new UnknownServerException()); + checkRequest(createFetchRequest(3)); + checkErrorResponse(createFetchRequest(3), new UnknownServerException()); + checkResponse(createFetchResponse(), 0); + checkRequest(createHeartBeatRequest()); + checkErrorResponse(createHeartBeatRequest(), new UnknownServerException()); + checkResponse(createHeartBeatResponse(), 0); + checkRequest(createJoinGroupRequest(1)); + checkErrorResponse(createJoinGroupRequest(0), new UnknownServerException()); + checkErrorResponse(createJoinGroupRequest(1), new UnknownServerException()); + checkResponse(createJoinGroupResponse(), 0); + checkRequest(createLeaveGroupRequest()); + checkErrorResponse(createLeaveGroupRequest(), new UnknownServerException()); + checkResponse(createLeaveGroupResponse(), 0); + checkRequest(createListGroupsRequest()); + checkErrorResponse(createListGroupsRequest(), new UnknownServerException()); + checkResponse(createListGroupsResponse(), 0); + checkRequest(createDescribeGroupRequest()); + checkErrorResponse(createDescribeGroupRequest(), new UnknownServerException()); + checkResponse(createDescribeGroupResponse(), 0); + checkRequest(createListOffsetRequest(1)); + checkErrorResponse(createListOffsetRequest(1), new UnknownServerException()); + checkResponse(createListOffsetResponse(1), 1); + checkRequest(MetadataRequest.Builder.allTopics().build((short) 2)); + checkRequest(createMetadataRequest(1, Arrays.asList("topic1"))); + checkErrorResponse(createMetadataRequest(1, Arrays.asList("topic1")), new UnknownServerException()); + checkResponse(createMetadataResponse(), 2); + checkErrorResponse(createMetadataRequest(2, Arrays.asList("topic1")), new UnknownServerException()); + checkRequest(createOffsetCommitRequest(2)); + checkErrorResponse(createOffsetCommitRequest(2), new UnknownServerException()); + checkResponse(createOffsetCommitResponse(), 0); + checkRequest(OffsetFetchRequest.forAllPartitions("group1")); + checkErrorResponse(OffsetFetchRequest.forAllPartitions("group1"), new NotCoordinatorForGroupException()); + checkRequest(createOffsetFetchRequest(0)); + checkRequest(createOffsetFetchRequest(1)); + checkRequest(createOffsetFetchRequest(2)); + checkRequest(OffsetFetchRequest.forAllPartitions("group1")); + checkErrorResponse(createOffsetFetchRequest(0), new UnknownServerException()); + checkErrorResponse(createOffsetFetchRequest(1), new UnknownServerException()); + checkErrorResponse(createOffsetFetchRequest(2), new UnknownServerException()); + checkResponse(createOffsetFetchResponse(), 0); + checkRequest(createProduceRequest()); + checkErrorResponse(createProduceRequest(), new UnknownServerException()); + checkResponse(createProduceResponse(), 2); + checkRequest(createStopReplicaRequest(true)); + checkRequest(createStopReplicaRequest(false)); + checkErrorResponse(createStopReplicaRequest(true), new UnknownServerException()); + checkResponse(createStopReplicaResponse(), 0); + checkRequest(createLeaderAndIsrRequest()); + checkErrorResponse(createLeaderAndIsrRequest(), new UnknownServerException()); + checkResponse(createLeaderAndIsrResponse(), 0); + checkRequest(createSaslHandshakeRequest()); + checkErrorResponse(createSaslHandshakeRequest(), new UnknownServerException()); + checkResponse(createSaslHandshakeResponse(), 0); + checkRequest(createApiVersionRequest()); + checkErrorResponse(createApiVersionRequest(), new UnknownServerException()); + checkResponse(createApiVersionResponse(), 0); + checkRequest(createCreateTopicRequest(0)); + checkErrorResponse(createCreateTopicRequest(0), new UnknownServerException()); + checkResponse(createCreateTopicResponse(), 0); + checkRequest(createCreateTopicRequest(1)); + checkErrorResponse(createCreateTopicRequest(1), new UnknownServerException()); + checkResponse(createCreateTopicResponse(), 1); + checkRequest(createDeleteTopicsRequest()); + checkErrorResponse(createDeleteTopicsRequest(), new UnknownServerException()); + checkResponse(createDeleteTopicsResponse(), 0); checkOlderFetchVersions(); - checkSerialization(createMetadataResponse(0), 0); - checkSerialization(createMetadataResponse(1), 1); - checkSerialization(createMetadataRequest(1, Arrays.asList("topic1")).getErrorResponse(new UnknownServerException()), 1); - checkSerialization(createOffsetCommitRequest(0), 0); - checkSerialization(createOffsetCommitRequest(0).getErrorResponse(new UnknownServerException()), 0); - checkSerialization(createOffsetCommitRequest(1), 1); - checkSerialization(createOffsetCommitRequest(1).getErrorResponse(new UnknownServerException()), 1); - checkSerialization(createJoinGroupRequest(0), 0); - checkSerialization(createUpdateMetadataRequest(0, null), 0); - checkSerialization(createUpdateMetadataRequest(0, null).getErrorResponse(new UnknownServerException()), 0); - checkSerialization(createUpdateMetadataRequest(1, null), 1); - checkSerialization(createUpdateMetadataRequest(1, "rack1"), 1); - checkSerialization(createUpdateMetadataRequest(1, null).getErrorResponse(new UnknownServerException()), 1); - checkSerialization(createUpdateMetadataRequest(2, "rack1"), 2); - checkSerialization(createUpdateMetadataRequest(2, null), 2); - checkSerialization(createUpdateMetadataRequest(2, "rack1").getErrorResponse(new UnknownServerException()), 2); - checkSerialization(createUpdateMetadataRequest(3, "rack1")); - checkSerialization(createUpdateMetadataRequest(3, null)); - checkSerialization(createUpdateMetadataRequest(3, "rack1").getErrorResponse(new UnknownServerException()), 3); - checkSerialization(createUpdateMetadataResponse(), null); - checkSerialization(createListOffsetRequest(0), 0); - checkSerialization(createListOffsetRequest(0).getErrorResponse(new UnknownServerException()), 0); - checkSerialization(createListOffsetResponse(0), 0); + checkResponse(createMetadataResponse(), 0); + checkResponse(createMetadataResponse(), 1); + checkErrorResponse(createMetadataRequest(1, Arrays.asList("topic1")), new UnknownServerException()); + checkRequest(createOffsetCommitRequest(0)); + checkErrorResponse(createOffsetCommitRequest(0), new UnknownServerException()); + checkRequest(createOffsetCommitRequest(1)); + checkErrorResponse(createOffsetCommitRequest(1), new UnknownServerException()); + checkRequest(createJoinGroupRequest(0)); + checkRequest(createUpdateMetadataRequest(0, null)); + checkErrorResponse(createUpdateMetadataRequest(0, null), new UnknownServerException()); + checkRequest(createUpdateMetadataRequest(1, null)); + checkRequest(createUpdateMetadataRequest(1, "rack1")); + checkErrorResponse(createUpdateMetadataRequest(1, null), new UnknownServerException()); + checkRequest(createUpdateMetadataRequest(2, "rack1")); + checkRequest(createUpdateMetadataRequest(2, null)); + checkErrorResponse(createUpdateMetadataRequest(2, "rack1"), new UnknownServerException()); + checkRequest(createUpdateMetadataRequest(3, "rack1")); + checkRequest(createUpdateMetadataRequest(3, null)); + checkErrorResponse(createUpdateMetadataRequest(3, "rack1"), new UnknownServerException()); + checkResponse(createUpdateMetadataResponse(), 0); + checkRequest(createListOffsetRequest(0)); + checkErrorResponse(createListOffsetRequest(0), new UnknownServerException()); + checkResponse(createListOffsetResponse(0), 0); + } + + @Test + public void testRequestHeader() { + RequestHeader header = createRequestHeader(); + ByteBuffer buffer = toBuffer(header.toStruct()); + RequestHeader deserialized = RequestHeader.parse(buffer); + assertEquals(header.apiVersion(), deserialized.apiVersion()); + assertEquals(header.apiKey(), deserialized.apiKey()); + assertEquals(header.clientId(), deserialized.clientId()); + assertEquals(header.correlationId(), deserialized.correlationId()); + } + + @Test + public void testResponseHeader() { + ResponseHeader header = createResponseHeader(); + ByteBuffer buffer = toBuffer(header.toStruct()); + ResponseHeader deserialized = ResponseHeader.parse(buffer); + assertEquals(header.correlationId(), deserialized.correlationId()); } private void checkOlderFetchVersions() throws Exception { int latestVersion = ProtoUtils.latestVersion(ApiKeys.FETCH.id); for (int i = 0; i < latestVersion; ++i) { - checkSerialization(createFetchRequest(i).getErrorResponse(new UnknownServerException()), i); - checkSerialization(createFetchRequest(i), i); + checkErrorResponse(createFetchRequest(i), new UnknownServerException()); + checkRequest(createFetchRequest(i)); } } - private void checkSerialization(AbstractRequest req) throws Exception { - checkSerialization(req, Integer.valueOf(req.version())); + private void checkErrorResponse(AbstractRequest req, Throwable e) throws Exception { + checkResponse(req.getErrorResponse(e), req.version()); + } + + private void checkRequest(AbstractRequest req) throws Exception { + // Check that we can serialize, deserialize and serialize again + // We don't check for equality or hashCode because it is likely to fail for any request containing a HashMap + Struct struct = req.toStruct(); + AbstractRequest deserialized = (AbstractRequest) deserialize(req, struct, req.version()); + deserialized.toStruct(); } - private void checkSerialization(AbstractRequestResponse req, Integer version) throws Exception { - ByteBuffer buffer = ByteBuffer.allocate(req.sizeOf()); - req.writeTo(buffer); + private void checkResponse(AbstractResponse response, int version) throws Exception { + // Check that we can serialize, deserialize and serialize again + // We don't check for equality or hashCode because it is likely to fail for any response containing a HashMap + Struct struct = response.toStruct((short) version); + AbstractResponse deserialized = (AbstractResponse) deserialize(response, struct, (short) version); + deserialized.toStruct((short) version); + } + + private AbstractRequestResponse deserialize(AbstractRequestResponse req, Struct struct, short version) throws NoSuchMethodException, IllegalAccessException, InvocationTargetException { + ByteBuffer buffer = toBuffer(struct); + Method deserializer = req.getClass().getDeclaredMethod("parse", ByteBuffer.class, Short.TYPE); + return (AbstractRequestResponse) deserializer.invoke(null, buffer, version); + } + + private ByteBuffer toBuffer(Struct struct) { + ByteBuffer buffer = ByteBuffer.allocate(struct.sizeOf()); + struct.writeTo(buffer); buffer.rewind(); - AbstractRequestResponse deserialized; - if (version == null) { - Method deserializer = req.getClass().getDeclaredMethod("parse", ByteBuffer.class); - deserialized = (AbstractRequestResponse) deserializer.invoke(null, buffer); - } else { - Method deserializer = req.getClass().getDeclaredMethod("parse", ByteBuffer.class, Integer.TYPE); - deserialized = (AbstractRequestResponse) deserializer.invoke(null, buffer, version); - } - assertEquals("The original and deserialized of " + req.getClass().getSimpleName() + - "(version " + version + ") should be the same.", req, deserialized); - assertEquals("The original and deserialized of " + req.getClass().getSimpleName() + " should have the same hashcode.", - req.hashCode(), deserialized.hashCode()); + return buffer; } @Test @@ -186,14 +215,17 @@ public class RequestResponseTest { responseData.put(new TopicPartition("test", 0), new ProduceResponse.PartitionResponse(Errors.NONE, 10000, Record.NO_TIMESTAMP)); ProduceResponse v0Response = new ProduceResponse(responseData); - ProduceResponse v1Response = new ProduceResponse(responseData, 10, 1); - ProduceResponse v2Response = new ProduceResponse(responseData, 10, 2); + ProduceResponse v1Response = new ProduceResponse(responseData, 10); + ProduceResponse v2Response = new ProduceResponse(responseData, 10); assertEquals("Throttle time must be zero", 0, v0Response.getThrottleTime()); assertEquals("Throttle time must be 10", 10, v1Response.getThrottleTime()); assertEquals("Throttle time must be 10", 10, v2Response.getThrottleTime()); - assertEquals("Should use schema version 0", ProtoUtils.responseSchema(ApiKeys.PRODUCE.id, 0), v0Response.toStruct().schema()); - assertEquals("Should use schema version 1", ProtoUtils.responseSchema(ApiKeys.PRODUCE.id, 1), v1Response.toStruct().schema()); - assertEquals("Should use schema version 2", ProtoUtils.responseSchema(ApiKeys.PRODUCE.id, 2), v2Response.toStruct().schema()); + assertEquals("Should use schema version 0", ProtoUtils.responseSchema(ApiKeys.PRODUCE.id, 0), + v0Response.toStruct((short) 0).schema()); + assertEquals("Should use schema version 1", ProtoUtils.responseSchema(ApiKeys.PRODUCE.id, 1), + v1Response.toStruct((short) 1).schema()); + assertEquals("Should use schema version 2", ProtoUtils.responseSchema(ApiKeys.PRODUCE.id, 2), + v2Response.toStruct((short) 2).schema()); assertEquals("Response data does not match", responseData, v0Response.responses()); assertEquals("Response data does not match", responseData, v1Response.responses()); assertEquals("Response data does not match", responseData, v2Response.responses()); @@ -206,12 +238,14 @@ public class RequestResponseTest { MemoryRecords records = MemoryRecords.readableRecords(ByteBuffer.allocate(10)); responseData.put(new TopicPartition("test", 0), new FetchResponse.PartitionData(Errors.NONE, 1000000, records)); - FetchResponse v0Response = new FetchResponse(0, responseData, 0); - FetchResponse v1Response = new FetchResponse(1, responseData, 10); - assertEquals("Throttle time must be zero", 0, v0Response.getThrottleTime()); - assertEquals("Throttle time must be 10", 10, v1Response.getThrottleTime()); - assertEquals("Should use schema version 0", ProtoUtils.responseSchema(ApiKeys.FETCH.id, 0), v0Response.toStruct().schema()); - assertEquals("Should use schema version 1", ProtoUtils.responseSchema(ApiKeys.FETCH.id, 1), v1Response.toStruct().schema()); + FetchResponse v0Response = new FetchResponse(responseData, 0); + FetchResponse v1Response = new FetchResponse(responseData, 10); + assertEquals("Throttle time must be zero", 0, v0Response.throttleTimeMs()); + assertEquals("Throttle time must be 10", 10, v1Response.throttleTimeMs()); + assertEquals("Should use schema version 0", ProtoUtils.responseSchema(ApiKeys.FETCH.id, 0), + v0Response.toStruct((short) 0).schema()); + assertEquals("Should use schema version 1", ProtoUtils.responseSchema(ApiKeys.FETCH.id, 1), + v1Response.toStruct((short) 1).schema()); assertEquals("Response data does not match", responseData, v0Response.responseData()); assertEquals("Response data does not match", responseData, v1Response.responseData()); } @@ -239,19 +273,18 @@ public class RequestResponseTest { // read the body Struct responseBody = ProtoUtils.responseSchema(ApiKeys.FETCH.id, header.apiVersion()).read(buf); - FetchResponse parsedResponse = new FetchResponse(responseBody); - assertEquals(parsedResponse, fetchResponse); + assertEquals(fetchResponse.toStruct(header.apiVersion()), responseBody); - assertEquals(size, responseHeader.sizeOf() + parsedResponse.sizeOf()); + assertEquals(size, responseHeader.sizeOf() + responseBody.sizeOf()); } @Test public void testControlledShutdownResponse() { ControlledShutdownResponse response = createControlledShutdownResponse(); - ByteBuffer buffer = ByteBuffer.allocate(response.sizeOf()); - response.writeTo(buffer); - buffer.rewind(); - ControlledShutdownResponse deserialized = ControlledShutdownResponse.parse(buffer); + short version = ProtoUtils.latestVersion(ApiKeys.CONTROLLED_SHUTDOWN_KEY.id); + Struct struct = response.toStruct(version); + ByteBuffer buffer = toBuffer(struct); + ControlledShutdownResponse deserialized = ControlledShutdownResponse.parse(buffer, version); assertEquals(response.error(), deserialized.error()); assertEquals(response.partitionsRemaining(), deserialized.partitionsRemaining()); } @@ -259,9 +292,8 @@ public class RequestResponseTest { @Test public void testRequestHeaderWithNullClientId() { RequestHeader header = new RequestHeader((short) 10, (short) 1, null, 10); - ByteBuffer buffer = ByteBuffer.allocate(header.sizeOf()); - header.writeTo(buffer); - buffer.rewind(); + Struct headerStruct = header.toStruct(); + ByteBuffer buffer = toBuffer(headerStruct); RequestHeader deserialized = RequestHeader.parse(buffer); assertEquals(header.apiKey(), deserialized.apiKey()); assertEquals(header.apiVersion(), deserialized.apiVersion()); @@ -294,8 +326,7 @@ public class RequestResponseTest { 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)); - return new FetchRequest.Builder(100, 100000, fetchData).setMaxBytes(1000). - setVersion((short) version).build(); + return FetchRequest.Builder.forConsumer(100, 100000, fetchData).setMaxBytes(1000).build((short) version); } private FetchResponse createFetchResponse() { @@ -320,7 +351,7 @@ public class RequestResponseTest { protocols.add(new JoinGroupRequest.ProtocolMetadata("consumer-range", metadata)); if (version == 0) { return new JoinGroupRequest.Builder("group1", 30000, "consumer1", "consumer", protocols). - setVersion((short) version).build(); + build((short) version); } else { return new JoinGroupRequest.Builder("group1", 10000, "consumer1", "consumer", protocols). setRebalanceTimeout(60000).build(); @@ -372,11 +403,11 @@ public class RequestResponseTest { Map<TopicPartition, ListOffsetRequest.PartitionData> offsetData = Collections.singletonMap( new TopicPartition("test", 0), new ListOffsetRequest.PartitionData(1000000L, 10)); - return new ListOffsetRequest.Builder().setOffsetData(offsetData).setVersion((short) version).build(); + return ListOffsetRequest.Builder.forConsumer((short) 0).setOffsetData(offsetData).build((short) version); } else if (version == 1) { Map<TopicPartition, Long> offsetData = Collections.singletonMap( new TopicPartition("test", 0), 1000000L); - return new ListOffsetRequest.Builder().setTargetTimes(offsetData).setVersion((short) version).build(); + return ListOffsetRequest.Builder.forConsumer((short) 1).setTargetTimes(offsetData).build((short) version); } else { throw new IllegalArgumentException("Illegal ListOffsetRequest version " + version); } @@ -386,24 +417,24 @@ public class RequestResponseTest { 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, Arrays.asList(100L))); + responseData.put(new TopicPartition("test", 0), + new ListOffsetResponse.PartitionData(Errors.NONE, Arrays.asList(100L))); return new ListOffsetResponse(responseData); } else if (version == 1) { Map<TopicPartition, ListOffsetResponse.PartitionData> responseData = new HashMap<>(); - responseData.put(new TopicPartition("test", 0), new ListOffsetResponse.PartitionData(Errors.NONE, 10000L, 100L)); - return new ListOffsetResponse(responseData, 1); + responseData.put(new TopicPartition("test", 0), + new ListOffsetResponse.PartitionData(Errors.NONE, 10000L, 100L)); + return new ListOffsetResponse(responseData); } else { throw new IllegalArgumentException("Illegal ListOffsetResponse version " + version); } } private MetadataRequest createMetadataRequest(int version, List<String> topics) { - return new MetadataRequest.Builder(topics). - setVersion((short) version). - build(); + return new MetadataRequest.Builder(topics).build((short) version); } - private MetadataResponse createMetadataResponse(int version) { + private MetadataResponse createMetadataResponse() { Node node = new Node(1, "host1", 1001); List<Node> replicas = Arrays.asList(node); List<Node> isr = Arrays.asList(node); @@ -414,7 +445,7 @@ public class RequestResponseTest { allTopicMetadata.add(new MetadataResponse.TopicMetadata(Errors.LEADER_NOT_AVAILABLE, "topic2", false, Collections.<MetadataResponse.PartitionMetadata>emptyList())); - return new MetadataResponse(Arrays.asList(node), null, MetadataResponse.NO_CONTROLLER_ID, allTopicMetadata, version); + return new MetadataResponse(Arrays.asList(node), null, MetadataResponse.NO_CONTROLLER_ID, allTopicMetadata); } private OffsetCommitRequest createOffsetCommitRequest(int version) { @@ -425,8 +456,7 @@ public class RequestResponseTest { .setGenerationId(100) .setMemberId("consumer1") .setRetentionTime(1000000) - .setVersion((short) version) - .build(); + .build((short) version); } private OffsetCommitResponse createOffsetCommitResponse() { @@ -437,8 +467,7 @@ public class RequestResponseTest { private OffsetFetchRequest createOffsetFetchRequest(int version) { return new OffsetFetchRequest.Builder("group1", singletonList(new TopicPartition("test11", 1))) - .setVersion((short) version) - .build(); + .build((short) version); } private OffsetFetchResponse createOffsetFetchResponse() { @@ -540,8 +569,8 @@ public class RequestResponseTest { new UpdateMetadataRequest.Broker(0, endPoints1, rack), new UpdateMetadataRequest.Broker(1, endPoints2, rack) )); - return new UpdateMetadataRequest.Builder(1, 10, partitionStates, liveBrokers). - setVersion((short) version).build(); + return new UpdateMetadataRequest.Builder((short) version, 1, 10, partitionStates, + liveBrokers).build(); } private UpdateMetadataResponse createUpdateMetadataResponse() { @@ -584,14 +613,14 @@ public class RequestResponseTest { Map<String, CreateTopicsRequest.TopicDetails> request = new HashMap<>(); request.put("my_t1", request1); request.put("my_t2", request2); - return new CreateTopicsRequest.Builder(request, 0, validateOnly).setVersion((short) version).build(); + return new CreateTopicsRequest.Builder(request, 0, validateOnly).build((short) version); } - private CreateTopicsResponse createCreateTopicResponse(int version) { + private CreateTopicsResponse createCreateTopicResponse() { Map<String, CreateTopicsResponse.Error> errors = new HashMap<>(); errors.put("t1", new CreateTopicsResponse.Error(Errors.INVALID_TOPIC_EXCEPTION, null)); errors.put("t2", new CreateTopicsResponse.Error(Errors.LEADER_NOT_AVAILABLE, "Leader with id 5 is not available.")); - return new CreateTopicsResponse(errors, (short) version); + return new CreateTopicsResponse(errors); } private DeleteTopicsRequest createDeleteTopicsRequest() { http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/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 76fb9b3..3a9e0ce 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 @@ -392,7 +392,7 @@ public class SaslAuthenticatorTest { selector.send(request.toSend(node, header)); ByteBuffer responseBuffer = waitForResponse(); ResponseHeader.parse(responseBuffer); - ApiVersionsResponse response = ApiVersionsResponse.parse(responseBuffer); + ApiVersionsResponse response = ApiVersionsResponse.parse(responseBuffer, (short) 0); assertEquals(Errors.UNSUPPORTED_VERSION, response.error()); // Send ApiVersionsRequest with a supported version. This should succeed. http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/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 97da1f5..f049821 100644 --- a/core/src/main/scala/kafka/api/FetchRequest.scala +++ b/core/src/main/scala/kafka/api/FetchRequest.scala @@ -205,7 +205,7 @@ case class FetchRequest(versionId: Short = FetchRequest.CurrentVersion, responseData.put(new TopicPartition(topic, partition), new JFetchResponse.PartitionData(Errors.forException(e), -1, MemoryRecords.EMPTY)) } - val errorResponse = new JFetchResponse(versionId, responseData, 0) + val errorResponse = new JFetchResponse(responseData, 0) // Magic value does not matter here because the message set is empty requestChannel.sendResponse(new RequestChannel.Response(request, errorResponse)) } http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/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 deleted file mode 100644 index 3783c29..0000000 --- a/core/src/main/scala/kafka/api/GenericRequestAndHeader.scala +++ /dev/null @@ -1,56 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE - * file distributed with this work for additional information regarding copyright ownership. The ASF licenses this file - * to You under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the - * License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on - * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the - * specific language governing permissions and limitations under the License. - */ - -package kafka.api - -import java.nio.ByteBuffer - -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: AbstractResponse, - val name: String, - override val requestId: Option[Short] = None) - extends RequestOrResponse(requestId) { - - def writeTo(buffer: ByteBuffer) { - buffer.putShort(versionId) - buffer.putInt(correlationId) - writeShortString(buffer, clientId) - body.writeTo(buffer) - } - - def sizeInBytes(): Int = { - 2 /* version id */ + - 4 /* correlation id */ + - (2 + clientId.length) /* client id */ + - body.sizeOf() - } - - override def toString: String = { - describe(true) - } - - override def describe(details: Boolean): String = { - val strBuffer = new StringBuilder - strBuffer.append("Name: " + name) - strBuffer.append("; Version: " + versionId) - strBuffer.append("; CorrelationId: " + correlationId) - strBuffer.append("; ClientId: " + clientId) - strBuffer.append("; Body: " + body.toString) - strBuffer.toString() - } -} http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/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 deleted file mode 100644 index be0c080..0000000 --- a/core/src/main/scala/kafka/api/GenericResponseAndHeader.scala +++ /dev/null @@ -1,47 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE - * file distributed with this work for additional information regarding copyright ownership. The ASF licenses this file - * to You under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the - * License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on - * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the - * specific language governing permissions and limitations under the License. - */ - -package kafka.api - -import java.nio.ByteBuffer - -import org.apache.kafka.common.requests.AbstractResponse - -private[kafka] abstract class GenericResponseAndHeader(val correlationId: Int, - val body: AbstractResponse, - val name: String, - override val requestId: Option[Short] = None) - extends RequestOrResponse(requestId) { - - def writeTo(buffer: ByteBuffer) { - buffer.putInt(correlationId) - body.writeTo(buffer) - } - - def sizeInBytes(): Int = { - 4 /* correlation id */ + - body.sizeOf() - } - - override def toString: String = { - describe(true) - } - - override def describe(details: Boolean): String = { - val strBuffer = new StringBuilder - strBuffer.append("Name: " + name) - strBuffer.append("; CorrelationId: " + correlationId) - strBuffer.append("; Body: " + body.toString) - strBuffer.toString() - } -} http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/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 a87cdc9..bd48388 100644 --- a/core/src/main/scala/kafka/api/ProducerRequest.scala +++ b/core/src/main/scala/kafka/api/ProducerRequest.scala @@ -129,7 +129,7 @@ case class ProducerRequest(versionId: Short = ProducerRequest.CurrentVersion, } override def handleError(e: Throwable, requestChannel: RequestChannel, request: RequestChannel.Request): Unit = { - if(request.body.asInstanceOf[org.apache.kafka.common.requests.ProduceRequest].acks == 0) { + if (request.body[org.apache.kafka.common.requests.ProduceRequest].acks == 0) { requestChannel.closeConnection(request.processor, request) } else { http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/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 d8e6a95..a2fee6b 100755 --- a/core/src/main/scala/kafka/controller/ControllerChannelManager.scala +++ b/core/src/main/scala/kafka/controller/ControllerChannelManager.scala @@ -364,9 +364,9 @@ class ControllerBrokerRequestBatch(controller: KafkaController) extends Logging partitionStateInfo.allReplicas.map(Integer.valueOf).asJava) topicPartition -> partitionState } - val leaderAndIsrRequest = new LeaderAndIsrRequest. - Builder(controllerId, controllerEpoch, partitionStates.asJava, leaders.asJava) - controller.sendRequest(broker, ApiKeys.LEADER_AND_ISR, leaderAndIsrRequest, null) + val leaderAndIsrRequest = new LeaderAndIsrRequest.Builder(controllerId, controllerEpoch, partitionStates.asJava, + leaders.asJava) + controller.sendRequest(broker, ApiKeys.LEADER_AND_ISR, leaderAndIsrRequest) } leaderAndIsrRequestMap.clear() @@ -405,9 +405,8 @@ class ControllerBrokerRequestBatch(controller: KafkaController) extends Logging new UpdateMetadataRequest.Broker(broker.id, endPoints.asJava, broker.rack.orNull) } } - new UpdateMetadataRequest.Builder( - controllerId, controllerEpoch, partitionStates.asJava, liveBrokers.asJava). - setVersion(version) + new UpdateMetadataRequest.Builder(version, controllerId, controllerEpoch, partitionStates.asJava, + liveBrokers.asJava) } updateMetadataRequestBrokerSet.foreach { broker => http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/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 c063801..a5d7160 100644 --- a/core/src/main/scala/kafka/network/RequestChannel.scala +++ b/core/src/main/scala/kafka/network/RequestChannel.scala @@ -19,7 +19,7 @@ package kafka.network import java.net.InetAddress import java.nio.ByteBuffer -import java.util.HashMap +import java.util.Collections import java.util.concurrent._ import com.yammer.metrics.core.Gauge @@ -37,16 +37,18 @@ import org.apache.kafka.common.security.auth.KafkaPrincipal import org.apache.kafka.common.utils.Time import org.apache.log4j.Logger +import scala.reflect.{classTag, ClassTag} + object RequestChannel extends Logging { val AllDone = Request(processor = 1, connectionId = "2", Session(KafkaPrincipal.ANONYMOUS, InetAddress.getLocalHost), - buffer = getShutdownReceive, startTimeMs = 0, listenerName = new ListenerName(""), + buffer = shutdownReceive, startTimeMs = 0, listenerName = new ListenerName(""), securityProtocol = SecurityProtocol.PLAINTEXT) private val requestLogger = Logger.getLogger("kafka.request.logger") - private def getShutdownReceive = { - val emptyProduceRequest = new ProduceRequest.Builder(0, 0, new HashMap[TopicPartition, MemoryRecords]()).build() + private def shutdownReceive: ByteBuffer = { + val emptyProduceRequest = new ProduceRequest.Builder(0, 0, Collections.emptyMap[TopicPartition, MemoryRecords]).build() val emptyRequestHeader = new RequestHeader(ApiKeys.PRODUCE.id, emptyProduceRequest.version, "", 0) - AbstractRequestResponse.serialize(emptyRequestHeader, emptyProduceRequest) + emptyProduceRequest.serialize(emptyRequestHeader) } case class Session(principal: KafkaPrincipal, clientAddress: InetAddress) { @@ -84,12 +86,13 @@ object RequestChannel extends Logging { } } else null - val body: AbstractRequest = + val bodyAndSize: RequestAndSize = if (requestObj == null) try { // For unsupported version of ApiVersionsRequest, create a dummy request to enable an error response to be returned later - if (header.apiKey == ApiKeys.API_VERSIONS.id && !Protocol.apiVersionSupported(header.apiKey, header.apiVersion)) - new ApiVersionsRequest.Builder().build() + if (header.apiKey == ApiKeys.API_VERSIONS.id && !Protocol.apiVersionSupported(header.apiKey, header.apiVersion)) { + new RequestAndSize(new ApiVersionsRequest.Builder().build(), 0) + } else AbstractRequest.getRequest(header.apiKey, header.apiVersion, buffer) } catch { @@ -108,6 +111,14 @@ object RequestChannel extends Logging { s"$header -- $body" } + def body[T <: AbstractRequest : ClassTag] = { + bodyAndSize.request match { + case r: T => r + case r => + throw new ClassCastException(s"Expected request with type ${classTag[T].runtimeClass}, but found ${r.getClass}") + } + } + trace("Processor %d received request : %s".format(processor, requestDesc(true))) def updateRequestMetrics() { @@ -132,7 +143,7 @@ object RequestChannel extends Logging { val totalTime = endTimeMs - startTimeMs val fetchMetricNames = if (requestId == ApiKeys.FETCH.id) { - val isFromFollower = body.asInstanceOf[FetchRequest].isFromFollower + val isFromFollower = body[FetchRequest].isFromFollower Seq( if (isFromFollower) RequestMetrics.followFetchMetricName else RequestMetrics.consumerFetchMetricName @@ -163,11 +174,8 @@ object RequestChannel extends Logging { case class Response(processor: Int, request: Request, responseSend: Send, responseAction: ResponseAction) { request.responseCompleteTimeMs = Time.SYSTEM.milliseconds - def this(processor: Int, request: Request, responseSend: Send) = - this(processor, request, responseSend, if (responseSend == null) NoOpAction else SendAction) - - def this(request: Request, send: Send) = - this(request.processor, request, send) + def this(request: Request, responseSend: Send) = + this(request.processor, request, responseSend, if (responseSend == null) NoOpAction else SendAction) def this(request: Request, response: AbstractResponse) = this(request, response.toSend(request.connectionId, request.header)) http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/core/src/main/scala/kafka/server/KafkaApis.scala ---------------------------------------------------------------------- diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala index 1308216..c90cace 100644 --- a/core/src/main/scala/kafka/server/KafkaApis.scala +++ b/core/src/main/scala/kafka/server/KafkaApis.scala @@ -108,7 +108,7 @@ class KafkaApis(val requestChannel: RequestChannel, request.requestObj.handleError(e, requestChannel, request) error("Error when handling request %s".format(request.requestObj), e) } else { - val response = request.body.getErrorResponse(e) + val response = request.body[AbstractRequest].getErrorResponse(e) /* If request doesn't have a default error response, we just close the connection. For example, when produce request has acks set to 0 */ @@ -128,7 +128,7 @@ class KafkaApis(val requestChannel: RequestChannel, // We can't have the ensureTopicExists check here since the controller sends it as an advisory to all brokers so they // stop serving data to clients for the topic being deleted val correlationId = request.header.correlationId - val leaderAndIsrRequest = request.body.asInstanceOf[LeaderAndIsrRequest] + val leaderAndIsrRequest = request.body[LeaderAndIsrRequest] try { def onLeadershipChange(updatedLeaders: Iterable[Partition], updatedFollowers: Iterable[Partition]) { @@ -167,7 +167,7 @@ class KafkaApis(val requestChannel: RequestChannel, // ensureTopicExists is only for client facing requests // We can't have the ensureTopicExists check here since the controller sends it as an advisory to all brokers so they // stop serving data to clients for the topic being deleted - val stopReplicaRequest = request.body.asInstanceOf[StopReplicaRequest] + val stopReplicaRequest = request.body[StopReplicaRequest] val response = if (authorize(request.session, ClusterAction, Resource.ClusterResource)) { @@ -195,7 +195,7 @@ class KafkaApis(val requestChannel: RequestChannel, def handleUpdateMetadataRequest(request: RequestChannel.Request) { val correlationId = request.header.correlationId - val updateMetadataRequest = request.body.asInstanceOf[UpdateMetadataRequest] + val updateMetadataRequest = request.body[UpdateMetadataRequest] val updateMetadataResponse = if (authorize(request.session, ClusterAction, Resource.ClusterResource)) { @@ -235,7 +235,7 @@ class KafkaApis(val requestChannel: RequestChannel, */ def handleOffsetCommitRequest(request: RequestChannel.Request) { val header = request.header - val offsetCommitRequest = request.body.asInstanceOf[OffsetCommitRequest] + val offsetCommitRequest = request.body[OffsetCommitRequest] // reject the request if not authorized to the group if (!authorize(request.session, Read, new Resource(Group, offsetCommitRequest.groupId))) { @@ -247,14 +247,13 @@ class KafkaApis(val requestChannel: RequestChannel, requestChannel.sendResponse(new RequestChannel.Response(request, response)) } else { val (existingAndAuthorizedForDescribeTopics, nonExistingOrUnauthorizedForDescribeTopics) = offsetCommitRequest.offsetData.asScala.toMap.partition { - case (topicPartition, _) => { + case (topicPartition, _) => val authorizedForDescribe = authorize(request.session, Describe, new Resource(auth.Topic, topicPartition.topic)) val exists = metadataCache.contains(topicPartition.topic) if (!authorizedForDescribe && exists) debug(s"Offset commit request with correlation id ${header.correlationId} from client ${header.clientId} " + s"on partition $topicPartition failing due to user not having DESCRIBE authorization, but returning UNKNOWN_TOPIC_OR_PARTITION") authorizedForDescribe && exists - } } val (authorizedTopics, unauthorizedForReadTopics) = existingAndAuthorizedForDescribeTopics.partition { @@ -349,8 +348,8 @@ class KafkaApis(val requestChannel: RequestChannel, * Handle a produce request */ def handleProducerRequest(request: RequestChannel.Request) { - val produceRequest = request.body.asInstanceOf[ProduceRequest] - val numBytesAppended = request.header.sizeOf + produceRequest.sizeOf + val produceRequest = request.body[ProduceRequest] + val numBytesAppended = request.header.toStruct.sizeOf + request.bodyAndSize.size val (existingAndAuthorizedForDescribeTopics, nonExistingOrUnauthorizedForDescribeTopics) = produceRequest.partitionRecords.asScala.partition { case (topicPartition, _) => authorize(request.session, Describe, new Resource(auth.Topic, topicPartition.topic)) && metadataCache.contains(topicPartition.topic) @@ -399,14 +398,7 @@ class KafkaApis(val requestChannel: RequestChannel, requestChannel.noOperation(request.processor, request) } } else { - val respBody = request.header.apiVersion match { - case 0 => new ProduceResponse(mergedResponseStatus.asJava) - case version@(1 | 2) => new ProduceResponse(mergedResponseStatus.asJava, delayTimeMs, version) - // This case shouldn't happen unless a new version of ProducerRequest is added without - // updating this part of the code to handle it properly. - case version => throw new IllegalArgumentException(s"Version `$version` of ProduceRequest is not handled. Code must be updated.") - } - + val respBody = new ProduceResponse(mergedResponseStatus.asJava, delayTimeMs) requestChannel.sendResponse(new RequestChannel.Response(request, respBody)) } } @@ -445,7 +437,7 @@ class KafkaApis(val requestChannel: RequestChannel, * Handle a fetch request */ def handleFetchRequest(request: RequestChannel.Request) { - val fetchRequest = request.body.asInstanceOf[FetchRequest] + val fetchRequest = request.body[FetchRequest] val versionId = request.header.apiVersion val clientId = request.header.clientId @@ -505,13 +497,13 @@ class KafkaApis(val requestChannel: RequestChannel, BrokerTopicStats.getBrokerAllTopicsStats().bytesOutRate.mark(data.records.sizeInBytes) } - val response = new FetchResponse(versionId, fetchedPartitionData, 0) + val response = new FetchResponse(fetchedPartitionData, 0) + val responseStruct = response.toStruct(versionId) - def fetchResponseCallback(delayTimeMs: Int) { - trace(s"Sending fetch response to client $clientId of " + - s"${convertedPartitionData.map { case (_, v) => v.records.sizeInBytes }.sum} bytes") - val fetchResponse = if (delayTimeMs > 0) new FetchResponse(versionId, fetchedPartitionData, delayTimeMs) else response - requestChannel.sendResponse(new RequestChannel.Response(request, fetchResponse)) + def fetchResponseCallback(throttleTimeMs: Int) { + trace(s"Sending fetch response to client $clientId of ${responseStruct.sizeOf} bytes.") + val responseSend = response.toSend(responseStruct, throttleTimeMs, request.connectionId, request.header) + requestChannel.sendResponse(new RequestChannel.Response(request, responseSend)) } // When this callback is triggered, the remote API call has completed @@ -521,9 +513,10 @@ class KafkaApis(val requestChannel: RequestChannel, // We've already evaluated against the quota and are good to go. Just need to record it now. val responseSize = sizeOfThrottledPartitions(versionId, fetchRequest, mergedPartitionData, quotas.leader) quotas.leader.record(responseSize) - fetchResponseCallback(0) + fetchResponseCallback(throttleTimeMs = 0) } else { - quotas.fetch.recordAndMaybeThrottle(request.session.sanitizedUser, clientId, response.sizeOf, fetchResponseCallback) + quotas.fetch.recordAndMaybeThrottle(request.session.sanitizedUser, clientId, responseStruct.sizeOf, + fetchResponseCallback) } } @@ -547,7 +540,7 @@ class KafkaApis(val requestChannel: RequestChannel, fetchRequest: FetchRequest, mergedPartitionData: Seq[(TopicPartition, FetchResponse.PartitionData)], quota: ReplicationQuotaManager): Int = { - val partitionData = new util.LinkedHashMap[TopicPartition, FetchResponse.PartitionData]() + val partitionData = new util.LinkedHashMap[TopicPartition, FetchResponse.PartitionData] mergedPartitionData.foreach { case (tp, data) => if (quota.isThrottled(tp)) partitionData.put(tp, data) @@ -570,14 +563,14 @@ class KafkaApis(val requestChannel: RequestChannel, else handleListOffsetRequestV1(request) - val response = new ListOffsetResponse(mergedResponseMap.asJava, version) + val response = new ListOffsetResponse(mergedResponseMap.asJava) requestChannel.sendResponse(new RequestChannel.Response(request, response)) } private def handleListOffsetRequestV0(request : RequestChannel.Request) : Map[TopicPartition, ListOffsetResponse.PartitionData] = { val correlationId = request.header.correlationId val clientId = request.header.clientId - val offsetRequest = request.body.asInstanceOf[ListOffsetRequest] + val offsetRequest = request.body[ListOffsetRequest] val (authorizedRequestInfo, unauthorizedRequestInfo) = offsetRequest.offsetData.asScala.partition { case (topicPartition, _) => authorize(request.session, Describe, new Resource(auth.Topic, topicPartition.topic)) @@ -628,7 +621,7 @@ class KafkaApis(val requestChannel: RequestChannel, private def handleListOffsetRequestV1(request : RequestChannel.Request): Map[TopicPartition, ListOffsetResponse.PartitionData] = { val correlationId = request.header.correlationId val clientId = request.header.clientId - val offsetRequest = request.body.asInstanceOf[ListOffsetRequest] + val offsetRequest = request.body[ListOffsetRequest] val (authorizedRequestInfo, unauthorizedRequestInfo) = offsetRequest.partitionTimestamps.asScala.partition { case (topicPartition, _) => authorize(request.session, Describe, new Resource(auth.Topic, topicPartition.topic)) @@ -824,7 +817,7 @@ class KafkaApis(val requestChannel: RequestChannel, * Handle a topic metadata request */ def handleTopicMetadataRequest(request: RequestChannel.Request) { - val metadataRequest = request.body.asInstanceOf[MetadataRequest] + val metadataRequest = request.body[MetadataRequest] val requestVersion = request.header.apiVersion() val topics = @@ -889,8 +882,7 @@ class KafkaApis(val requestChannel: RequestChannel, brokers.map(_.getNode(request.listenerName)).asJava, clusterId, metadataCache.getControllerId.getOrElse(MetadataResponse.NO_CONTROLLER_ID), - completeTopicMetadata.asJava, - requestVersion + completeTopicMetadata.asJava ) requestChannel.sendResponse(new RequestChannel.Response(request, responseBody)) } @@ -900,7 +892,7 @@ class KafkaApis(val requestChannel: RequestChannel, */ def handleOffsetFetchRequest(request: RequestChannel.Request) { val header = request.header - val offsetFetchRequest = request.body.asInstanceOf[OffsetFetchRequest] + val offsetFetchRequest = request.body[OffsetFetchRequest] def authorizeTopicDescribe(partition: TopicPartition) = authorize(request.session, Describe, new Resource(auth.Topic, partition.topic)) @@ -938,7 +930,7 @@ class KafkaApis(val requestChannel: RequestChannel, }.toMap val unauthorizedPartitionData = unauthorizedPartitions.map(_ -> OffsetFetchResponse.UNKNOWN_PARTITION).toMap - new OffsetFetchResponse(Errors.NONE, (authorizedPartitionData ++ unauthorizedPartitionData).asJava, header.apiVersion) + new OffsetFetchResponse(Errors.NONE, (authorizedPartitionData ++ unauthorizedPartitionData).asJava) } else { // versions 1 and above read offsets from Kafka if (offsetFetchRequest.isAllPartitions) { @@ -948,7 +940,7 @@ class KafkaApis(val requestChannel: RequestChannel, else { // clients are not allowed to see offsets for topics that are not authorized for Describe val authorizedPartitionData = allPartitionData.filter { case (topicPartition, _) => authorizeTopicDescribe(topicPartition) } - new OffsetFetchResponse(Errors.NONE, authorizedPartitionData.asJava, header.apiVersion) + new OffsetFetchResponse(Errors.NONE, authorizedPartitionData.asJava) } } else { val (authorizedPartitions, unauthorizedPartitions) = offsetFetchRequest.partitions.asScala @@ -959,7 +951,7 @@ class KafkaApis(val requestChannel: RequestChannel, offsetFetchRequest.getErrorResponse(error) else { val unauthorizedPartitionData = unauthorizedPartitions.map(_ -> OffsetFetchResponse.UNKNOWN_PARTITION).toMap - new OffsetFetchResponse(Errors.NONE, (authorizedPartitionData ++ unauthorizedPartitionData).asJava, header.apiVersion) + new OffsetFetchResponse(Errors.NONE, (authorizedPartitionData ++ unauthorizedPartitionData).asJava) } } } @@ -970,7 +962,7 @@ class KafkaApis(val requestChannel: RequestChannel, } def handleGroupCoordinatorRequest(request: RequestChannel.Request) { - val groupCoordinatorRequest = request.body.asInstanceOf[GroupCoordinatorRequest] + val groupCoordinatorRequest = request.body[GroupCoordinatorRequest] if (!authorize(request.session, Describe, new Resource(Group, groupCoordinatorRequest.groupId))) { val responseBody = new GroupCoordinatorResponse(Errors.GROUP_AUTHORIZATION_FAILED, Node.noNode) @@ -1003,7 +995,7 @@ class KafkaApis(val requestChannel: RequestChannel, } def handleDescribeGroupRequest(request: RequestChannel.Request) { - val describeRequest = request.body.asInstanceOf[DescribeGroupsRequest] + val describeRequest = request.body[DescribeGroupsRequest] val groups = describeRequest.groupIds().asScala.map { groupId => if (!authorize(request.session, Describe, new Resource(Group, groupId))) { @@ -1036,12 +1028,12 @@ class KafkaApis(val requestChannel: RequestChannel, } def handleJoinGroupRequest(request: RequestChannel.Request) { - val joinGroupRequest = request.body.asInstanceOf[JoinGroupRequest] + val joinGroupRequest = request.body[JoinGroupRequest] // the callback for sending a join-group response def sendResponseCallback(joinResult: JoinGroupResult) { val members = joinResult.members map { case (memberId, metadataArray) => (memberId, ByteBuffer.wrap(metadataArray)) } - val responseBody = new JoinGroupResponse(request.header.apiVersion, joinResult.error, joinResult.generationId, + val responseBody = new JoinGroupResponse(joinResult.error, joinResult.generationId, joinResult.subProtocol, joinResult.memberId, joinResult.leaderId, members.asJava) trace("Sending join group response %s for correlation id %d to client %s." @@ -1051,7 +1043,6 @@ class KafkaApis(val requestChannel: RequestChannel, if (!authorize(request.session, Read, new Resource(Group, joinGroupRequest.groupId()))) { val responseBody = new JoinGroupResponse( - request.header.apiVersion, Errors.GROUP_AUTHORIZATION_FAILED, JoinGroupResponse.UNKNOWN_GENERATION_ID, JoinGroupResponse.UNKNOWN_PROTOCOL, @@ -1077,7 +1068,7 @@ class KafkaApis(val requestChannel: RequestChannel, } def handleSyncGroupRequest(request: RequestChannel.Request) { - val syncGroupRequest = request.body.asInstanceOf[SyncGroupRequest] + val syncGroupRequest = request.body[SyncGroupRequest] def sendResponseCallback(memberState: Array[Byte], error: Errors) { val responseBody = new SyncGroupResponse(error, ByteBuffer.wrap(memberState)) @@ -1098,7 +1089,7 @@ class KafkaApis(val requestChannel: RequestChannel, } def handleHeartbeatRequest(request: RequestChannel.Request) { - val heartbeatRequest = request.body.asInstanceOf[HeartbeatRequest] + val heartbeatRequest = request.body[HeartbeatRequest] // the callback for sending a heartbeat response def sendResponseCallback(error: Errors) { @@ -1123,7 +1114,7 @@ class KafkaApis(val requestChannel: RequestChannel, } def handleLeaveGroupRequest(request: RequestChannel.Request) { - val leaveGroupRequest = request.body.asInstanceOf[LeaveGroupRequest] + val leaveGroupRequest = request.body[LeaveGroupRequest] // the callback for sending a leave-group response def sendResponseCallback(error: Errors) { @@ -1157,11 +1148,11 @@ class KafkaApis(val requestChannel: RequestChannel, // If this is considered to leak information about the broker version a workaround is to use SSL // with client authentication which is performed at an earlier stage of the connection where the // ApiVersionRequest is not available. - val responseBody = if (Protocol.apiVersionSupported(ApiKeys.API_VERSIONS.id, request.header.apiVersion)) - ApiVersionsResponse.API_VERSIONS_RESPONSE - else - ApiVersionsResponse.fromError(Errors.UNSUPPORTED_VERSION) - requestChannel.sendResponse(new RequestChannel.Response(request, responseBody)) + val responseSend = + if (Protocol.apiVersionSupported(ApiKeys.API_VERSIONS.id, request.header.apiVersion)) + ApiVersionsResponse.API_VERSIONS_RESPONSE.toSend(request.connectionId, request.header) + else ApiVersionsResponse.unsupportedVersionSend(request.connectionId, request.header) + requestChannel.sendResponse(new RequestChannel.Response(request, responseSend)) } def close() { @@ -1170,10 +1161,10 @@ class KafkaApis(val requestChannel: RequestChannel, } def handleCreateTopicsRequest(request: RequestChannel.Request) { - val createTopicsRequest = request.body.asInstanceOf[CreateTopicsRequest] + val createTopicsRequest = request.body[CreateTopicsRequest] def sendResponseCallback(results: Map[String, CreateTopicsResponse.Error]): Unit = { - val responseBody = new CreateTopicsResponse(results.asJava, request.header.apiVersion) + val responseBody = new CreateTopicsResponse(results.asJava) trace(s"Sending create topics response $responseBody for correlation id ${request.header.correlationId} to client ${request.header.clientId}.") requestChannel.sendResponse(new RequestChannel.Response(request, responseBody)) } @@ -1220,7 +1211,7 @@ class KafkaApis(val requestChannel: RequestChannel, } def handleDeleteTopicsRequest(request: RequestChannel.Request) { - val deleteTopicRequest = request.body.asInstanceOf[DeleteTopicsRequest] + val deleteTopicRequest = request.body[DeleteTopicsRequest] val (existingAndAuthorizedForDescribeTopics, nonExistingOrUnauthorizedForDescribeTopics) = deleteTopicRequest.topics.asScala.partition { topic => authorize(request.session, Describe, new Resource(auth.Topic, topic)) && metadataCache.contains(topic) http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala ---------------------------------------------------------------------- diff --git a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala b/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala index c99d7c5..7fb02a3 100644 --- a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala +++ b/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala @@ -261,17 +261,13 @@ class ReplicaFetcherThread(name: String, } - private def earliestOrLatestOffset(topicPartition: TopicPartition, earliestOrLatest: Long, consumerId: Int): Long = { + private def earliestOrLatestOffset(topicPartition: TopicPartition, earliestOrLatest: Long, replicaId: Int): Long = { val requestBuilder = if (brokerConfig.interBrokerProtocolVersion >= KAFKA_0_10_1_IV2) { val partitions = Map(topicPartition -> (earliestOrLatest: java.lang.Long)) - new ListOffsetRequest.Builder(consumerId). - setTargetTimes(partitions.asJava). - setVersion(1) + ListOffsetRequest.Builder.forReplica(1, replicaId).setTargetTimes(partitions.asJava) } else { val partitions = Map(topicPartition -> new ListOffsetRequest.PartitionData(earliestOrLatest, 1)) - new ListOffsetRequest.Builder(consumerId). - setOffsetData(partitions.asJava). - setVersion(0) + ListOffsetRequest.Builder.forReplica(0, replicaId).setOffsetData(partitions.asJava) } val clientResponse = sendRequest(requestBuilder) val response = clientResponse.responseBody.asInstanceOf[ListOffsetResponse] @@ -295,9 +291,8 @@ class ReplicaFetcherThread(name: String, requestMap.put(topicPartition, new JFetchRequest.PartitionData(partitionFetchState.offset, fetchSize)) } - val requestBuilder = new JFetchRequest.Builder(maxWait, minBytes, requestMap). - setReplicaId(replicaId).setMaxBytes(maxBytes) - requestBuilder.setVersion(fetchRequestVersion) + val requestBuilder = JFetchRequest.Builder.forReplica(fetchRequestVersion, replicaId, maxWait, minBytes, requestMap) + .setMaxBytes(maxBytes) new FetchRequest(requestBuilder) } http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala index 3285bf2..4f71258 100644 --- a/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala +++ b/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala @@ -27,7 +27,7 @@ import org.apache.kafka.clients.consumer.internals.NoOpConsumerRebalanceListener import org.apache.kafka.clients.consumer._ import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord} import org.apache.kafka.common.errors._ -import org.apache.kafka.common.protocol.{ApiKeys, Errors, SecurityProtocol} +import org.apache.kafka.common.protocol.{ApiKeys, Errors, ProtoUtils, SecurityProtocol} import org.apache.kafka.common.requests._ import CreateTopicsRequest.TopicDetails import org.apache.kafka.common.security.auth.KafkaPrincipal @@ -191,11 +191,12 @@ class AuthorizerIntegrationTest extends BaseRequestTest { private def createFetchRequest = { val partitionMap = new util.LinkedHashMap[TopicPartition, requests.FetchRequest.PartitionData] partitionMap.put(tp, new requests.FetchRequest.PartitionData(0, 100)) - new requests.FetchRequest.Builder(100, Int.MaxValue, partitionMap).setReplicaId(5000).build() + val version = ProtoUtils.latestVersion(ApiKeys.FETCH.id) + requests.FetchRequest.Builder.forReplica(version, 5000, 100, Int.MaxValue, partitionMap).build() } private def createListOffsetsRequest = { - new requests.ListOffsetRequest.Builder().setTargetTimes( + requests.ListOffsetRequest.Builder.forConsumer(0).setTargetTimes( Map(tp -> (0L: java.lang.Long)).asJava). build() } @@ -214,7 +215,8 @@ class AuthorizerIntegrationTest extends BaseRequestTest { val brokers = Set(new requests.UpdateMetadataRequest.Broker(brokerId, Seq(new requests.UpdateMetadataRequest.EndPoint("localhost", 0, securityProtocol, ListenerName.forSecurityProtocol(securityProtocol))).asJava, null)).asJava - new requests.UpdateMetadataRequest.Builder(brokerId, Int.MaxValue, partitionState, brokers).build() + val version = ProtoUtils.latestVersion(ApiKeys.UPDATE_METADATA_KEY.id) + new requests.UpdateMetadataRequest.Builder(version, brokerId, Int.MaxValue, partitionState, brokers).build() } private def createJoinGroupRequest = { @@ -770,17 +772,18 @@ class AuthorizerIntegrationTest extends BaseRequestTest { @Test def testUnauthorizedDeleteWithoutDescribe() { - val response = send(deleteTopicsRequest, ApiKeys.DELETE_TOPICS) - val deleteResponse = DeleteTopicsResponse.parse(response) - + val response = connectAndSend(deleteTopicsRequest, ApiKeys.DELETE_TOPICS) + val version = ProtoUtils.latestVersion(ApiKeys.DELETE_TOPICS.id) + val deleteResponse = DeleteTopicsResponse.parse(response, version) assertEquals(Errors.UNKNOWN_TOPIC_OR_PARTITION, deleteResponse.errors.asScala.head._2) } @Test def testUnauthorizedDeleteWithDescribe() { addAndVerifyAcls(Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, Describe)), deleteTopicResource) - val response = send(deleteTopicsRequest, ApiKeys.DELETE_TOPICS) - val deleteResponse = DeleteTopicsResponse.parse(response) + val response = connectAndSend(deleteTopicsRequest, ApiKeys.DELETE_TOPICS) + val version = ProtoUtils.latestVersion(ApiKeys.DELETE_TOPICS.id) + val deleteResponse = DeleteTopicsResponse.parse(response, version) assertEquals(Errors.TOPIC_AUTHORIZATION_FAILED, deleteResponse.errors.asScala.head._2) } @@ -788,8 +791,9 @@ class AuthorizerIntegrationTest extends BaseRequestTest { @Test def testDeleteWithWildCardAuth() { addAndVerifyAcls(Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, Delete)), new Resource(Topic, "*")) - val response = send(deleteTopicsRequest, ApiKeys.DELETE_TOPICS) - val deleteResponse = DeleteTopicsResponse.parse(response) + val response = connectAndSend(deleteTopicsRequest, ApiKeys.DELETE_TOPICS) + val version = ProtoUtils.latestVersion(ApiKeys.DELETE_TOPICS.id) + val deleteResponse = DeleteTopicsResponse.parse(response, version) assertEquals(Errors.NONE, deleteResponse.errors.asScala.head._2) } @@ -807,8 +811,9 @@ class AuthorizerIntegrationTest extends BaseRequestTest { isAuthorized: Boolean, isAuthorizedTopicDescribe: Boolean, topicExists: Boolean = true): AbstractResponse = { - val resp = send(request, apiKey) - val response = RequestKeyToResponseDeserializer(apiKey).getMethod("parse", classOf[ByteBuffer]).invoke(null, resp).asInstanceOf[AbstractResponse] + val resp = connectAndSend(request, apiKey) + val response = RequestKeyToResponseDeserializer(apiKey).getMethod("parse", classOf[ByteBuffer], classOf[Short]).invoke( + null, resp, request.version: java.lang.Short).asInstanceOf[AbstractResponse] val error = RequestKeyToError(apiKey).asInstanceOf[(AbstractResponse) => Errors](response) val authorizationErrorCodes = resources.flatMap { resourceType => @@ -877,7 +882,7 @@ class AuthorizerIntegrationTest extends BaseRequestTest { private def sendOffsetFetchRequest(request: requests.OffsetFetchRequest, socketServer: SocketServer): requests.OffsetFetchResponse = { - val response = send(request, ApiKeys.OFFSET_FETCH, socketServer) + val response = connectAndSend(request, ApiKeys.OFFSET_FETCH, socketServer) requests.OffsetFetchResponse.parse(response, request.version) } http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/core/src/test/scala/unit/kafka/network/SocketServerTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala index 37bc238..3875604 100644 --- a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala +++ b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala @@ -33,7 +33,7 @@ import org.apache.kafka.common.metrics.Metrics import org.apache.kafka.common.network.{ListenerName, NetworkSend} import org.apache.kafka.common.protocol.{ApiKeys, SecurityProtocol} import org.apache.kafka.common.record.MemoryRecords -import org.apache.kafka.common.requests.{ProduceRequest, RequestHeader} +import org.apache.kafka.common.requests.{AbstractRequest, ProduceRequest, RequestHeader} import org.apache.kafka.common.security.auth.KafkaPrincipal import org.apache.kafka.common.utils.Time import org.junit.Assert._ @@ -89,13 +89,11 @@ class SocketServerTest extends JUnitSuite { } def processRequest(channel: RequestChannel, request: RequestChannel.Request) { - val byteBuffer = ByteBuffer.allocate(request.header.sizeOf + request.body.sizeOf) - request.header.writeTo(byteBuffer) - request.body.writeTo(byteBuffer) + val byteBuffer = request.body[AbstractRequest].serialize(request.header) byteBuffer.rewind() val send = new NetworkSend(request.connectionId, byteBuffer) - channel.sendResponse(new RequestChannel.Response(request.processor, request, send)) + channel.sendResponse(new RequestChannel.Response(request, send)) } def connect(s: SocketServer = server, protocol: SecurityProtocol = SecurityProtocol.PLAINTEXT) = { @@ -119,14 +117,11 @@ class SocketServerTest extends JUnitSuite { val ackTimeoutMs = 10000 val ack = 0: Short - val emptyRequest = new ProduceRequest.Builder( - ack, ackTimeoutMs, new HashMap[TopicPartition, MemoryRecords]()).build() + val emptyRequest = new ProduceRequest.Builder(ack, ackTimeoutMs, new HashMap[TopicPartition, MemoryRecords]()).build() val emptyHeader = new RequestHeader(apiKey, emptyRequest.version, clientId, correlationId) - - val byteBuffer = ByteBuffer.allocate(emptyHeader.sizeOf + emptyRequest.sizeOf) - emptyHeader.writeTo(byteBuffer) - emptyRequest.writeTo(byteBuffer) + val byteBuffer = emptyRequest.serialize(emptyHeader) byteBuffer.rewind() + val serializedBytes = new Array[Byte](byteBuffer.remaining) byteBuffer.get(serializedBytes) serializedBytes @@ -289,13 +284,10 @@ class SocketServerTest extends JUnitSuite { val clientId = "" val ackTimeoutMs = 10000 val ack = 0: Short - val emptyRequest = new ProduceRequest.Builder( - ack, ackTimeoutMs, new HashMap[TopicPartition, MemoryRecords]()).build() + val emptyRequest = new ProduceRequest.Builder(ack, ackTimeoutMs, new HashMap[TopicPartition, MemoryRecords]()).build() val emptyHeader = new RequestHeader(apiKey, emptyRequest.version, clientId, correlationId) - val byteBuffer = ByteBuffer.allocate(emptyHeader.sizeOf() + emptyRequest.sizeOf()) - emptyHeader.writeTo(byteBuffer) - emptyRequest.writeTo(byteBuffer) + val byteBuffer = emptyRequest.serialize(emptyHeader) byteBuffer.rewind() val serializedBytes = new Array[Byte](byteBuffer.remaining) byteBuffer.get(serializedBytes) @@ -355,7 +347,7 @@ class SocketServerTest extends JUnitSuite { // detected. If the buffer is larger than 102400 bytes, a second write is attempted and it fails with an // IOException. val send = new NetworkSend(request.connectionId, ByteBuffer.allocate(550000)) - channel.sendResponse(new RequestChannel.Response(request.processor, request, send)) + channel.sendResponse(new RequestChannel.Response(request, send)) TestUtils.waitUntilTrue(() => totalTimeHistCount() == expectedTotalTimeCount, s"request metrics not updated, expected: $expectedTotalTimeCount, actual: ${totalTimeHistCount()}") http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/core/src/test/scala/unit/kafka/server/AbstractCreateTopicsRequestTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/unit/kafka/server/AbstractCreateTopicsRequestTest.scala b/core/src/test/scala/unit/kafka/server/AbstractCreateTopicsRequestTest.scala index ef98531..448fce1 100644 --- a/core/src/test/scala/unit/kafka/server/AbstractCreateTopicsRequestTest.scala +++ b/core/src/test/scala/unit/kafka/server/AbstractCreateTopicsRequestTest.scala @@ -82,13 +82,13 @@ class AbstractCreateTopicsRequestTest extends BaseRequestTest { protected def error(error: Errors, errorMessage: Option[String] = None): CreateTopicsResponse.Error = new CreateTopicsResponse.Error(error, errorMessage.orNull) - protected def duplicateFirstTopic(request: CreateTopicsRequest) = { + protected def toStructWithDuplicateFirstTopic(request: CreateTopicsRequest): Struct = { val struct = request.toStruct val topics = struct.getArray("create_topic_requests") val firstTopic = topics(0).asInstanceOf[Struct] val newTopics = firstTopic :: topics.toList struct.set("create_topic_requests", newTopics.toArray) - new CreateTopicsRequest(struct, request.version) + struct } protected def addPartitionsAndReplicationFactorToFirstTopic(request: CreateTopicsRequest) = { @@ -102,8 +102,10 @@ class AbstractCreateTopicsRequestTest extends BaseRequestTest { protected def validateErrorCreateTopicsRequests(request: CreateTopicsRequest, expectedResponse: Map[String, CreateTopicsResponse.Error], - checkErrorMessage: Boolean = true): Unit = { - val response = sendCreateTopicRequest(request) + checkErrorMessage: Boolean = true, + requestStruct: Option[Struct] = None): Unit = { + val response = requestStruct.map(sendCreateTopicRequestStruct(_, request.version)).getOrElse( + sendCreateTopicRequest(request)) val errors = response.errors.asScala assertEquals("The response size should match", expectedResponse.size, response.errors.size) @@ -133,14 +135,20 @@ class AbstractCreateTopicsRequestTest extends BaseRequestTest { assignments.map { case (k, v) => (k: Integer, v.map { i => i: Integer }.asJava) }.asJava } + protected def sendCreateTopicRequestStruct(requestStruct: Struct, apiVersion: Short, + socketServer: SocketServer = controllerSocketServer): CreateTopicsResponse = { + val response = connectAndSendStruct(requestStruct, ApiKeys.CREATE_TOPICS, apiVersion, socketServer) + CreateTopicsResponse.parse(response, apiVersion) + } + protected def sendCreateTopicRequest(request: CreateTopicsRequest, socketServer: SocketServer = controllerSocketServer): CreateTopicsResponse = { - val response = send(request, ApiKeys.CREATE_TOPICS, socketServer) + val response = connectAndSend(request, ApiKeys.CREATE_TOPICS, socketServer) CreateTopicsResponse.parse(response, request.version) } protected def sendMetadataRequest(request: MetadataRequest, destination: SocketServer = anySocketServer): MetadataResponse = { val version = ProtoUtils.latestVersion(ApiKeys.METADATA.id) - val response = send(request, ApiKeys.METADATA, destination = destination) + val response = connectAndSend(request, ApiKeys.METADATA, destination = destination) MetadataResponse.parse(response, version) } http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/core/src/test/scala/unit/kafka/server/ApiVersionsRequestTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/unit/kafka/server/ApiVersionsRequestTest.scala b/core/src/test/scala/unit/kafka/server/ApiVersionsRequestTest.scala index ffe82d1..248b91e 100644 --- a/core/src/test/scala/unit/kafka/server/ApiVersionsRequestTest.scala +++ b/core/src/test/scala/unit/kafka/server/ApiVersionsRequestTest.scala @@ -17,8 +17,7 @@ package kafka.server -import org.apache.kafka.common.protocol.types.Struct -import org.apache.kafka.common.protocol.{ApiKeys, Errors, ProtoUtils} +import org.apache.kafka.common.protocol.{ApiKeys, Errors} import org.apache.kafka.common.requests.ApiVersionsResponse.ApiVersion import org.apache.kafka.common.requests.{ApiVersionsRequest, ApiVersionsResponse} import org.junit.Assert._ @@ -51,14 +50,13 @@ class ApiVersionsRequestTest extends BaseRequestTest { @Test def testApiVersionsRequestWithUnsupportedVersion() { - val apiVersionsRequest = new ApiVersionsRequest( - new Struct(ProtoUtils.currentRequestSchema(ApiKeys.API_VERSIONS.id)), Short.MaxValue) - val apiVersionsResponse = sendApiVersionsRequest(apiVersionsRequest) + val apiVersionsRequest = new ApiVersionsRequest(0) + val apiVersionsResponse = sendApiVersionsRequest(apiVersionsRequest, Some(Short.MaxValue)) assertEquals(Errors.UNSUPPORTED_VERSION, apiVersionsResponse.error) } - private def sendApiVersionsRequest(request: ApiVersionsRequest): ApiVersionsResponse = { - val response = send(request, ApiKeys.API_VERSIONS) - ApiVersionsResponse.parse(response) + private def sendApiVersionsRequest(request: ApiVersionsRequest, apiVersion: Option[Short] = None): ApiVersionsResponse = { + val response = connectAndSend(request, ApiKeys.API_VERSIONS, apiVersion = apiVersion) + ApiVersionsResponse.parse(response, 0) } }
