http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java index 64bd3d3..dee6a5f 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java @@ -23,7 +23,7 @@ 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.types.Schema; + import org.apache.kafka.common.protocol.types.Struct; import org.apache.kafka.common.protocol.types.Type; import org.apache.kafka.common.record.Records; @@ -39,7 +39,6 @@ import java.util.Map; */ public class FetchResponse extends AbstractResponse { - private static final Schema CURRENT_SCHEMA = ProtoUtils.currentResponseSchema(ApiKeys.FETCH.id); private static final String RESPONSES_KEY_NAME = "responses"; // topic level field names @@ -71,7 +70,7 @@ public class FetchResponse extends AbstractResponse { public static final long INVALID_HIGHWATERMARK = -1L; private final LinkedHashMap<TopicPartition, PartitionData> responseData; - private final int throttleTime; + private final int throttleTimeMs; public static final class PartitionData { public final Errors error; @@ -92,35 +91,20 @@ public class FetchResponse extends AbstractResponse { } /** - * Constructor for version 3. - * - * The entries in `responseData` should be in the same order as the entries in `FetchRequest.fetchData`. - * - * @param responseData fetched data grouped by topic-partition - * @param throttleTime Time in milliseconds the response was throttled - */ - public FetchResponse(LinkedHashMap<TopicPartition, PartitionData> responseData, int throttleTime) { - this(3, responseData, throttleTime); - } - - /** * Constructor for all versions. * * From version 3, the entries in `responseData` should be in the same order as the entries in * `FetchRequest.fetchData`. * * @param responseData fetched data grouped by topic-partition - * @param throttleTime Time in milliseconds the response was throttled + * @param throttleTimeMs Time in milliseconds the response was throttled */ - public FetchResponse(int version, LinkedHashMap<TopicPartition, PartitionData> responseData, int throttleTime) { - super(writeStruct(new Struct(ProtoUtils.responseSchema(ApiKeys.FETCH.id, version)), version, responseData, - throttleTime)); + public FetchResponse(LinkedHashMap<TopicPartition, PartitionData> responseData, int throttleTimeMs) { this.responseData = responseData; - this.throttleTime = throttleTime; + this.throttleTimeMs = throttleTimeMs; } public FetchResponse(Struct struct) { - super(struct); LinkedHashMap<TopicPartition, PartitionData> responseData = new LinkedHashMap<>(); for (Object topicResponseObj : struct.getArray(RESPONSES_KEY_NAME)) { Struct topicResponse = (Struct) topicResponseObj; @@ -137,22 +121,31 @@ public class FetchResponse extends AbstractResponse { } } this.responseData = responseData; - this.throttleTime = struct.hasField(THROTTLE_TIME_KEY_NAME) ? struct.getInt(THROTTLE_TIME_KEY_NAME) : DEFAULT_THROTTLE_TIME; + this.throttleTimeMs = struct.hasField(THROTTLE_TIME_KEY_NAME) ? struct.getInt(THROTTLE_TIME_KEY_NAME) : DEFAULT_THROTTLE_TIME; + } + + @Override + public Struct toStruct(short version) { + return toStruct(version, responseData, throttleTimeMs); } @Override public Send toSend(String dest, RequestHeader requestHeader) { - ResponseHeader responseHeader = new ResponseHeader(requestHeader.correlationId()); + return toSend(toStruct(requestHeader.apiVersion()), throttleTimeMs, dest, requestHeader); + } + + public Send toSend(Struct responseStruct, int throttleTimeMs, String dest, RequestHeader requestHeader) { + Struct responseHeader = new ResponseHeader(requestHeader.correlationId()).toStruct(); // write the total size and the response header ByteBuffer buffer = ByteBuffer.allocate(responseHeader.sizeOf() + 4); - buffer.putInt(responseHeader.sizeOf() + struct.sizeOf()); + buffer.putInt(responseHeader.sizeOf() + responseStruct.sizeOf()); responseHeader.writeTo(buffer); buffer.rewind(); List<Send> sends = new ArrayList<>(); sends.add(new ByteBufferSend(dest, buffer)); - addResponseData(dest, sends); + addResponseData(responseStruct, throttleTimeMs, dest, sends); return new MultiSend(dest, sends); } @@ -160,25 +153,20 @@ public class FetchResponse extends AbstractResponse { return responseData; } - public int getThrottleTime() { - return this.throttleTime; + public int throttleTimeMs() { + return this.throttleTimeMs; } - public static FetchResponse parse(ByteBuffer buffer) { - return new FetchResponse(CURRENT_SCHEMA.read(buffer)); - } - - public static FetchResponse parse(ByteBuffer buffer, int version) { + public static FetchResponse parse(ByteBuffer buffer, short version) { return new FetchResponse(ProtoUtils.responseSchema(ApiKeys.FETCH.id, version).read(buffer)); } - private void addResponseData(String dest, List<Send> sends) { + private static void addResponseData(Struct struct, int throttleTimeMs, String dest, List<Send> sends) { Object[] allTopicData = struct.getArray(RESPONSES_KEY_NAME); if (struct.hasField(THROTTLE_TIME_KEY_NAME)) { - int throttleTime = struct.getInt(THROTTLE_TIME_KEY_NAME); ByteBuffer buffer = ByteBuffer.allocate(8); - buffer.putInt(throttleTime); + buffer.putInt(throttleTimeMs); buffer.putInt(allTopicData.length); buffer.rewind(); sends.add(new ByteBufferSend(dest, buffer)); @@ -193,7 +181,7 @@ public class FetchResponse extends AbstractResponse { addTopicData(dest, sends, (Struct) topicData); } - private void addTopicData(String dest, List<Send> sends, Struct topicData) { + private static void addTopicData(String dest, List<Send> sends, Struct topicData) { String topic = topicData.getString(TOPIC_KEY_NAME); Object[] allPartitionData = topicData.getArray(PARTITIONS_KEY_NAME); @@ -208,7 +196,7 @@ public class FetchResponse extends AbstractResponse { addPartitionData(dest, sends, (Struct) partitionData); } - private void addPartitionData(String dest, List<Send> sends, Struct partitionData) { + private static void addPartitionData(String dest, List<Send> sends, Struct partitionData) { Struct header = partitionData.getStruct(PARTITION_HEADER_KEY_NAME); Records records = partitionData.getRecords(RECORD_SET_KEY_NAME); @@ -223,10 +211,8 @@ public class FetchResponse extends AbstractResponse { sends.add(new RecordsSend(dest, records)); } - private static Struct writeStruct(Struct struct, - int version, - LinkedHashMap<TopicPartition, PartitionData> responseData, - int throttleTime) { + private static Struct toStruct(short version, LinkedHashMap<TopicPartition, PartitionData> responseData, int throttleTime) { + Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.FETCH.id, version)); List<FetchRequest.TopicAndPartitionData<PartitionData>> topicsData = FetchRequest.TopicAndPartitionData.batchByTopic(responseData); List<Struct> topicArray = new ArrayList<>(); for (FetchRequest.TopicAndPartitionData<PartitionData> topicEntry: topicsData) { @@ -255,10 +241,8 @@ public class FetchResponse extends AbstractResponse { return struct; } - public static int sizeOf(int version, LinkedHashMap<TopicPartition, PartitionData> responseData) { - Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.FETCH.id, version)); - writeStruct(struct, version, responseData, 0); - return 4 + struct.sizeOf(); + public static int sizeOf(short version, LinkedHashMap<TopicPartition, PartitionData> responseData) { + return 4 + toStruct(version, responseData, 0).sizeOf(); } }
http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/clients/src/main/java/org/apache/kafka/common/requests/GroupCoordinatorRequest.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/requests/GroupCoordinatorRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/GroupCoordinatorRequest.java index ed56f39..83d6cba 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/GroupCoordinatorRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/GroupCoordinatorRequest.java @@ -32,8 +32,7 @@ public class GroupCoordinatorRequest extends AbstractRequest { } @Override - public GroupCoordinatorRequest build() { - short version = version(); + public GroupCoordinatorRequest build(short version) { return new GroupCoordinatorRequest(this.groupId, version); } @@ -49,14 +48,12 @@ public class GroupCoordinatorRequest extends AbstractRequest { private final String groupId; private GroupCoordinatorRequest(String groupId, short version) { - super(new Struct(ProtoUtils.requestSchema(ApiKeys.GROUP_COORDINATOR.id, version)), - version); - struct.set(GROUP_ID_KEY_NAME, groupId); + super(version); this.groupId = groupId; } public GroupCoordinatorRequest(Struct struct, short versionId) { - super(struct, versionId); + super(versionId); groupId = struct.getString(GROUP_ID_KEY_NAME); } @@ -76,12 +73,15 @@ public class GroupCoordinatorRequest extends AbstractRequest { return groupId; } - public static GroupCoordinatorRequest parse(ByteBuffer buffer, int versionId) { + public static GroupCoordinatorRequest parse(ByteBuffer buffer, short versionId) { return new GroupCoordinatorRequest(ProtoUtils.parseRequest(ApiKeys.GROUP_COORDINATOR.id, versionId, buffer), - (short) versionId); + versionId); } - public static GroupCoordinatorRequest parse(ByteBuffer buffer) { - return parse(buffer, ProtoUtils.latestVersion(ApiKeys.GROUP_COORDINATOR.id)); + @Override + protected Struct toStruct() { + Struct struct = new Struct(ProtoUtils.requestSchema(ApiKeys.GROUP_COORDINATOR.id, version())); + struct.set(GROUP_ID_KEY_NAME, groupId); + return struct; } } http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/clients/src/main/java/org/apache/kafka/common/requests/GroupCoordinatorResponse.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/requests/GroupCoordinatorResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/GroupCoordinatorResponse.java index fc3d358..c13cf3c 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/GroupCoordinatorResponse.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/GroupCoordinatorResponse.java @@ -16,14 +16,12 @@ import org.apache.kafka.common.Node; 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.types.Schema; import org.apache.kafka.common.protocol.types.Struct; import java.nio.ByteBuffer; public class GroupCoordinatorResponse extends AbstractResponse { - private static final Schema CURRENT_SCHEMA = ProtoUtils.currentResponseSchema(ApiKeys.GROUP_COORDINATOR.id); private static final String ERROR_CODE_KEY_NAME = "error_code"; private static final String COORDINATOR_KEY_NAME = "coordinator"; @@ -45,19 +43,11 @@ public class GroupCoordinatorResponse extends AbstractResponse { private final Node node; public GroupCoordinatorResponse(Errors error, Node node) { - super(new Struct(CURRENT_SCHEMA)); - struct.set(ERROR_CODE_KEY_NAME, error.code()); - Struct coordinator = struct.instance(COORDINATOR_KEY_NAME); - coordinator.set(NODE_ID_KEY_NAME, node.id()); - coordinator.set(HOST_KEY_NAME, node.host()); - coordinator.set(PORT_KEY_NAME, node.port()); - struct.set(COORDINATOR_KEY_NAME, coordinator); this.error = error; this.node = node; } public GroupCoordinatorResponse(Struct struct) { - super(struct); error = Errors.forCode(struct.getShort(ERROR_CODE_KEY_NAME)); Struct broker = (Struct) struct.get(COORDINATOR_KEY_NAME); int nodeId = broker.getInt(NODE_ID_KEY_NAME); @@ -74,7 +64,19 @@ public class GroupCoordinatorResponse extends AbstractResponse { return node; } - public static GroupCoordinatorResponse parse(ByteBuffer buffer) { - return new GroupCoordinatorResponse(CURRENT_SCHEMA.read(buffer)); + @Override + protected Struct toStruct(short version) { + Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.GROUP_COORDINATOR.id, version)); + struct.set(ERROR_CODE_KEY_NAME, error.code()); + Struct coordinator = struct.instance(COORDINATOR_KEY_NAME); + coordinator.set(NODE_ID_KEY_NAME, node.id()); + coordinator.set(HOST_KEY_NAME, node.host()); + coordinator.set(PORT_KEY_NAME, node.port()); + struct.set(COORDINATOR_KEY_NAME, coordinator); + return struct; + } + + public static GroupCoordinatorResponse parse(ByteBuffer buffer, short version) { + return new GroupCoordinatorResponse(ProtoUtils.parseResponse(ApiKeys.GROUP_COORDINATOR.id, version, buffer)); } } http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/clients/src/main/java/org/apache/kafka/common/requests/HeartbeatRequest.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/requests/HeartbeatRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/HeartbeatRequest.java index 7e79c8a..4440830 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/HeartbeatRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/HeartbeatRequest.java @@ -37,8 +37,8 @@ public class HeartbeatRequest extends AbstractRequest { } @Override - public HeartbeatRequest build() { - return new HeartbeatRequest(groupId, groupGenerationId, memberId, version()); + public HeartbeatRequest build(short version) { + return new HeartbeatRequest(groupId, groupGenerationId, memberId, version); } @Override @@ -58,18 +58,14 @@ public class HeartbeatRequest extends AbstractRequest { private final String memberId; private HeartbeatRequest(String groupId, int groupGenerationId, String memberId, short version) { - super(new Struct(ProtoUtils.requestSchema(ApiKeys.HEARTBEAT.id, version)), - version); - struct.set(GROUP_ID_KEY_NAME, groupId); - struct.set(GROUP_GENERATION_ID_KEY_NAME, groupGenerationId); - struct.set(MEMBER_ID_KEY_NAME, memberId); + super(version); this.groupId = groupId; this.groupGenerationId = groupGenerationId; this.memberId = memberId; } - public HeartbeatRequest(Struct struct, short versionId) { - super(struct, versionId); + public HeartbeatRequest(Struct struct, short version) { + super(version); groupId = struct.getString(GROUP_ID_KEY_NAME); groupGenerationId = struct.getInt(GROUP_GENERATION_ID_KEY_NAME); memberId = struct.getString(MEMBER_ID_KEY_NAME); @@ -99,11 +95,16 @@ public class HeartbeatRequest extends AbstractRequest { return memberId; } - public static HeartbeatRequest parse(ByteBuffer buffer, int versionId) { - return new HeartbeatRequest(ProtoUtils.parseRequest(ApiKeys.HEARTBEAT.id, versionId, buffer), (short) versionId); + public static HeartbeatRequest parse(ByteBuffer buffer, short versionId) { + return new HeartbeatRequest(ProtoUtils.parseRequest(ApiKeys.HEARTBEAT.id, versionId, buffer), versionId); } - public static HeartbeatRequest parse(ByteBuffer buffer) { - return parse(buffer, ProtoUtils.latestVersion(ApiKeys.HEARTBEAT.id)); + @Override + protected Struct toStruct() { + Struct struct = new Struct(ProtoUtils.requestSchema(ApiKeys.HEARTBEAT.id, version())); + struct.set(GROUP_ID_KEY_NAME, groupId); + struct.set(GROUP_GENERATION_ID_KEY_NAME, groupGenerationId); + struct.set(MEMBER_ID_KEY_NAME, memberId); + return struct; } } http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/clients/src/main/java/org/apache/kafka/common/requests/HeartbeatResponse.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/requests/HeartbeatResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/HeartbeatResponse.java index f36dec4..4cca846 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/HeartbeatResponse.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/HeartbeatResponse.java @@ -15,14 +15,12 @@ package org.apache.kafka.common.requests; 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.types.Schema; import org.apache.kafka.common.protocol.types.Struct; import java.nio.ByteBuffer; public class HeartbeatResponse extends AbstractResponse { - private static final Schema CURRENT_SCHEMA = ProtoUtils.currentResponseSchema(ApiKeys.HEARTBEAT.id); private static final String ERROR_CODE_KEY_NAME = "error_code"; /** @@ -35,17 +33,13 @@ public class HeartbeatResponse extends AbstractResponse { * REBALANCE_IN_PROGRESS (27) * GROUP_AUTHORIZATION_FAILED (30) */ - private final Errors error; public HeartbeatResponse(Errors error) { - super(new Struct(CURRENT_SCHEMA)); - struct.set(ERROR_CODE_KEY_NAME, error.code()); this.error = error; } public HeartbeatResponse(Struct struct) { - super(struct); error = Errors.forCode(struct.getShort(ERROR_CODE_KEY_NAME)); } @@ -53,7 +47,14 @@ public class HeartbeatResponse extends AbstractResponse { return error; } - public static HeartbeatResponse parse(ByteBuffer buffer) { - return new HeartbeatResponse(CURRENT_SCHEMA.read(buffer)); + @Override + protected Struct toStruct(short version) { + Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.HEARTBEAT.id, version)); + struct.set(ERROR_CODE_KEY_NAME, error.code()); + return struct; + } + + public static HeartbeatResponse parse(ByteBuffer buffer, short version) { + return new HeartbeatResponse(ProtoUtils.parseResponse(ApiKeys.HEARTBEAT.id, version, buffer)); } } http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/clients/src/main/java/org/apache/kafka/common/requests/JoinGroupRequest.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/requests/JoinGroupRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/JoinGroupRequest.java index ad0cdd0..37906a6 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/JoinGroupRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/JoinGroupRequest.java @@ -85,8 +85,7 @@ public class JoinGroupRequest extends AbstractRequest { } @Override - public JoinGroupRequest build() { - short version = version(); + public JoinGroupRequest build(short version) { if (version < 1) { rebalanceTimeout = -1; } @@ -112,23 +111,7 @@ public class JoinGroupRequest extends AbstractRequest { private JoinGroupRequest(short version, String groupId, int sessionTimeout, int rebalanceTimeout, String memberId, String protocolType, List<ProtocolMetadata> groupProtocols) { - super(new Struct(ProtoUtils. - requestSchema(ApiKeys.JOIN_GROUP.id, version)), version); - struct.set(GROUP_ID_KEY_NAME, groupId); - struct.set(SESSION_TIMEOUT_KEY_NAME, sessionTimeout); - if (version >= 1) { - struct.set(REBALANCE_TIMEOUT_KEY_NAME, rebalanceTimeout); - } - struct.set(MEMBER_ID_KEY_NAME, memberId); - struct.set(PROTOCOL_TYPE_KEY_NAME, protocolType); - List<Struct> groupProtocolsList = new ArrayList<>(groupProtocols.size()); - for (ProtocolMetadata protocol : groupProtocols) { - Struct protocolStruct = struct.instance(GROUP_PROTOCOLS_KEY_NAME); - protocolStruct.set(PROTOCOL_NAME_KEY_NAME, protocol.name); - protocolStruct.set(PROTOCOL_METADATA_KEY_NAME, protocol.metadata); - groupProtocolsList.add(protocolStruct); - } - struct.set(GROUP_PROTOCOLS_KEY_NAME, groupProtocolsList.toArray()); + super(version); this.groupId = groupId; this.sessionTimeout = sessionTimeout; this.rebalanceTimeout = rebalanceTimeout; @@ -138,7 +121,7 @@ public class JoinGroupRequest extends AbstractRequest { } public JoinGroupRequest(Struct struct, short versionId) { - super(struct, versionId); + super(versionId); groupId = struct.getString(GROUP_ID_KEY_NAME); sessionTimeout = struct.getInt(SESSION_TIMEOUT_KEY_NAME); @@ -169,7 +152,6 @@ public class JoinGroupRequest extends AbstractRequest { case 0: case 1: return new JoinGroupResponse( - versionId, Errors.forException(e), JoinGroupResponse.UNKNOWN_GENERATION_ID, JoinGroupResponse.UNKNOWN_PROTOCOL, @@ -207,12 +189,29 @@ public class JoinGroupRequest extends AbstractRequest { return protocolType; } - public static JoinGroupRequest parse(ByteBuffer buffer, int versionId) { - return new JoinGroupRequest(ProtoUtils.parseRequest(ApiKeys.JOIN_GROUP.id, versionId, buffer), - (short) versionId); + public static JoinGroupRequest parse(ByteBuffer buffer, short version) { + return new JoinGroupRequest(ProtoUtils.parseRequest(ApiKeys.JOIN_GROUP.id, version, buffer), version); } - public static JoinGroupRequest parse(ByteBuffer buffer) { - return parse(buffer, ProtoUtils.latestVersion(ApiKeys.JOIN_GROUP.id)); + @Override + protected Struct toStruct() { + short version = version(); + Struct struct = new Struct(ProtoUtils.requestSchema(ApiKeys.JOIN_GROUP.id, version)); + struct.set(GROUP_ID_KEY_NAME, groupId); + struct.set(SESSION_TIMEOUT_KEY_NAME, sessionTimeout); + if (version >= 1) { + struct.set(REBALANCE_TIMEOUT_KEY_NAME, rebalanceTimeout); + } + struct.set(MEMBER_ID_KEY_NAME, memberId); + struct.set(PROTOCOL_TYPE_KEY_NAME, protocolType); + List<Struct> groupProtocolsList = new ArrayList<>(groupProtocols.size()); + for (ProtocolMetadata protocol : groupProtocols) { + Struct protocolStruct = struct.instance(GROUP_PROTOCOLS_KEY_NAME); + protocolStruct.set(PROTOCOL_NAME_KEY_NAME, protocol.name); + protocolStruct.set(PROTOCOL_METADATA_KEY_NAME, protocol.metadata); + groupProtocolsList.add(protocolStruct); + } + struct.set(GROUP_PROTOCOLS_KEY_NAME, groupProtocolsList.toArray()); + return struct; } } http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/clients/src/main/java/org/apache/kafka/common/requests/JoinGroupResponse.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/requests/JoinGroupResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/JoinGroupResponse.java index bc9366a..d2a323b 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/JoinGroupResponse.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/JoinGroupResponse.java @@ -15,7 +15,6 @@ package org.apache.kafka.common.requests; 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.types.Schema; import org.apache.kafka.common.protocol.types.Struct; import java.nio.ByteBuffer; @@ -26,8 +25,6 @@ import java.util.Map; public class JoinGroupResponse extends AbstractResponse { - private static final short CURRENT_VERSION = ProtoUtils.latestVersion(ApiKeys.JOIN_GROUP.id); - private static final Schema CURRENT_SCHEMA = ProtoUtils.currentResponseSchema(ApiKeys.JOIN_GROUP.id); private static final String ERROR_CODE_KEY_NAME = "error_code"; /** @@ -67,33 +64,6 @@ public class JoinGroupResponse extends AbstractResponse { String memberId, String leaderId, Map<String, ByteBuffer> groupMembers) { - this(CURRENT_VERSION, error, generationId, groupProtocol, memberId, leaderId, groupMembers); - } - - public JoinGroupResponse(int version, - Errors error, - int generationId, - String groupProtocol, - String memberId, - String leaderId, - Map<String, ByteBuffer> groupMembers) { - super(new Struct(ProtoUtils.responseSchema(ApiKeys.JOIN_GROUP.id, version))); - - struct.set(ERROR_CODE_KEY_NAME, error.code()); - struct.set(GENERATION_ID_KEY_NAME, generationId); - struct.set(GROUP_PROTOCOL_KEY_NAME, groupProtocol); - struct.set(MEMBER_ID_KEY_NAME, memberId); - struct.set(LEADER_ID_KEY_NAME, leaderId); - - List<Struct> memberArray = new ArrayList<>(); - for (Map.Entry<String, ByteBuffer> entries: groupMembers.entrySet()) { - Struct memberData = struct.instance(MEMBERS_KEY_NAME); - memberData.set(MEMBER_ID_KEY_NAME, entries.getKey()); - memberData.set(MEMBER_METADATA_KEY_NAME, entries.getValue()); - memberArray.add(memberData); - } - struct.set(MEMBERS_KEY_NAME, memberArray.toArray()); - this.error = error; this.generationId = generationId; this.groupProtocol = groupProtocol; @@ -103,7 +73,6 @@ public class JoinGroupResponse extends AbstractResponse { } public JoinGroupResponse(Struct struct) { - super(struct); members = new HashMap<>(); for (Object memberDataObj : struct.getArray(MEMBERS_KEY_NAME)) { @@ -147,11 +116,29 @@ public class JoinGroupResponse extends AbstractResponse { return members; } - public static JoinGroupResponse parse(ByteBuffer buffer, int version) { - return new JoinGroupResponse(ProtoUtils.responseSchema(ApiKeys.JOIN_GROUP.id, version).read(buffer)); + public static JoinGroupResponse parse(ByteBuffer buffer, short version) { + return new JoinGroupResponse(ProtoUtils.parseResponse(ApiKeys.JOIN_GROUP.id, version, buffer)); } - public static JoinGroupResponse parse(ByteBuffer buffer) { - return new JoinGroupResponse(CURRENT_SCHEMA.read(buffer)); + @Override + protected Struct toStruct(short version) { + Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.JOIN_GROUP.id, version)); + + struct.set(ERROR_CODE_KEY_NAME, error.code()); + struct.set(GENERATION_ID_KEY_NAME, generationId); + struct.set(GROUP_PROTOCOL_KEY_NAME, groupProtocol); + struct.set(MEMBER_ID_KEY_NAME, memberId); + struct.set(LEADER_ID_KEY_NAME, leaderId); + + List<Struct> memberArray = new ArrayList<>(); + for (Map.Entry<String, ByteBuffer> entries : members.entrySet()) { + Struct memberData = struct.instance(MEMBERS_KEY_NAME); + memberData.set(MEMBER_ID_KEY_NAME, entries.getKey()); + memberData.set(MEMBER_METADATA_KEY_NAME, entries.getValue()); + memberArray.add(memberData); + } + struct.set(MEMBERS_KEY_NAME, memberArray.toArray()); + + return struct; } } http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/clients/src/main/java/org/apache/kafka/common/requests/LeaderAndIsrRequest.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/requests/LeaderAndIsrRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/LeaderAndIsrRequest.java index fde184a..c564b43 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/LeaderAndIsrRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/LeaderAndIsrRequest.java @@ -69,9 +69,8 @@ public class LeaderAndIsrRequest extends AbstractRequest { } @Override - public LeaderAndIsrRequest build() { - return new LeaderAndIsrRequest(controllerId, controllerEpoch, partitionStates, - liveLeaders, version()); + public LeaderAndIsrRequest build(short version) { + return new LeaderAndIsrRequest(controllerId, controllerEpoch, partitionStates, liveLeaders, version); } @Override @@ -94,46 +93,15 @@ public class LeaderAndIsrRequest extends AbstractRequest { private LeaderAndIsrRequest(int controllerId, int controllerEpoch, Map<TopicPartition, PartitionState> partitionStates, Set<Node> liveLeaders, short version) { - super(new Struct(ProtoUtils.requestSchema(ApiKeys.LEADER_AND_ISR.id, version)), - version); - struct.set(CONTROLLER_ID_KEY_NAME, controllerId); - struct.set(CONTROLLER_EPOCH_KEY_NAME, controllerEpoch); - - List<Struct> partitionStatesData = new ArrayList<>(partitionStates.size()); - for (Map.Entry<TopicPartition, PartitionState> entry : partitionStates.entrySet()) { - Struct partitionStateData = struct.instance(PARTITION_STATES_KEY_NAME); - TopicPartition topicPartition = entry.getKey(); - partitionStateData.set(TOPIC_KEY_NAME, topicPartition.topic()); - partitionStateData.set(PARTITION_KEY_NAME, topicPartition.partition()); - PartitionState partitionState = entry.getValue(); - partitionStateData.set(CONTROLLER_EPOCH_KEY_NAME, partitionState.controllerEpoch); - partitionStateData.set(LEADER_KEY_NAME, partitionState.leader); - partitionStateData.set(LEADER_EPOCH_KEY_NAME, partitionState.leaderEpoch); - partitionStateData.set(ISR_KEY_NAME, partitionState.isr.toArray()); - partitionStateData.set(ZK_VERSION_KEY_NAME, partitionState.zkVersion); - partitionStateData.set(REPLICAS_KEY_NAME, partitionState.replicas.toArray()); - partitionStatesData.add(partitionStateData); - } - struct.set(PARTITION_STATES_KEY_NAME, partitionStatesData.toArray()); - - List<Struct> leadersData = new ArrayList<>(liveLeaders.size()); - for (Node leader : liveLeaders) { - Struct leaderData = struct.instance(LIVE_LEADERS_KEY_NAME); - leaderData.set(END_POINT_ID_KEY_NAME, leader.id()); - leaderData.set(HOST_KEY_NAME, leader.host()); - leaderData.set(PORT_KEY_NAME, leader.port()); - leadersData.add(leaderData); - } - struct.set(LIVE_LEADERS_KEY_NAME, leadersData.toArray()); - + super(version); this.controllerId = controllerId; this.controllerEpoch = controllerEpoch; this.partitionStates = partitionStates; this.liveLeaders = liveLeaders; } - public LeaderAndIsrRequest(Struct struct, short versionId) { - super(struct, versionId); + public LeaderAndIsrRequest(Struct struct, short version) { + super(version); Map<TopicPartition, PartitionState> partitionStates = new HashMap<>(); for (Object partitionStateDataObj : struct.getArray(PARTITION_STATES_KEY_NAME)) { @@ -177,6 +145,42 @@ public class LeaderAndIsrRequest extends AbstractRequest { } @Override + protected Struct toStruct() { + short version = version(); + Struct struct = new Struct(ProtoUtils.requestSchema(ApiKeys.LEADER_AND_ISR.id, version)); + struct.set(CONTROLLER_ID_KEY_NAME, controllerId); + struct.set(CONTROLLER_EPOCH_KEY_NAME, controllerEpoch); + + List<Struct> partitionStatesData = new ArrayList<>(partitionStates.size()); + for (Map.Entry<TopicPartition, PartitionState> entry : partitionStates.entrySet()) { + Struct partitionStateData = struct.instance(PARTITION_STATES_KEY_NAME); + TopicPartition topicPartition = entry.getKey(); + partitionStateData.set(TOPIC_KEY_NAME, topicPartition.topic()); + partitionStateData.set(PARTITION_KEY_NAME, topicPartition.partition()); + PartitionState partitionState = entry.getValue(); + partitionStateData.set(CONTROLLER_EPOCH_KEY_NAME, partitionState.controllerEpoch); + partitionStateData.set(LEADER_KEY_NAME, partitionState.leader); + partitionStateData.set(LEADER_EPOCH_KEY_NAME, partitionState.leaderEpoch); + partitionStateData.set(ISR_KEY_NAME, partitionState.isr.toArray()); + partitionStateData.set(ZK_VERSION_KEY_NAME, partitionState.zkVersion); + partitionStateData.set(REPLICAS_KEY_NAME, partitionState.replicas.toArray()); + partitionStatesData.add(partitionStateData); + } + struct.set(PARTITION_STATES_KEY_NAME, partitionStatesData.toArray()); + + List<Struct> leadersData = new ArrayList<>(liveLeaders.size()); + for (Node leader : liveLeaders) { + Struct leaderData = struct.instance(LIVE_LEADERS_KEY_NAME); + leaderData.set(END_POINT_ID_KEY_NAME, leader.id()); + leaderData.set(HOST_KEY_NAME, leader.host()); + leaderData.set(PORT_KEY_NAME, leader.port()); + leadersData.add(leaderData); + } + struct.set(LIVE_LEADERS_KEY_NAME, leadersData.toArray()); + return struct; + } + + @Override public AbstractResponse getErrorResponse(Throwable e) { Map<TopicPartition, Errors> responses = new HashMap<>(partitionStates.size()); for (TopicPartition partition : partitionStates.keySet()) { @@ -209,12 +213,8 @@ public class LeaderAndIsrRequest extends AbstractRequest { return liveLeaders; } - public static LeaderAndIsrRequest parse(ByteBuffer buffer, int versionId) { - return new LeaderAndIsrRequest(ProtoUtils.parseRequest(ApiKeys.LEADER_AND_ISR.id, versionId, buffer), - (short) versionId); + public static LeaderAndIsrRequest parse(ByteBuffer buffer, short versionId) { + return new LeaderAndIsrRequest(ProtoUtils.parseRequest(ApiKeys.LEADER_AND_ISR.id, versionId, buffer), versionId); } - public static LeaderAndIsrRequest parse(ByteBuffer buffer) { - return parse(buffer, ProtoUtils.latestVersion(ApiKeys.LEADER_AND_ISR.id)); - } } http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/clients/src/main/java/org/apache/kafka/common/requests/LeaderAndIsrResponse.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/requests/LeaderAndIsrResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/LeaderAndIsrResponse.java index 4d0a05d..2b02daf 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/LeaderAndIsrResponse.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/LeaderAndIsrResponse.java @@ -17,7 +17,6 @@ import org.apache.kafka.common.TopicPartition; 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.types.Schema; import org.apache.kafka.common.protocol.types.Struct; import java.nio.ByteBuffer; @@ -27,7 +26,6 @@ import java.util.List; import java.util.Map; public class LeaderAndIsrResponse extends AbstractResponse { - private static final Schema CURRENT_SCHEMA = ProtoUtils.currentResponseSchema(ApiKeys.LEADER_AND_ISR.id); private static final String ERROR_CODE_KEY_NAME = "error_code"; private static final String PARTITIONS_KEY_NAME = "partitions"; @@ -45,33 +43,12 @@ public class LeaderAndIsrResponse extends AbstractResponse { private final Map<TopicPartition, Errors> responses; - public LeaderAndIsrResponse(Map<TopicPartition, Errors> responses) { - this(Errors.NONE, responses); - } - public LeaderAndIsrResponse(Errors error, Map<TopicPartition, Errors> responses) { - super(new Struct(CURRENT_SCHEMA)); - - List<Struct> responseDatas = new ArrayList<>(responses.size()); - for (Map.Entry<TopicPartition, Errors> response : responses.entrySet()) { - Struct partitionData = struct.instance(PARTITIONS_KEY_NAME); - TopicPartition partition = response.getKey(); - partitionData.set(PARTITIONS_TOPIC_KEY_NAME, partition.topic()); - partitionData.set(PARTITIONS_PARTITION_KEY_NAME, partition.partition()); - partitionData.set(PARTITIONS_ERROR_CODE_KEY_NAME, response.getValue().code()); - responseDatas.add(partitionData); - } - - struct.set(PARTITIONS_KEY_NAME, responseDatas.toArray()); - struct.set(ERROR_CODE_KEY_NAME, error.code()); - this.responses = responses; this.error = error; } public LeaderAndIsrResponse(Struct struct) { - super(struct); - responses = new HashMap<>(); for (Object responseDataObj : struct.getArray(PARTITIONS_KEY_NAME)) { Struct responseData = (Struct) responseDataObj; @@ -92,12 +69,27 @@ public class LeaderAndIsrResponse extends AbstractResponse { return error; } - public static LeaderAndIsrResponse parse(ByteBuffer buffer, int version) { + public static LeaderAndIsrResponse parse(ByteBuffer buffer, short version) { return new LeaderAndIsrResponse(ProtoUtils.parseResponse(ApiKeys.LEADER_AND_ISR.id, version, buffer)); } - public static LeaderAndIsrResponse parse(ByteBuffer buffer) { - return new LeaderAndIsrResponse(CURRENT_SCHEMA.read(buffer)); - } + @Override + protected Struct toStruct(short version) { + Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.LEADER_AND_ISR.id, version)); + + List<Struct> responseDatas = new ArrayList<>(responses.size()); + for (Map.Entry<TopicPartition, Errors> response : responses.entrySet()) { + Struct partitionData = struct.instance(PARTITIONS_KEY_NAME); + TopicPartition partition = response.getKey(); + partitionData.set(PARTITIONS_TOPIC_KEY_NAME, partition.topic()); + partitionData.set(PARTITIONS_PARTITION_KEY_NAME, partition.partition()); + partitionData.set(PARTITIONS_ERROR_CODE_KEY_NAME, response.getValue().code()); + responseDatas.add(partitionData); + } + struct.set(PARTITIONS_KEY_NAME, responseDatas.toArray()); + struct.set(ERROR_CODE_KEY_NAME, error.code()); + + return struct; + } } http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/clients/src/main/java/org/apache/kafka/common/requests/LeaveGroupRequest.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/requests/LeaveGroupRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/LeaveGroupRequest.java index 2a7b70e..16622e4 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/LeaveGroupRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/LeaveGroupRequest.java @@ -33,8 +33,8 @@ public class LeaveGroupRequest extends AbstractRequest { } @Override - public LeaveGroupRequest build() { - return new LeaveGroupRequest(groupId, memberId, version()); + public LeaveGroupRequest build(short version) { + return new LeaveGroupRequest(groupId, memberId, version); } @Override @@ -52,16 +52,13 @@ public class LeaveGroupRequest extends AbstractRequest { private final String memberId; private LeaveGroupRequest(String groupId, String memberId, short version) { - super(new Struct(ProtoUtils.requestSchema(ApiKeys.LEAVE_GROUP.id, version)), - version); - struct.set(GROUP_ID_KEY_NAME, groupId); - struct.set(MEMBER_ID_KEY_NAME, memberId); + super(version); this.groupId = groupId; this.memberId = memberId; } public LeaveGroupRequest(Struct struct, short version) { - super(struct, version); + super(version); groupId = struct.getString(GROUP_ID_KEY_NAME); memberId = struct.getString(MEMBER_ID_KEY_NAME); } @@ -86,12 +83,15 @@ public class LeaveGroupRequest extends AbstractRequest { return memberId; } - public static LeaveGroupRequest parse(ByteBuffer buffer, int versionId) { - return new LeaveGroupRequest(ProtoUtils.parseRequest(ApiKeys.LEAVE_GROUP.id, versionId, buffer), - (short) versionId); + public static LeaveGroupRequest parse(ByteBuffer buffer, short versionId) { + return new LeaveGroupRequest(ProtoUtils.parseRequest(ApiKeys.LEAVE_GROUP.id, versionId, buffer), versionId); } - public static LeaveGroupRequest parse(ByteBuffer buffer) { - return parse(buffer, ProtoUtils.latestVersion(ApiKeys.LEAVE_GROUP.id)); + @Override + protected Struct toStruct() { + Struct struct = new Struct(ProtoUtils.requestSchema(ApiKeys.LEAVE_GROUP.id, version())); + struct.set(GROUP_ID_KEY_NAME, groupId); + struct.set(MEMBER_ID_KEY_NAME, memberId); + return struct; } } http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/clients/src/main/java/org/apache/kafka/common/requests/LeaveGroupResponse.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/requests/LeaveGroupResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/LeaveGroupResponse.java index bd1c84d..a28816a 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/LeaveGroupResponse.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/LeaveGroupResponse.java @@ -12,16 +12,15 @@ */ package org.apache.kafka.common.requests; -import java.nio.ByteBuffer; 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.types.Schema; import org.apache.kafka.common.protocol.types.Struct; +import java.nio.ByteBuffer; + public class LeaveGroupResponse extends AbstractResponse { - private static final Schema CURRENT_SCHEMA = ProtoUtils.currentResponseSchema(ApiKeys.LEAVE_GROUP.id); private static final String ERROR_CODE_KEY_NAME = "error_code"; /** @@ -33,16 +32,13 @@ public class LeaveGroupResponse extends AbstractResponse { * UNKNOWN_CONSUMER_ID (25) * GROUP_AUTHORIZATION_FAILED (30) */ - private final Errors error; + public LeaveGroupResponse(Errors error) { - super(new Struct(CURRENT_SCHEMA)); - struct.set(ERROR_CODE_KEY_NAME, error.code()); this.error = error; } public LeaveGroupResponse(Struct struct) { - super(struct); error = Errors.forCode(struct.getShort(ERROR_CODE_KEY_NAME)); } @@ -50,7 +46,15 @@ public class LeaveGroupResponse extends AbstractResponse { return error; } - public static LeaveGroupResponse parse(ByteBuffer buffer) { - return new LeaveGroupResponse(CURRENT_SCHEMA.read(buffer)); + @Override + public Struct toStruct(short version) { + Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.LEAVE_GROUP.id, version)); + struct.set(ERROR_CODE_KEY_NAME, error.code()); + return struct; } + + public static LeaveGroupResponse parse(ByteBuffer buffer, short versionId) { + return new LeaveGroupResponse(ProtoUtils.parseResponse(ApiKeys.LEAVE_GROUP.id, versionId, buffer)); + } + } http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/clients/src/main/java/org/apache/kafka/common/requests/ListGroupsRequest.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/requests/ListGroupsRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/ListGroupsRequest.java index 235f4e4..badb527 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/ListGroupsRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/ListGroupsRequest.java @@ -27,8 +27,8 @@ public class ListGroupsRequest extends AbstractRequest { } @Override - public ListGroupsRequest build() { - return new ListGroupsRequest(version()); + public ListGroupsRequest build(short version) { + return new ListGroupsRequest(version); } @Override @@ -38,12 +38,11 @@ public class ListGroupsRequest extends AbstractRequest { } public ListGroupsRequest(short version) { - super(new Struct(ProtoUtils.requestSchema(ApiKeys.LIST_GROUPS.id, version)), - version); + super(version); } public ListGroupsRequest(Struct struct, short versionId) { - super(struct, versionId); + super(versionId); } @Override @@ -58,12 +57,12 @@ public class ListGroupsRequest extends AbstractRequest { } } - public static ListGroupsRequest parse(ByteBuffer buffer, int versionId) { - return new ListGroupsRequest(ProtoUtils.parseRequest(ApiKeys.LIST_GROUPS.id, versionId, buffer), - (short) versionId); + public static ListGroupsRequest parse(ByteBuffer buffer, short versionId) { + return new ListGroupsRequest(ProtoUtils.parseRequest(ApiKeys.LIST_GROUPS.id, versionId, buffer), versionId); } - public static ListGroupsRequest parse(ByteBuffer buffer) { - return parse(buffer, ProtoUtils.latestVersion(ApiKeys.LIST_GROUPS.id)); + @Override + protected Struct toStruct() { + return new Struct(ProtoUtils.requestSchema(ApiKeys.LIST_GROUPS.id, version())); } } http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/clients/src/main/java/org/apache/kafka/common/requests/ListGroupsResponse.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/requests/ListGroupsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/ListGroupsResponse.java index f421064..e05a4b1 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/ListGroupsResponse.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/ListGroupsResponse.java @@ -15,7 +15,6 @@ package org.apache.kafka.common.requests; 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.types.Schema; import org.apache.kafka.common.protocol.types.Struct; import java.nio.ByteBuffer; @@ -25,8 +24,6 @@ import java.util.List; public class ListGroupsResponse extends AbstractResponse { - private static final Schema CURRENT_SCHEMA = ProtoUtils.currentResponseSchema(ApiKeys.LIST_GROUPS.id); - public static final String ERROR_CODE_KEY_NAME = "error_code"; public static final String GROUPS_KEY_NAME = "groups"; public static final String GROUP_ID_KEY_NAME = "group_id"; @@ -43,22 +40,11 @@ public class ListGroupsResponse extends AbstractResponse { private final List<Group> groups; public ListGroupsResponse(Errors error, List<Group> groups) { - super(new Struct(CURRENT_SCHEMA)); - struct.set(ERROR_CODE_KEY_NAME, error.code()); - List<Struct> groupList = new ArrayList<>(); - for (Group group : groups) { - Struct groupStruct = struct.instance(GROUPS_KEY_NAME); - groupStruct.set(GROUP_ID_KEY_NAME, group.groupId); - groupStruct.set(PROTOCOL_TYPE_KEY_NAME, group.protocolType); - groupList.add(groupStruct); - } - struct.set(GROUPS_KEY_NAME, groupList.toArray()); this.error = error; this.groups = groups; } public ListGroupsResponse(Struct struct) { - super(struct); this.error = Errors.forCode(struct.getShort(ERROR_CODE_KEY_NAME)); this.groups = new ArrayList<>(); for (Object groupObj : struct.getArray(GROUPS_KEY_NAME)) { @@ -96,12 +82,27 @@ public class ListGroupsResponse extends AbstractResponse { } - public static ListGroupsResponse parse(ByteBuffer buffer) { - return new ListGroupsResponse(CURRENT_SCHEMA.read(buffer)); + @Override + protected Struct toStruct(short version) { + Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.LIST_GROUPS.id, version)); + struct.set(ERROR_CODE_KEY_NAME, error.code()); + List<Struct> groupList = new ArrayList<>(); + for (Group group : groups) { + Struct groupStruct = struct.instance(GROUPS_KEY_NAME); + groupStruct.set(GROUP_ID_KEY_NAME, group.groupId); + groupStruct.set(PROTOCOL_TYPE_KEY_NAME, group.protocolType); + groupList.add(groupStruct); + } + struct.set(GROUPS_KEY_NAME, groupList.toArray()); + return struct; } public static ListGroupsResponse fromError(Errors error) { return new ListGroupsResponse(error, Collections.<Group>emptyList()); } + public static ListGroupsResponse parse(ByteBuffer buffer, short versionId) { + return new ListGroupsResponse(ProtoUtils.parseResponse(ApiKeys.LIST_GROUPS.id, versionId, buffer)); + } + } http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/clients/src/main/java/org/apache/kafka/common/requests/ListOffsetRequest.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/requests/ListOffsetRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/ListOffsetRequest.java index 6214a56..3e2ad7c 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/ListOffsetRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/ListOffsetRequest.java @@ -60,16 +60,21 @@ public class ListOffsetRequest extends AbstractRequest { public static class Builder extends AbstractRequest.Builder<ListOffsetRequest> { private final int replicaId; + private final short minVersion; private Map<TopicPartition, PartitionData> offsetData = null; private Map<TopicPartition, Long> partitionTimestamps = null; - private short minVersion = (short) 0; - public Builder() { - this(CONSUMER_REPLICA_ID); + public static Builder forReplica(short desiredVersion, int replicaId) { + return new Builder((short) 0, desiredVersion, replicaId); } - public Builder(int replicaId) { - super(ApiKeys.LIST_OFFSETS); + public static Builder forConsumer(short minVersion) { + return new Builder(minVersion, null, CONSUMER_REPLICA_ID); + } + + private Builder(short minVersion, Short desiredVersion, int replicaId) { + super(ApiKeys.LIST_OFFSETS, desiredVersion); + this.minVersion = minVersion; this.replicaId = replicaId; } @@ -84,8 +89,7 @@ public class ListOffsetRequest extends AbstractRequest { } @Override - public ListOffsetRequest build() { - short version = version(); + public ListOffsetRequest build(short version) { if (version < minVersion) { throw new UnsupportedVersionException("Cannot create a v" + version + " ListOffsetRequest because " + "we require features supported only in " + minVersion + " or later."); @@ -117,14 +121,6 @@ public class ListOffsetRequest extends AbstractRequest { return new ListOffsetRequest(replicaId, m, version); } - /** - * Set the minimum version which we will produce for this request. - */ - public Builder setMinVersion(short minVersion) { - this.minVersion = minVersion; - return this; - } - @Override public String toString() { StringBuilder bld = new StringBuilder(); @@ -170,44 +166,15 @@ public class ListOffsetRequest extends AbstractRequest { */ @SuppressWarnings("unchecked") private ListOffsetRequest(int replicaId, Map<TopicPartition, ?> targetTimes, short version) { - super(new Struct(ProtoUtils.requestSchema(ApiKeys.LIST_OFFSETS.id, version)), version); - Map<String, Map<Integer, Object>> topicsData = - CollectionUtils.groupDataByTopic((Map<TopicPartition, Object>) targetTimes); - - struct.set(REPLICA_ID_KEY_NAME, replicaId); - List<Struct> topicArray = new ArrayList<Struct>(); - for (Map.Entry<String, Map<Integer, Object>> topicEntry: topicsData.entrySet()) { - Struct topicData = struct.instance(TOPICS_KEY_NAME); - topicData.set(TOPIC_KEY_NAME, topicEntry.getKey()); - List<Struct> partitionArray = new ArrayList<>(); - for (Map.Entry<Integer, Object> partitionEntry : topicEntry.getValue().entrySet()) { - if (version == 0) { - PartitionData offsetPartitionData = (PartitionData) partitionEntry.getValue(); - Struct partitionData = topicData.instance(PARTITIONS_KEY_NAME); - partitionData.set(PARTITION_KEY_NAME, partitionEntry.getKey()); - partitionData.set(TIMESTAMP_KEY_NAME, offsetPartitionData.timestamp); - partitionData.set(MAX_NUM_OFFSETS_KEY_NAME, offsetPartitionData.maxNumOffsets); - partitionArray.add(partitionData); - } else { - Long timestamp = (Long) partitionEntry.getValue(); - Struct partitionData = topicData.instance(PARTITIONS_KEY_NAME); - partitionData.set(PARTITION_KEY_NAME, partitionEntry.getKey()); - partitionData.set(TIMESTAMP_KEY_NAME, timestamp); - partitionArray.add(partitionData); - } - } - topicData.set(PARTITIONS_KEY_NAME, partitionArray.toArray()); - topicArray.add(topicData); - } - struct.set(TOPICS_KEY_NAME, topicArray.toArray()); + super(version); this.replicaId = replicaId; this.offsetData = version == 0 ? (Map<TopicPartition, PartitionData>) targetTimes : null; this.partitionTimestamps = version == 1 ? (Map<TopicPartition, Long>) targetTimes : null; this.duplicatePartitions = Collections.emptySet(); } - public ListOffsetRequest(Struct struct, short versionId) { - super(struct, versionId); + public ListOffsetRequest(Struct struct, short version) { + super(version); Set<TopicPartition> duplicatePartitions = new HashSet<>(); replicaId = struct.getInt(REPLICA_ID_KEY_NAME); offsetData = new HashMap<>(); @@ -236,17 +203,19 @@ public class ListOffsetRequest extends AbstractRequest { @Override @SuppressWarnings("deprecation") public AbstractResponse getErrorResponse(Throwable e) { - Map<TopicPartition, ListOffsetResponse.PartitionData> responseData = new HashMap<TopicPartition, ListOffsetResponse.PartitionData>(); + Map<TopicPartition, ListOffsetResponse.PartitionData> responseData = new HashMap<>(); short versionId = version(); if (versionId == 0) { for (Map.Entry<TopicPartition, PartitionData> entry : offsetData.entrySet()) { - ListOffsetResponse.PartitionData partitionResponse = new ListOffsetResponse.PartitionData(Errors.forException(e), new ArrayList<Long>()); + ListOffsetResponse.PartitionData partitionResponse = new ListOffsetResponse.PartitionData( + Errors.forException(e), Collections.<Long>emptyList()); responseData.put(entry.getKey(), partitionResponse); } } else { for (Map.Entry<TopicPartition, Long> entry : partitionTimestamps.entrySet()) { - ListOffsetResponse.PartitionData partitionResponse = new ListOffsetResponse.PartitionData(Errors.forException(e), -1L, -1L); + ListOffsetResponse.PartitionData partitionResponse = new ListOffsetResponse.PartitionData( + Errors.forException(e), -1L, -1L); responseData.put(entry.getKey(), partitionResponse); } } @@ -254,7 +223,7 @@ public class ListOffsetRequest extends AbstractRequest { switch (versionId) { case 0: case 1: - return new ListOffsetResponse(responseData, versionId); + return new ListOffsetResponse(responseData); default: throw new IllegalArgumentException(String.format("Version %d is not valid. Valid versions for %s are 0 to %d", versionId, this.getClass().getSimpleName(), ProtoUtils.latestVersion(ApiKeys.LIST_OFFSETS.id))); @@ -278,12 +247,44 @@ public class ListOffsetRequest extends AbstractRequest { return duplicatePartitions; } - public static ListOffsetRequest parse(ByteBuffer buffer, int versionId) { - return new ListOffsetRequest(ProtoUtils.parseRequest(ApiKeys.LIST_OFFSETS.id, versionId, buffer), - (short) versionId); + public static ListOffsetRequest parse(ByteBuffer buffer, short versionId) { + return new ListOffsetRequest(ProtoUtils.parseRequest(ApiKeys.LIST_OFFSETS.id, versionId, buffer), versionId); } - public static ListOffsetRequest parse(ByteBuffer buffer) { - return parse(buffer, ProtoUtils.latestVersion(ApiKeys.LIST_OFFSETS.id)); + @Override + protected Struct toStruct() { + short version = version(); + Struct struct = new Struct(ProtoUtils.requestSchema(ApiKeys.LIST_OFFSETS.id, version)); + + Map<TopicPartition, ?> targetTimes = partitionTimestamps == null ? offsetData : partitionTimestamps; + Map<String, Map<Integer, Object>> topicsData = CollectionUtils.groupDataByTopic(targetTimes); + + struct.set(REPLICA_ID_KEY_NAME, replicaId); + List<Struct> topicArray = new ArrayList<>(); + for (Map.Entry<String, Map<Integer, Object>> topicEntry: topicsData.entrySet()) { + Struct topicData = struct.instance(TOPICS_KEY_NAME); + topicData.set(TOPIC_KEY_NAME, topicEntry.getKey()); + List<Struct> partitionArray = new ArrayList<>(); + for (Map.Entry<Integer, Object> partitionEntry : topicEntry.getValue().entrySet()) { + if (version == 0) { + PartitionData offsetPartitionData = (PartitionData) partitionEntry.getValue(); + Struct partitionData = topicData.instance(PARTITIONS_KEY_NAME); + partitionData.set(PARTITION_KEY_NAME, partitionEntry.getKey()); + partitionData.set(TIMESTAMP_KEY_NAME, offsetPartitionData.timestamp); + partitionData.set(MAX_NUM_OFFSETS_KEY_NAME, offsetPartitionData.maxNumOffsets); + partitionArray.add(partitionData); + } else { + Long timestamp = (Long) partitionEntry.getValue(); + Struct partitionData = topicData.instance(PARTITIONS_KEY_NAME); + partitionData.set(PARTITION_KEY_NAME, partitionEntry.getKey()); + partitionData.set(TIMESTAMP_KEY_NAME, timestamp); + partitionArray.add(partitionData); + } + } + topicData.set(PARTITIONS_KEY_NAME, partitionArray.toArray()); + topicArray.add(topicData); + } + struct.set(TOPICS_KEY_NAME, topicArray.toArray()); + return struct; } } http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/clients/src/main/java/org/apache/kafka/common/requests/ListOffsetResponse.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/requests/ListOffsetResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/ListOffsetResponse.java index b815a53..cb3bafc 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/ListOffsetResponse.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/ListOffsetResponse.java @@ -20,7 +20,6 @@ import org.apache.kafka.common.TopicPartition; 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.types.Schema; import org.apache.kafka.common.protocol.types.Struct; import org.apache.kafka.common.utils.CollectionUtils; import org.apache.kafka.common.utils.Utils; @@ -35,7 +34,6 @@ public class ListOffsetResponse extends AbstractResponse { public static final long UNKNOWN_TIMESTAMP = -1L; public static final long UNKNOWN_OFFSET = -1L; - private static final Schema CURRENT_SCHEMA = ProtoUtils.currentResponseSchema(ApiKeys.LIST_OFFSETS.id); private static final String RESPONSES_KEY_NAME = "responses"; // topic level field names @@ -61,8 +59,6 @@ public class ListOffsetResponse extends AbstractResponse { private static final String TIMESTAMP_KEY_NAME = "timestamp"; private static final String OFFSET_KEY_NAME = "offset"; - private final Map<TopicPartition, PartitionData> responseData; - public static final class PartitionData { public final Errors error; // The offsets list is only used in ListOffsetResponse v0. @@ -110,46 +106,17 @@ public class ListOffsetResponse extends AbstractResponse { } } + private final Map<TopicPartition, PartitionData> responseData; + /** - * Constructor for ListOffsetResponse v0. + * Constructor for all versions. */ - @Deprecated public ListOffsetResponse(Map<TopicPartition, PartitionData> responseData) { - this(responseData, 0); - } - - public ListOffsetResponse(Map<TopicPartition, PartitionData> responseData, int version) { - super(new Struct(ProtoUtils.responseSchema(ApiKeys.LIST_OFFSETS.id, version))); - Map<String, Map<Integer, PartitionData>> topicsData = CollectionUtils.groupDataByTopic(responseData); - - List<Struct> topicArray = new ArrayList<Struct>(); - for (Map.Entry<String, Map<Integer, PartitionData>> topicEntry: topicsData.entrySet()) { - Struct topicData = struct.instance(RESPONSES_KEY_NAME); - topicData.set(TOPIC_KEY_NAME, topicEntry.getKey()); - List<Struct> partitionArray = new ArrayList<Struct>(); - for (Map.Entry<Integer, PartitionData> partitionEntry : topicEntry.getValue().entrySet()) { - PartitionData offsetPartitionData = partitionEntry.getValue(); - Struct partitionData = topicData.instance(PARTITIONS_KEY_NAME); - partitionData.set(PARTITION_KEY_NAME, partitionEntry.getKey()); - partitionData.set(ERROR_CODE_KEY_NAME, offsetPartitionData.error.code()); - if (version == 0) - partitionData.set(OFFSETS_KEY_NAME, offsetPartitionData.offsets.toArray()); - else { - partitionData.set(TIMESTAMP_KEY_NAME, offsetPartitionData.timestamp); - partitionData.set(OFFSET_KEY_NAME, offsetPartitionData.offset); - } - partitionArray.add(partitionData); - } - topicData.set(PARTITIONS_KEY_NAME, partitionArray.toArray()); - topicArray.add(topicData); - } - struct.set(RESPONSES_KEY_NAME, topicArray.toArray()); this.responseData = responseData; } public ListOffsetResponse(Struct struct) { - super(struct); - responseData = new HashMap<TopicPartition, PartitionData>(); + responseData = new HashMap<>(); for (Object topicResponseObj : struct.getArray(RESPONSES_KEY_NAME)) { Struct topicResponse = (Struct) topicResponseObj; String topic = topicResponse.getString(TOPIC_KEY_NAME); @@ -178,11 +145,38 @@ public class ListOffsetResponse extends AbstractResponse { return responseData; } - public static ListOffsetResponse parse(ByteBuffer buffer) { - return new ListOffsetResponse(CURRENT_SCHEMA.read(buffer)); + public static ListOffsetResponse parse(ByteBuffer buffer, short version) { + return new ListOffsetResponse(ProtoUtils.parseResponse(ApiKeys.LIST_OFFSETS.id, version, buffer)); } - public static ListOffsetResponse parse(ByteBuffer buffer, int version) { - return new ListOffsetResponse(ProtoUtils.responseSchema(ApiKeys.LIST_OFFSETS.id, version).read(buffer)); + @Override + protected Struct toStruct(short version) { + Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.LIST_OFFSETS.id, version)); + Map<String, Map<Integer, PartitionData>> topicsData = CollectionUtils.groupDataByTopic(responseData); + + List<Struct> topicArray = new ArrayList<>(); + for (Map.Entry<String, Map<Integer, PartitionData>> topicEntry: topicsData.entrySet()) { + Struct topicData = struct.instance(RESPONSES_KEY_NAME); + topicData.set(TOPIC_KEY_NAME, topicEntry.getKey()); + List<Struct> partitionArray = new ArrayList<>(); + for (Map.Entry<Integer, PartitionData> partitionEntry : topicEntry.getValue().entrySet()) { + PartitionData offsetPartitionData = partitionEntry.getValue(); + Struct partitionData = topicData.instance(PARTITIONS_KEY_NAME); + partitionData.set(PARTITION_KEY_NAME, partitionEntry.getKey()); + partitionData.set(ERROR_CODE_KEY_NAME, offsetPartitionData.error.code()); + if (version == 0) + partitionData.set(OFFSETS_KEY_NAME, offsetPartitionData.offsets.toArray()); + else { + partitionData.set(TIMESTAMP_KEY_NAME, offsetPartitionData.timestamp); + partitionData.set(OFFSET_KEY_NAME, offsetPartitionData.offset); + } + partitionArray.add(partitionData); + } + topicData.set(PARTITIONS_KEY_NAME, partitionArray.toArray()); + topicArray.add(topicData); + } + struct.set(RESPONSES_KEY_NAME, topicArray.toArray()); + + return struct; } } http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/clients/src/main/java/org/apache/kafka/common/requests/MetadataRequest.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/requests/MetadataRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/MetadataRequest.java index 16af1b7..f31315f 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/MetadataRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/MetadataRequest.java @@ -51,8 +51,7 @@ public class MetadataRequest extends AbstractRequest { } @Override - public MetadataRequest build() { - short version = version(); + public MetadataRequest build(short version) { if (version < 1) { throw new UnsupportedVersionException("MetadataRequest " + "versions older than 1 are not supported."); @@ -79,27 +78,18 @@ public class MetadataRequest extends AbstractRequest { private final List<String> topics; - public static MetadataRequest allTopics(short version) { - return new MetadataRequest.Builder(null).setVersion(version).build(); - } - /** * In v0 null is not allowed and and empty list indicates requesting all topics. * Note: modern clients do not support sending v0 requests. * In v1 null indicates requesting all topics, and an empty list indicates requesting no topics. */ public MetadataRequest(List<String> topics, short version) { - super(new Struct(ProtoUtils.requestSchema(ApiKeys.METADATA.id, version)), - version); - if (topics == null) - struct.set(TOPICS_KEY_NAME, null); - else - struct.set(TOPICS_KEY_NAME, topics.toArray()); + super(version); this.topics = topics; } public MetadataRequest(Struct struct, short version) { - super(struct, version); + super(version); Object[] topicArray = struct.getArray(TOPICS_KEY_NAME); if (topicArray != null) { topics = new ArrayList<>(); @@ -127,7 +117,7 @@ public class MetadataRequest extends AbstractRequest { case 0: case 1: case 2: - return new MetadataResponse(Collections.<Node>emptyList(), null, MetadataResponse.NO_CONTROLLER_ID, topicMetadatas, versionId); + return new MetadataResponse(Collections.<Node>emptyList(), null, MetadataResponse.NO_CONTROLLER_ID, topicMetadatas); default: throw new IllegalArgumentException(String.format("Version %d is not valid. Valid versions for %s are 0 to %d", versionId, this.getClass().getSimpleName(), ProtoUtils.latestVersion(ApiKeys.METADATA.id))); @@ -142,12 +132,17 @@ public class MetadataRequest extends AbstractRequest { return topics; } - public static MetadataRequest parse(ByteBuffer buffer, int versionId) { - return new MetadataRequest(ProtoUtils.parseRequest(ApiKeys.METADATA.id, versionId, buffer), - (short) versionId); + public static MetadataRequest parse(ByteBuffer buffer, short versionId) { + return new MetadataRequest(ProtoUtils.parseRequest(ApiKeys.METADATA.id, versionId, buffer), versionId); } - public static MetadataRequest parse(ByteBuffer buffer) { - return parse(buffer, ProtoUtils.latestVersion(ApiKeys.METADATA.id)); + @Override + protected Struct toStruct() { + Struct struct = new Struct(ProtoUtils.requestSchema(ApiKeys.METADATA.id, version())); + if (topics == null) + struct.set(TOPICS_KEY_NAME, null); + else + struct.set(TOPICS_KEY_NAME, topics.toArray()); + return struct; } } http://git-wip-us.apache.org/repos/asf/kafka/blob/fc1cfe47/clients/src/main/java/org/apache/kafka/common/requests/MetadataResponse.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/requests/MetadataResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/MetadataResponse.java index a8baee5..268bf84 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/MetadataResponse.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/MetadataResponse.java @@ -31,7 +31,6 @@ import java.util.Set; public class MetadataResponse extends AbstractResponse { - private static final short CURRENT_VERSION = ProtoUtils.latestVersion(ApiKeys.METADATA.id); private static final String BROKERS_KEY_NAME = "brokers"; private static final String TOPIC_METADATA_KEY_NAME = "topic_metadata"; @@ -83,78 +82,16 @@ public class MetadataResponse extends AbstractResponse { private final String clusterId; /** - * Constructor for the latest version + * Constructor for all versions. */ public MetadataResponse(List<Node> brokers, String clusterId, int controllerId, List<TopicMetadata> topicMetadata) { - this(brokers, clusterId, controllerId, topicMetadata, CURRENT_VERSION); - } - - /** - * Constructor for a specific version - */ - public MetadataResponse(List<Node> brokers, String clusterId, int controllerId, List<TopicMetadata> topicMetadata, int version) { - super(new Struct(ProtoUtils.responseSchema(ApiKeys.METADATA.id, version))); this.brokers = brokers; this.controller = getControllerNode(controllerId, brokers); this.topicMetadata = topicMetadata; this.clusterId = clusterId; - - List<Struct> brokerArray = new ArrayList<>(); - for (Node node : brokers) { - Struct broker = struct.instance(BROKERS_KEY_NAME); - broker.set(NODE_ID_KEY_NAME, node.id()); - broker.set(HOST_KEY_NAME, node.host()); - broker.set(PORT_KEY_NAME, node.port()); - // This field only exists in v1+ - if (broker.hasField(RACK_KEY_NAME)) - broker.set(RACK_KEY_NAME, node.rack()); - brokerArray.add(broker); - } - struct.set(BROKERS_KEY_NAME, brokerArray.toArray()); - - // This field only exists in v1+ - if (struct.hasField(CONTROLLER_ID_KEY_NAME)) - struct.set(CONTROLLER_ID_KEY_NAME, controllerId); - - // This field only exists in v2+ - if (struct.hasField(CLUSTER_ID_KEY_NAME)) - struct.set(CLUSTER_ID_KEY_NAME, clusterId); - - List<Struct> topicMetadataArray = new ArrayList<>(topicMetadata.size()); - for (TopicMetadata metadata : topicMetadata) { - Struct topicData = struct.instance(TOPIC_METADATA_KEY_NAME); - topicData.set(TOPIC_KEY_NAME, metadata.topic); - topicData.set(TOPIC_ERROR_CODE_KEY_NAME, metadata.error.code()); - // This field only exists in v1+ - if (topicData.hasField(IS_INTERNAL_KEY_NAME)) - topicData.set(IS_INTERNAL_KEY_NAME, metadata.isInternal()); - - List<Struct> partitionMetadataArray = new ArrayList<>(metadata.partitionMetadata.size()); - for (PartitionMetadata partitionMetadata : metadata.partitionMetadata()) { - Struct partitionData = topicData.instance(PARTITION_METADATA_KEY_NAME); - partitionData.set(PARTITION_ERROR_CODE_KEY_NAME, partitionMetadata.error.code()); - partitionData.set(PARTITION_KEY_NAME, partitionMetadata.partition); - partitionData.set(LEADER_KEY_NAME, partitionMetadata.leader.id()); - ArrayList<Integer> replicas = new ArrayList<>(partitionMetadata.replicas.size()); - for (Node node : partitionMetadata.replicas) - replicas.add(node.id()); - partitionData.set(REPLICAS_KEY_NAME, replicas.toArray()); - ArrayList<Integer> isr = new ArrayList<>(partitionMetadata.isr.size()); - for (Node node : partitionMetadata.isr) - isr.add(node.id()); - partitionData.set(ISR_KEY_NAME, isr.toArray()); - partitionMetadataArray.add(partitionData); - - } - topicData.set(PARTITION_METADATA_KEY_NAME, partitionMetadataArray.toArray()); - topicMetadataArray.add(topicData); - } - struct.set(TOPIC_METADATA_KEY_NAME, topicMetadataArray.toArray()); } public MetadataResponse(Struct struct) { - super(struct); - Map<Integer, Node> brokers = new HashMap<>(); Object[] brokerStructs = (Object[]) struct.get(BROKERS_KEY_NAME); for (int i = 0; i < brokerStructs.length; i++) { @@ -317,12 +254,8 @@ public class MetadataResponse extends AbstractResponse { return this.clusterId; } - public static MetadataResponse parse(ByteBuffer buffer) { - return parse(buffer, CURRENT_VERSION); - } - - public static MetadataResponse parse(ByteBuffer buffer, int version) { - return new MetadataResponse(ProtoUtils.responseSchema(ApiKeys.METADATA.id, version).read(buffer)); + public static MetadataResponse parse(ByteBuffer buffer, short version) { + return new MetadataResponse(ProtoUtils.parseResponse(ApiKeys.METADATA.id, version, buffer)); } public static class TopicMetadata { @@ -400,4 +333,60 @@ public class MetadataResponse extends AbstractResponse { } + @Override + protected Struct toStruct(short version) { + Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.METADATA.id, version)); + List<Struct> brokerArray = new ArrayList<>(); + for (Node node : brokers) { + Struct broker = struct.instance(BROKERS_KEY_NAME); + broker.set(NODE_ID_KEY_NAME, node.id()); + broker.set(HOST_KEY_NAME, node.host()); + broker.set(PORT_KEY_NAME, node.port()); + // This field only exists in v1+ + if (broker.hasField(RACK_KEY_NAME)) + broker.set(RACK_KEY_NAME, node.rack()); + brokerArray.add(broker); + } + struct.set(BROKERS_KEY_NAME, brokerArray.toArray()); + + // This field only exists in v1+ + if (struct.hasField(CONTROLLER_ID_KEY_NAME)) + struct.set(CONTROLLER_ID_KEY_NAME, controller == null ? NO_CONTROLLER_ID : controller.id()); + + // This field only exists in v2+ + if (struct.hasField(CLUSTER_ID_KEY_NAME)) + struct.set(CLUSTER_ID_KEY_NAME, clusterId); + + List<Struct> topicMetadataArray = new ArrayList<>(topicMetadata.size()); + for (TopicMetadata metadata : topicMetadata) { + Struct topicData = struct.instance(TOPIC_METADATA_KEY_NAME); + topicData.set(TOPIC_KEY_NAME, metadata.topic); + topicData.set(TOPIC_ERROR_CODE_KEY_NAME, metadata.error.code()); + // This field only exists in v1+ + if (topicData.hasField(IS_INTERNAL_KEY_NAME)) + topicData.set(IS_INTERNAL_KEY_NAME, metadata.isInternal()); + + List<Struct> partitionMetadataArray = new ArrayList<>(metadata.partitionMetadata.size()); + for (PartitionMetadata partitionMetadata : metadata.partitionMetadata()) { + Struct partitionData = topicData.instance(PARTITION_METADATA_KEY_NAME); + partitionData.set(PARTITION_ERROR_CODE_KEY_NAME, partitionMetadata.error.code()); + partitionData.set(PARTITION_KEY_NAME, partitionMetadata.partition); + partitionData.set(LEADER_KEY_NAME, partitionMetadata.leader.id()); + ArrayList<Integer> replicas = new ArrayList<>(partitionMetadata.replicas.size()); + for (Node node : partitionMetadata.replicas) + replicas.add(node.id()); + partitionData.set(REPLICAS_KEY_NAME, replicas.toArray()); + ArrayList<Integer> isr = new ArrayList<>(partitionMetadata.isr.size()); + for (Node node : partitionMetadata.isr) + isr.add(node.id()); + partitionData.set(ISR_KEY_NAME, isr.toArray()); + partitionMetadataArray.add(partitionData); + + } + topicData.set(PARTITION_METADATA_KEY_NAME, partitionMetadataArray.toArray()); + topicMetadataArray.add(topicData); + } + struct.set(TOPIC_METADATA_KEY_NAME, topicMetadataArray.toArray()); + return struct; + } }
