You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by jg...@apache.org on 2017/02/22 23:30:17 UTC
[1/2] kafka git commit: MINOR: Move ProtoUtils methods to ApiKeys
Repository: kafka
Updated Branches:
refs/heads/trunk 5916ef022 -> 015f1d738
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java
index 94de4b1..189575e 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java
@@ -23,7 +23,6 @@ import java.util.Map;
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.Struct;
import org.apache.kafka.common.utils.CollectionUtils;
@@ -138,12 +137,12 @@ public class OffsetFetchResponse extends AbstractResponse {
}
public static OffsetFetchResponse parse(ByteBuffer buffer, short version) {
- return new OffsetFetchResponse(ProtoUtils.parseResponse(ApiKeys.OFFSET_FETCH.id, version, buffer));
+ return new OffsetFetchResponse(ApiKeys.OFFSET_FETCH.parseResponse(version, buffer));
}
@Override
protected Struct toStruct(short version) {
- Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.OFFSET_FETCH.id, version));
+ Struct struct = new Struct(ApiKeys.OFFSET_FETCH.responseSchema(version));
Map<String, Map<Integer, PartitionData>> topicsData = CollectionUtils.groupDataByTopic(responseData);
List<Struct> topicArray = new ArrayList<>();
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java
index df70e20..7b454e8 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java
@@ -17,7 +17,6 @@ import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.UnsupportedVersionException;
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.Struct;
import org.apache.kafka.common.record.MemoryRecords;
import org.apache.kafka.common.utils.CollectionUtils;
@@ -107,7 +106,7 @@ public class ProduceRequest extends AbstractRequest {
*/
@Override
public Struct toStruct() {
- Struct struct = new Struct(ProtoUtils.requestSchema(ApiKeys.PRODUCE.id, version()));
+ Struct struct = new Struct(ApiKeys.PRODUCE.requestSchema(version()));
Map<String, Map<Integer, MemoryRecords>> recordsByTopic = CollectionUtils.groupDataByTopic(partitionRecords);
struct.set(ACKS_KEY_NAME, acks);
struct.set(TIMEOUT_KEY_NAME, timeout);
@@ -150,7 +149,7 @@ public class ProduceRequest extends AbstractRequest {
return new ProduceResponse(responseMap);
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.PRODUCE.id)));
+ versionId, this.getClass().getSimpleName(), ApiKeys.PRODUCE.latestVersion()));
}
}
@@ -170,7 +169,7 @@ public class ProduceRequest extends AbstractRequest {
partitionRecords.clear();
}
- public static ProduceRequest parse(ByteBuffer buffer, short versionId) {
- return new ProduceRequest(ProtoUtils.parseRequest(ApiKeys.PRODUCE.id, versionId, buffer), versionId);
+ public static ProduceRequest parse(ByteBuffer buffer, short version) {
+ return new ProduceRequest(ApiKeys.PRODUCE.parseRequest(version, buffer), version);
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/common/requests/ProduceResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/ProduceResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/ProduceResponse.java
index 7a022af..8234732 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/ProduceResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/ProduceResponse.java
@@ -15,7 +15,6 @@ package org.apache.kafka.common.requests;
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.Struct;
import org.apache.kafka.common.record.Record;
import org.apache.kafka.common.utils.CollectionUtils;
@@ -107,7 +106,7 @@ public class ProduceResponse extends AbstractResponse {
@Override
protected Struct toStruct(short version) {
- Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.PRODUCE.id, version));
+ Struct struct = new Struct(ApiKeys.PRODUCE.responseSchema(version));
Map<String, Map<Integer, PartitionResponse>> responseByTopic = CollectionUtils.groupDataByTopic(responses);
List<Struct> topicDatas = new ArrayList<>(responseByTopic.size());
@@ -174,6 +173,6 @@ public class ProduceResponse extends AbstractResponse {
}
public static ProduceResponse parse(ByteBuffer buffer, short version) {
- return new ProduceResponse(ProtoUtils.responseSchema(ApiKeys.PRODUCE.id, version).read(buffer));
+ return new ProduceResponse(ApiKeys.PRODUCE.responseSchema(version).read(buffer));
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/common/requests/SaslHandshakeRequest.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/SaslHandshakeRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/SaslHandshakeRequest.java
index a1f3f0e..9cc3f1f 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/SaslHandshakeRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/SaslHandshakeRequest.java
@@ -24,7 +24,6 @@ import java.util.List;
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.Struct;
@@ -44,7 +43,7 @@ public class SaslHandshakeRequest extends AbstractRequest {
private final String mechanism;
public SaslHandshakeRequest(String mechanism) {
- super(ProtoUtils.latestVersion(ApiKeys.SASL_HANDSHAKE.id));
+ super(ApiKeys.SASL_HANDSHAKE.latestVersion());
this.mechanism = mechanism;
}
@@ -66,17 +65,17 @@ public class SaslHandshakeRequest extends AbstractRequest {
return new SaslHandshakeResponse(Errors.forException(e), enabledMechanisms);
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.SASL_HANDSHAKE.id)));
+ versionId, this.getClass().getSimpleName(), ApiKeys.SASL_HANDSHAKE.latestVersion()));
}
}
- public static SaslHandshakeRequest parse(ByteBuffer buffer, short versionId) {
- return new SaslHandshakeRequest(ProtoUtils.parseRequest(ApiKeys.SASL_HANDSHAKE.id, versionId, buffer), versionId);
+ public static SaslHandshakeRequest parse(ByteBuffer buffer, short version) {
+ return new SaslHandshakeRequest(ApiKeys.SASL_HANDSHAKE.parseRequest(version, buffer), version);
}
@Override
protected Struct toStruct() {
- Struct struct = new Struct(ProtoUtils.requestSchema(ApiKeys.SASL_HANDSHAKE.id, version()));
+ Struct struct = new Struct(ApiKeys.SASL_HANDSHAKE.requestSchema(version()));
struct.set(MECHANISM_KEY_NAME, mechanism);
return struct;
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/common/requests/SaslHandshakeResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/SaslHandshakeResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/SaslHandshakeResponse.java
index 9d38c6a..e1a4c87 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/SaslHandshakeResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/SaslHandshakeResponse.java
@@ -25,7 +25,6 @@ import java.util.List;
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.Struct;
@@ -66,7 +65,7 @@ public class SaslHandshakeResponse extends AbstractResponse {
@Override
public Struct toStruct(short version) {
- Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.SASL_HANDSHAKE.id, version));
+ Struct struct = new Struct(ApiKeys.SASL_HANDSHAKE.responseSchema(version));
struct.set(ERROR_CODE_KEY_NAME, error.code());
struct.set(ENABLED_MECHANISMS_KEY_NAME, enabledMechanisms.toArray());
return struct;
@@ -77,7 +76,7 @@ public class SaslHandshakeResponse extends AbstractResponse {
}
public static SaslHandshakeResponse parse(ByteBuffer buffer, short version) {
- return new SaslHandshakeResponse(ProtoUtils.parseResponse(ApiKeys.SASL_HANDSHAKE.id, version, buffer));
+ return new SaslHandshakeResponse(ApiKeys.SASL_HANDSHAKE.parseResponse(version, buffer));
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/common/requests/StopReplicaRequest.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/StopReplicaRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/StopReplicaRequest.java
index 91806f1..7b79cd8 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/StopReplicaRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/StopReplicaRequest.java
@@ -16,7 +16,6 @@ package org.apache.kafka.common.requests;
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.Struct;
import org.apache.kafka.common.utils.Utils;
@@ -113,7 +112,7 @@ public class StopReplicaRequest extends AbstractRequest {
return new StopReplicaResponse(Errors.NONE, responses);
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.STOP_REPLICA.id)));
+ versionId, this.getClass().getSimpleName(), ApiKeys.STOP_REPLICA.latestVersion()));
}
}
@@ -133,13 +132,13 @@ public class StopReplicaRequest extends AbstractRequest {
return partitions;
}
- public static StopReplicaRequest parse(ByteBuffer buffer, short versionId) {
- return new StopReplicaRequest(ProtoUtils.parseRequest(ApiKeys.STOP_REPLICA.id, versionId, buffer), versionId);
+ public static StopReplicaRequest parse(ByteBuffer buffer, short version) {
+ return new StopReplicaRequest(ApiKeys.STOP_REPLICA.parseRequest(version, buffer), version);
}
@Override
protected Struct toStruct() {
- Struct struct = new Struct(ProtoUtils.requestSchema(ApiKeys.STOP_REPLICA.id, version()));
+ Struct struct = new Struct(ApiKeys.STOP_REPLICA.requestSchema(version()));
struct.set(CONTROLLER_ID_KEY_NAME, controllerId);
struct.set(CONTROLLER_EPOCH_KEY_NAME, controllerEpoch);
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/common/requests/StopReplicaResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/StopReplicaResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/StopReplicaResponse.java
index 5ae5cc1..617b1c6 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/StopReplicaResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/StopReplicaResponse.java
@@ -16,7 +16,6 @@ package org.apache.kafka.common.requests;
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.Struct;
import java.nio.ByteBuffer;
@@ -69,13 +68,13 @@ public class StopReplicaResponse extends AbstractResponse {
return error;
}
- public static StopReplicaResponse parse(ByteBuffer buffer, short versionId) {
- return new StopReplicaResponse(ProtoUtils.parseResponse(ApiKeys.STOP_REPLICA.id, versionId, buffer));
+ public static StopReplicaResponse parse(ByteBuffer buffer, short version) {
+ return new StopReplicaResponse(ApiKeys.STOP_REPLICA.parseResponse(version, buffer));
}
@Override
protected Struct toStruct(short version) {
- Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.STOP_REPLICA.id, version));
+ Struct struct = new Struct(ApiKeys.STOP_REPLICA.responseSchema(version));
List<Struct> responseDatas = new ArrayList<>(responses.size());
for (Map.Entry<TopicPartition, Errors> response : responses.entrySet()) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/common/requests/SyncGroupRequest.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/SyncGroupRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/SyncGroupRequest.java
index 7ad5c9a..b55ccff 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/SyncGroupRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/SyncGroupRequest.java
@@ -18,7 +18,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.Struct;
import org.apache.kafka.common.utils.Utils;
@@ -108,7 +107,7 @@ public class SyncGroupRequest extends AbstractRequest {
ByteBuffer.wrap(new byte[]{}));
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.SYNC_GROUP.id)));
+ versionId, this.getClass().getSimpleName(), ApiKeys.SYNC_GROUP.latestVersion()));
}
}
@@ -128,13 +127,13 @@ public class SyncGroupRequest extends AbstractRequest {
return memberId;
}
- public static SyncGroupRequest parse(ByteBuffer buffer, short versionId) {
- return new SyncGroupRequest(ProtoUtils.parseRequest(ApiKeys.SYNC_GROUP.id, versionId, buffer), versionId);
+ public static SyncGroupRequest parse(ByteBuffer buffer, short version) {
+ return new SyncGroupRequest(ApiKeys.SYNC_GROUP.parseRequest(version, buffer), version);
}
@Override
protected Struct toStruct() {
- Struct struct = new Struct(ProtoUtils.requestSchema(ApiKeys.SYNC_GROUP.id, version()));
+ Struct struct = new Struct(ApiKeys.SYNC_GROUP.requestSchema(version()));
struct.set(GROUP_ID_KEY_NAME, groupId);
struct.set(GENERATION_ID_KEY_NAME, generationId);
struct.set(MEMBER_ID_KEY_NAME, memberId);
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/common/requests/SyncGroupResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/SyncGroupResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/SyncGroupResponse.java
index ff198aa..148815c 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/SyncGroupResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/SyncGroupResponse.java
@@ -18,7 +18,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.Struct;
import java.nio.ByteBuffer;
@@ -62,14 +61,14 @@ public class SyncGroupResponse extends AbstractResponse {
@Override
protected Struct toStruct(short version) {
- Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.SYNC_GROUP.id, version));
+ Struct struct = new Struct(ApiKeys.SYNC_GROUP.responseSchema(version));
struct.set(ERROR_CODE_KEY_NAME, error.code());
struct.set(MEMBER_ASSIGNMENT_KEY_NAME, memberState);
return struct;
}
public static SyncGroupResponse parse(ByteBuffer buffer, short version) {
- return new SyncGroupResponse(ProtoUtils.parseResponse(ApiKeys.SYNC_GROUP.id, version, buffer));
+ return new SyncGroupResponse(ApiKeys.SYNC_GROUP.parseResponse(version, buffer));
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataRequest.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataRequest.java
index 8dd852d..98bf83b 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataRequest.java
@@ -18,7 +18,6 @@ import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.network.ListenerName;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
-import org.apache.kafka.common.protocol.ProtoUtils;
import org.apache.kafka.common.protocol.SecurityProtocol;
import org.apache.kafka.common.protocol.types.Struct;
import org.apache.kafka.common.utils.Utils;
@@ -226,7 +225,7 @@ public class UpdateMetadataRequest extends AbstractRequest {
@Override
protected Struct toStruct() {
short version = version();
- Struct struct = new Struct(ProtoUtils.requestSchema(ApiKeys.UPDATE_METADATA_KEY.id, version));
+ Struct struct = new Struct(ApiKeys.UPDATE_METADATA_KEY.requestSchema(version));
struct.set(CONTROLLER_ID_KEY_NAME, controllerId);
struct.set(CONTROLLER_EPOCH_KEY_NAME, controllerEpoch);
@@ -288,7 +287,7 @@ public class UpdateMetadataRequest extends AbstractRequest {
return new UpdateMetadataResponse(Errors.forException(e));
else
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.UPDATE_METADATA_KEY.id)));
+ versionId, this.getClass().getSimpleName(), ApiKeys.UPDATE_METADATA_KEY.latestVersion()));
}
public int controllerId() {
@@ -307,9 +306,8 @@ public class UpdateMetadataRequest extends AbstractRequest {
return liveBrokers;
}
- public static UpdateMetadataRequest parse(ByteBuffer buffer, short versionId) {
- return new UpdateMetadataRequest(ProtoUtils.parseRequest(ApiKeys.UPDATE_METADATA_KEY.id, versionId, buffer),
- versionId);
+ public static UpdateMetadataRequest parse(ByteBuffer buffer, short version) {
+ return new UpdateMetadataRequest(ApiKeys.UPDATE_METADATA_KEY.parseRequest(version, buffer), version);
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataResponse.java
index 0032fca..5b8b46d 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataResponse.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.Struct;
import java.nio.ByteBuffer;
@@ -44,12 +43,12 @@ public class UpdateMetadataResponse extends AbstractResponse {
}
public static UpdateMetadataResponse parse(ByteBuffer buffer, short version) {
- return new UpdateMetadataResponse(ProtoUtils.parseResponse(ApiKeys.UPDATE_METADATA_KEY.id, version, buffer));
+ return new UpdateMetadataResponse(ApiKeys.UPDATE_METADATA_KEY.parseResponse(version, buffer));
}
@Override
protected Struct toStruct(short version) {
- Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.UPDATE_METADATA_KEY.id, version));
+ Struct struct = new Struct(ApiKeys.UPDATE_METADATA_KEY.responseSchema(version));
struct.set(ERROR_CODE_KEY_NAME, error.code());
return struct;
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java b/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java
index c89cc24..eb6a259 100644
--- a/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java
@@ -21,7 +21,6 @@ import org.apache.kafka.common.Node;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.network.NetworkReceive;
import org.apache.kafka.common.protocol.ApiKeys;
-import org.apache.kafka.common.protocol.ProtoUtils;
import org.apache.kafka.common.protocol.types.Struct;
import org.apache.kafka.common.record.MemoryRecords;
import org.apache.kafka.common.requests.ApiVersionsResponse;
@@ -134,7 +133,7 @@ public class NetworkClientTest {
networkClient.poll(1, time.milliseconds());
assertEquals(1, networkClient.inFlightRequestCount());
ResponseHeader respHeader = new ResponseHeader(request.correlationId());
- Struct resp = new Struct(ProtoUtils.currentResponseSchema(ApiKeys.PRODUCE.id));
+ Struct resp = new Struct(ApiKeys.PRODUCE.responseSchema(ApiKeys.PRODUCE.latestVersion()));
resp.set("responses", new Object[0]);
Struct responseHeaderStruct = respHeader.toStruct();
int size = responseHeaderStruct.sizeOf() + resp.sizeOf();
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/test/java/org/apache/kafka/clients/NodeApiVersionsTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/clients/NodeApiVersionsTest.java b/clients/src/test/java/org/apache/kafka/clients/NodeApiVersionsTest.java
index 53c47c8..9f0868f 100644
--- a/clients/src/test/java/org/apache/kafka/clients/NodeApiVersionsTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/NodeApiVersionsTest.java
@@ -18,7 +18,6 @@ package org.apache.kafka.clients;
import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.protocol.ApiKeys;
-import org.apache.kafka.common.protocol.ProtoUtils;
import org.apache.kafka.common.requests.ApiVersionsResponse;
import org.apache.kafka.common.requests.ApiVersionsResponse.ApiVersion;
import org.junit.Assert;
@@ -64,8 +63,7 @@ public class NodeApiVersionsTest {
} else if (apiKey == ApiKeys.DELETE_TOPICS) {
versionList.add(new ApiVersion(apiKey.id, (short) 10000, (short) 10001));
} else {
- versionList.add(new ApiVersion(apiKey.id,
- ProtoUtils.oldestVersion(apiKey.id), ProtoUtils.latestVersion(apiKey.id)));
+ versionList.add(new ApiVersion(apiKey));
}
}
NodeApiVersions versions = new NodeApiVersions(versionList);
@@ -80,15 +78,15 @@ public class NodeApiVersionsTest {
} else {
bld.append(apiKey.name).append("(").
append(apiKey.id).append("): ");
- if (ProtoUtils.oldestVersion(apiKey.id) ==
- ProtoUtils.latestVersion(apiKey.id)) {
- bld.append(ProtoUtils.oldestVersion(apiKey.id));
+ if (apiKey.oldestVersion() ==
+ apiKey.latestVersion()) {
+ bld.append(apiKey.oldestVersion());
} else {
- bld.append(ProtoUtils.oldestVersion(apiKey.id)).
+ bld.append(apiKey.oldestVersion()).
append(" to ").
- append(ProtoUtils.latestVersion(apiKey.id));
+ append(apiKey.latestVersion());
}
- bld.append(" [usable: ").append(ProtoUtils.latestVersion(apiKey.id)).
+ bld.append(" [usable: ").append(apiKey.latestVersion()).
append("]");
}
prefix = ", ";
@@ -129,7 +127,7 @@ public class NodeApiVersionsTest {
versionList.add(new ApiVersion((short) 100, (short) 0, (short) 1));
NodeApiVersions versions = new NodeApiVersions(versionList);
for (ApiKeys apiKey: ApiKeys.values()) {
- assertEquals(ProtoUtils.latestVersion(apiKey.id), versions.usableVersion(apiKey));
+ assertEquals(apiKey.latestVersion(), versions.usableVersion(apiKey));
}
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/test/java/org/apache/kafka/common/protocol/ApiKeysTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/protocol/ApiKeysTest.java b/clients/src/test/java/org/apache/kafka/common/protocol/ApiKeysTest.java
index f177ae6..8917a22 100644
--- a/clients/src/test/java/org/apache/kafka/common/protocol/ApiKeysTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/protocol/ApiKeysTest.java
@@ -31,4 +31,9 @@ public class ApiKeysTest {
ApiKeys.forId(10000);
}
+ @Test(expected = IllegalArgumentException.class)
+ public void schemaVersionOutOfRange() {
+ ApiKeys.PRODUCE.requestSchema((short) Protocol.REQUESTS[ApiKeys.PRODUCE.id].length);
+ }
+
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/test/java/org/apache/kafka/common/protocol/ProtoUtilsTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/protocol/ProtoUtilsTest.java b/clients/src/test/java/org/apache/kafka/common/protocol/ProtoUtilsTest.java
deleted file mode 100644
index 440ca49..0000000
--- a/clients/src/test/java/org/apache/kafka/common/protocol/ProtoUtilsTest.java
+++ /dev/null
@@ -1,26 +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 org.apache.kafka.common.protocol;
-
-import org.junit.Test;
-
-public class ProtoUtilsTest {
- @Test(expected = IllegalArgumentException.class)
- public void schemaVersionOutOfRange() {
- ProtoUtils.requestSchema(ApiKeys.PRODUCE.id, Protocol.REQUESTS[ApiKeys.PRODUCE.id].length);
- }
-}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/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 d0b9639..1367ba1 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
@@ -22,7 +22,6 @@ import org.apache.kafka.common.network.ListenerName;
import org.apache.kafka.common.network.Send;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
-import org.apache.kafka.common.protocol.ProtoUtils;
import org.apache.kafka.common.protocol.SecurityProtocol;
import org.apache.kafka.common.protocol.types.Struct;
import org.apache.kafka.common.record.MemoryRecords;
@@ -169,7 +168,7 @@ public class RequestResponseTest {
}
private void checkOlderFetchVersions() throws Exception {
- int latestVersion = ProtoUtils.latestVersion(ApiKeys.FETCH.id);
+ int latestVersion = ApiKeys.FETCH.latestVersion();
for (int i = 0; i < latestVersion; ++i) {
checkErrorResponse(createFetchRequest(i), new UnknownServerException());
checkRequest(createFetchRequest(i));
@@ -220,11 +219,11 @@ public class RequestResponseTest {
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),
+ assertEquals("Should use schema version 0", ApiKeys.PRODUCE.responseSchema((short) 0),
v0Response.toStruct((short) 0).schema());
- assertEquals("Should use schema version 1", ProtoUtils.responseSchema(ApiKeys.PRODUCE.id, 1),
+ assertEquals("Should use schema version 1", ApiKeys.PRODUCE.responseSchema((short) 1),
v1Response.toStruct((short) 1).schema());
- assertEquals("Should use schema version 2", ProtoUtils.responseSchema(ApiKeys.PRODUCE.id, 2),
+ assertEquals("Should use schema version 2", ApiKeys.PRODUCE.responseSchema((short) 2),
v2Response.toStruct((short) 2).schema());
assertEquals("Response data does not match", responseData, v0Response.responses());
assertEquals("Response data does not match", responseData, v1Response.responses());
@@ -242,9 +241,9 @@ public class RequestResponseTest {
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),
+ assertEquals("Should use schema version 0", ApiKeys.FETCH.responseSchema((short) 0),
v0Response.toStruct((short) 0).schema());
- assertEquals("Should use schema version 1", ProtoUtils.responseSchema(ApiKeys.FETCH.id, 1),
+ assertEquals("Should use schema version 1", ApiKeys.FETCH.responseSchema((short) 1),
v1Response.toStruct((short) 1).schema());
assertEquals("Response data does not match", responseData, v0Response.responseData());
assertEquals("Response data does not match", responseData, v1Response.responseData());
@@ -253,7 +252,7 @@ public class RequestResponseTest {
@Test
public void verifyFetchResponseFullWrite() throws Exception {
FetchResponse fetchResponse = createFetchResponse();
- RequestHeader header = new RequestHeader(ApiKeys.FETCH.id, ProtoUtils.latestVersion(ApiKeys.FETCH.id),
+ RequestHeader header = new RequestHeader(ApiKeys.FETCH.id, ApiKeys.FETCH.latestVersion(),
"client", 15);
Send send = fetchResponse.toSend("1", header);
@@ -272,7 +271,7 @@ public class RequestResponseTest {
assertEquals(header.correlationId(), responseHeader.correlationId());
// read the body
- Struct responseBody = ProtoUtils.responseSchema(ApiKeys.FETCH.id, header.apiVersion()).read(buf);
+ Struct responseBody = ApiKeys.FETCH.responseSchema(header.apiVersion()).read(buf);
assertEquals(fetchResponse.toStruct(header.apiVersion()), responseBody);
assertEquals(size, responseHeader.sizeOf() + responseBody.sizeOf());
@@ -281,7 +280,7 @@ public class RequestResponseTest {
@Test
public void testControlledShutdownResponse() {
ControlledShutdownResponse response = createControlledShutdownResponse();
- short version = ProtoUtils.latestVersion(ApiKeys.CONTROLLED_SHUTDOWN_KEY.id);
+ short version = ApiKeys.CONTROLLED_SHUTDOWN_KEY.latestVersion();
Struct struct = response.toStruct(version);
ByteBuffer buffer = toBuffer(struct);
ControlledShutdownResponse deserialized = ControlledShutdownResponse.parse(buffer, version);
@@ -403,11 +402,11 @@ public class RequestResponseTest {
Map<TopicPartition, ListOffsetRequest.PartitionData> offsetData = Collections.singletonMap(
new TopicPartition("test", 0),
new ListOffsetRequest.PartitionData(1000000L, 10));
- return ListOffsetRequest.Builder.forConsumer((short) 0).setOffsetData(offsetData).build((short) version);
+ return ListOffsetRequest.Builder.forConsumer(false).setOffsetData(offsetData).build((short) version);
} else if (version == 1) {
Map<TopicPartition, Long> offsetData = Collections.singletonMap(
new TopicPartition("test", 0), 1000000L);
- return ListOffsetRequest.Builder.forConsumer((short) 1).setTargetTimes(offsetData).build((short) version);
+ return ListOffsetRequest.Builder.forConsumer(true).setTargetTimes(offsetData).build((short) version);
} else {
throw new IllegalArgumentException("Illegal ListOffsetRequest version " + version);
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/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 3a9e0ce..8a93532 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
@@ -27,7 +27,6 @@ import org.apache.kafka.common.network.Selector;
import org.apache.kafka.common.network.Send;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
-import org.apache.kafka.common.protocol.ProtoUtils;
import org.apache.kafka.common.protocol.SecurityProtocol;
import org.apache.kafka.common.requests.AbstractRequest;
import org.apache.kafka.common.requests.AbstractResponse;
@@ -758,8 +757,8 @@ public class SaslAuthenticatorTest {
// Send ApiVersionsRequest and check response
ApiVersionsResponse versionsResponse = sendVersionRequestReceiveResponse(node);
- assertEquals(ProtoUtils.oldestVersion(ApiKeys.SASL_HANDSHAKE.id), versionsResponse.apiVersion(ApiKeys.SASL_HANDSHAKE.id).minVersion);
- assertEquals(ProtoUtils.latestVersion(ApiKeys.SASL_HANDSHAKE.id), versionsResponse.apiVersion(ApiKeys.SASL_HANDSHAKE.id).maxVersion);
+ assertEquals(ApiKeys.SASL_HANDSHAKE.oldestVersion(), versionsResponse.apiVersion(ApiKeys.SASL_HANDSHAKE.id).minVersion);
+ assertEquals(ApiKeys.SASL_HANDSHAKE.latestVersion(), versionsResponse.apiVersion(ApiKeys.SASL_HANDSHAKE.id).maxVersion);
// Send SaslHandshakeRequest and check response
SaslHandshakeResponse handshakeResponse = sendHandshakeRequestReceiveResponse(node);
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/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 4f71258..3368c09 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, ProtoUtils, SecurityProtocol}
+import org.apache.kafka.common.protocol.{ApiKeys, Errors, SecurityProtocol}
import org.apache.kafka.common.requests._
import CreateTopicsRequest.TopicDetails
import org.apache.kafka.common.security.auth.KafkaPrincipal
@@ -191,12 +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))
- val version = ProtoUtils.latestVersion(ApiKeys.FETCH.id)
+ val version = ApiKeys.FETCH.latestVersion
requests.FetchRequest.Builder.forReplica(version, 5000, 100, Int.MaxValue, partitionMap).build()
}
private def createListOffsetsRequest = {
- requests.ListOffsetRequest.Builder.forConsumer(0).setTargetTimes(
+ requests.ListOffsetRequest.Builder.forConsumer(false).setTargetTimes(
Map(tp -> (0L: java.lang.Long)).asJava).
build()
}
@@ -215,7 +215,7 @@ 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
- val version = ProtoUtils.latestVersion(ApiKeys.UPDATE_METADATA_KEY.id)
+ val version = ApiKeys.UPDATE_METADATA_KEY.latestVersion
new requests.UpdateMetadataRequest.Builder(version, brokerId, Int.MaxValue, partitionState, brokers).build()
}
@@ -773,7 +773,7 @@ class AuthorizerIntegrationTest extends BaseRequestTest {
@Test
def testUnauthorizedDeleteWithoutDescribe() {
val response = connectAndSend(deleteTopicsRequest, ApiKeys.DELETE_TOPICS)
- val version = ProtoUtils.latestVersion(ApiKeys.DELETE_TOPICS.id)
+ val version = ApiKeys.DELETE_TOPICS.latestVersion
val deleteResponse = DeleteTopicsResponse.parse(response, version)
assertEquals(Errors.UNKNOWN_TOPIC_OR_PARTITION, deleteResponse.errors.asScala.head._2)
}
@@ -782,7 +782,7 @@ class AuthorizerIntegrationTest extends BaseRequestTest {
def testUnauthorizedDeleteWithDescribe() {
addAndVerifyAcls(Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, Describe)), deleteTopicResource)
val response = connectAndSend(deleteTopicsRequest, ApiKeys.DELETE_TOPICS)
- val version = ProtoUtils.latestVersion(ApiKeys.DELETE_TOPICS.id)
+ val version = ApiKeys.DELETE_TOPICS.latestVersion
val deleteResponse = DeleteTopicsResponse.parse(response, version)
assertEquals(Errors.TOPIC_AUTHORIZATION_FAILED, deleteResponse.errors.asScala.head._2)
@@ -792,7 +792,7 @@ class AuthorizerIntegrationTest extends BaseRequestTest {
def testDeleteWithWildCardAuth() {
addAndVerifyAcls(Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, Delete)), new Resource(Topic, "*"))
val response = connectAndSend(deleteTopicsRequest, ApiKeys.DELETE_TOPICS)
- val version = ProtoUtils.latestVersion(ApiKeys.DELETE_TOPICS.id)
+ val version = ApiKeys.DELETE_TOPICS.latestVersion
val deleteResponse = DeleteTopicsResponse.parse(response, version)
assertEquals(Errors.NONE, deleteResponse.errors.asScala.head._2)
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/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 448fce1..b82ddf9 100644
--- a/core/src/test/scala/unit/kafka/server/AbstractCreateTopicsRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/AbstractCreateTopicsRequestTest.scala
@@ -22,7 +22,7 @@ import java.util.Properties
import kafka.network.SocketServer
import kafka.utils.TestUtils
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.{CreateTopicsRequest, CreateTopicsResponse, MetadataRequest, MetadataResponse}
import org.junit.Assert.{assertEquals, assertFalse, assertNotNull, assertTrue}
@@ -147,9 +147,8 @@ class AbstractCreateTopicsRequestTest extends BaseRequestTest {
}
protected def sendMetadataRequest(request: MetadataRequest, destination: SocketServer = anySocketServer): MetadataResponse = {
- val version = ProtoUtils.latestVersion(ApiKeys.METADATA.id)
val response = connectAndSend(request, ApiKeys.METADATA, destination = destination)
- MetadataResponse.parse(response, version)
+ MetadataResponse.parse(response, ApiKeys.METADATA.latestVersion)
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/core/src/test/scala/unit/kafka/server/ApiVersionsTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/server/ApiVersionsTest.scala b/core/src/test/scala/unit/kafka/server/ApiVersionsTest.scala
index 981ed6a..2555a91 100644
--- a/core/src/test/scala/unit/kafka/server/ApiVersionsTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ApiVersionsTest.scala
@@ -18,7 +18,7 @@
package kafka.server
import org.apache.kafka.common.requests.ApiVersionsResponse
-import org.apache.kafka.common.protocol.{ApiKeys, ProtoUtils, Protocol}
+import org.apache.kafka.common.protocol.{ApiKeys, Protocol}
import org.junit.Assert._
import org.junit.Test
@@ -32,8 +32,8 @@ class ApiVersionsTest {
for (key <- ApiKeys.values) {
val version = ApiVersionsResponse.API_VERSIONS_RESPONSE.apiVersion(key.id)
assertNotNull(s"Could not find ApiVersion for API ${key.name}", version)
- assertEquals(s"Incorrect min version for Api ${key.name}.", version.minVersion, ProtoUtils.oldestVersion(key.id))
- assertEquals(s"Incorrect max version for Api ${key.name}.", version.maxVersion, ProtoUtils.latestVersion(key.id))
+ assertEquals(s"Incorrect min version for Api ${key.name}.", version.minVersion, key.oldestVersion)
+ assertEquals(s"Incorrect max version for Api ${key.name}.", version.maxVersion, key.latestVersion)
// Check if versions less than min version are indeed set as null, i.e., deprecated.
for (i <- 0 until version.minVersion) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/core/src/test/scala/unit/kafka/server/FetchRequestTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/server/FetchRequestTest.scala b/core/src/test/scala/unit/kafka/server/FetchRequestTest.scala
index 64be5b3..5616956 100644
--- a/core/src/test/scala/unit/kafka/server/FetchRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/FetchRequestTest.scala
@@ -24,7 +24,7 @@ import kafka.utils.TestUtils
import kafka.utils.TestUtils._
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.kafka.common.TopicPartition
-import org.apache.kafka.common.protocol.{ApiKeys, Errors, ProtoUtils}
+import org.apache.kafka.common.protocol.{ApiKeys, Errors}
import org.apache.kafka.common.record.LogEntry
import org.apache.kafka.common.requests.{FetchRequest, FetchResponse}
import org.apache.kafka.common.serialization.StringSerializer
@@ -70,7 +70,7 @@ class FetchRequestTest extends BaseRequestTest {
private def sendFetchRequest(leaderId: Int, request: FetchRequest): FetchResponse = {
val response = connectAndSend(request, ApiKeys.FETCH, destination = brokerSocketServer(leaderId))
- FetchResponse.parse(response, ProtoUtils.latestVersion(ApiKeys.FETCH.id))
+ FetchResponse.parse(response, ApiKeys.FETCH.latestVersion)
}
@Test
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/core/src/test/scala/unit/kafka/server/MetadataCacheTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/server/MetadataCacheTest.scala b/core/src/test/scala/unit/kafka/server/MetadataCacheTest.scala
index 1b5007d..57c1846 100644
--- a/core/src/test/scala/unit/kafka/server/MetadataCacheTest.scala
+++ b/core/src/test/scala/unit/kafka/server/MetadataCacheTest.scala
@@ -22,7 +22,7 @@ import util.Arrays.asList
import kafka.common.BrokerEndPointNotAvailableException
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.network.ListenerName
-import org.apache.kafka.common.protocol.{ApiKeys, Errors, ProtoUtils, SecurityProtocol}
+import org.apache.kafka.common.protocol.{ApiKeys, Errors, SecurityProtocol}
import org.apache.kafka.common.requests.{PartitionState, UpdateMetadataRequest}
import org.apache.kafka.common.requests.UpdateMetadataRequest.{Broker, EndPoint}
import org.junit.Test
@@ -69,7 +69,7 @@ class MetadataCacheTest {
new TopicPartition(topic, 1) -> new PartitionState(controllerEpoch, 1, 1, asList(1), zkVersion, asSet(1)),
new TopicPartition(topic, 2) -> new PartitionState(controllerEpoch, 2, 2, asList(2), zkVersion, asSet(2)))
- val version = ProtoUtils.latestVersion(ApiKeys.UPDATE_METADATA_KEY.id)
+ val version = ApiKeys.UPDATE_METADATA_KEY.latestVersion
val updateMetadataRequest = new UpdateMetadataRequest.Builder(version, controllerId, controllerEpoch,
partitionStates.asJava, brokers.asJava).build()
cache.updateCache(15, updateMetadataRequest)
@@ -121,7 +121,7 @@ class MetadataCacheTest {
val partitionStates = Map(
new TopicPartition(topic, 0) -> new PartitionState(controllerEpoch, leader, leaderEpoch, asList(0), zkVersion, asSet(0)))
- val version = ProtoUtils.latestVersion(ApiKeys.UPDATE_METADATA_KEY.id)
+ val version = ApiKeys.UPDATE_METADATA_KEY.latestVersion
val updateMetadataRequest = new UpdateMetadataRequest.Builder(version, controllerId, controllerEpoch,
partitionStates.asJava, brokers.asJava).build()
cache.updateCache(15, updateMetadataRequest)
@@ -165,7 +165,7 @@ class MetadataCacheTest {
val partitionStates = Map(
new TopicPartition(topic, 0) -> new PartitionState(controllerEpoch, leader, leaderEpoch, isr, zkVersion, replicas))
- val version = ProtoUtils.latestVersion(ApiKeys.UPDATE_METADATA_KEY.id)
+ val version = ApiKeys.UPDATE_METADATA_KEY.latestVersion
val updateMetadataRequest = new UpdateMetadataRequest.Builder(version, controllerId, controllerEpoch,
partitionStates.asJava, brokers.asJava).build()
cache.updateCache(15, updateMetadataRequest)
@@ -225,7 +225,7 @@ class MetadataCacheTest {
val partitionStates = Map(
new TopicPartition(topic, 0) -> new PartitionState(controllerEpoch, leader, leaderEpoch, isr, zkVersion, replicas))
- val version = ProtoUtils.latestVersion(ApiKeys.UPDATE_METADATA_KEY.id)
+ val version = ApiKeys.UPDATE_METADATA_KEY.latestVersion
val updateMetadataRequest = new UpdateMetadataRequest.Builder(version, controllerId, controllerEpoch,
partitionStates.asJava, brokers.asJava).build()
cache.updateCache(15, updateMetadataRequest)
@@ -277,7 +277,7 @@ class MetadataCacheTest {
val isr = asList[Integer](0, 1)
val partitionStates = Map(
new TopicPartition(topic, 0) -> new PartitionState(controllerEpoch, leader, leaderEpoch, isr, 3, replicas))
- val version = ProtoUtils.latestVersion(ApiKeys.UPDATE_METADATA_KEY.id)
+ val version = ApiKeys.UPDATE_METADATA_KEY.latestVersion
val updateMetadataRequest = new UpdateMetadataRequest.Builder(version, 2, controllerEpoch, partitionStates.asJava,
brokers.asJava).build()
cache.updateCache(15, updateMetadataRequest)
@@ -310,7 +310,7 @@ class MetadataCacheTest {
val isr = asList[Integer](0, 1)
val partitionStates = Map(
new TopicPartition(topic, 0) -> new PartitionState(controllerEpoch, leader, leaderEpoch, isr, 3, replicas))
- val version = ProtoUtils.latestVersion(ApiKeys.UPDATE_METADATA_KEY.id)
+ val version = ApiKeys.UPDATE_METADATA_KEY.latestVersion
val updateMetadataRequest = new UpdateMetadataRequest.Builder(version, 2, controllerEpoch, partitionStates.asJava,
brokers.asJava).build()
cache.updateCache(15, updateMetadataRequest)
[2/2] kafka git commit: MINOR: Move ProtoUtils methods to ApiKeys
Posted by jg...@apache.org.
MINOR: Move ProtoUtils methods to ApiKeys
Also move `requireTimestamp` to `minVersion` logic from `Fetcher` to
`ListOffsetRequest.Builder.forConsumer()`.
Author: Ismael Juma <is...@juma.me.uk>
Reviewers: Colin P. Mccabe <cm...@confluent.io>, Jason Gustafson <ja...@confluent.io>
Closes #2580 from ijuma/move-proto-utils-to-api-keys
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/015f1d73
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/015f1d73
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/015f1d73
Branch: refs/heads/trunk
Commit: 015f1d73811e52b3b7a578cf4c1ca4c48b6c2f5c
Parents: 5916ef0
Author: Ismael Juma <is...@juma.me.uk>
Authored: Wed Feb 22 15:11:12 2017 -0800
Committer: Jason Gustafson <ja...@confluent.io>
Committed: Wed Feb 22 15:11:12 2017 -0800
----------------------------------------------------------------------
.../org/apache/kafka/clients/ClientRequest.java | 2 +-
.../org/apache/kafka/clients/NetworkClient.java | 12 ++--
.../apache/kafka/clients/NodeApiVersions.java | 22 +++---
.../clients/consumer/internals/Fetcher.java | 4 +-
.../apache/kafka/common/protocol/ApiKeys.java | 47 +++++++++++++
.../kafka/common/protocol/ProtoUtils.java | 73 --------------------
.../kafka/common/requests/AbstractRequest.java | 5 +-
.../kafka/common/requests/AbstractResponse.java | 3 +-
.../common/requests/ApiVersionsRequest.java | 7 +-
.../common/requests/ApiVersionsResponse.java | 11 +--
.../requests/ControlledShutdownRequest.java | 9 ++-
.../requests/ControlledShutdownResponse.java | 5 +-
.../common/requests/CreateTopicsRequest.java | 9 ++-
.../common/requests/CreateTopicsResponse.java | 5 +-
.../common/requests/DeleteTopicsRequest.java | 9 ++-
.../common/requests/DeleteTopicsResponse.java | 5 +-
.../common/requests/DescribeGroupsRequest.java | 13 ++--
.../common/requests/DescribeGroupsResponse.java | 7 +-
.../kafka/common/requests/FetchRequest.java | 7 +-
.../kafka/common/requests/FetchResponse.java | 5 +-
.../requests/GroupCoordinatorRequest.java | 10 ++-
.../requests/GroupCoordinatorResponse.java | 5 +-
.../kafka/common/requests/HeartbeatRequest.java | 9 ++-
.../common/requests/HeartbeatResponse.java | 5 +-
.../kafka/common/requests/JoinGroupRequest.java | 7 +-
.../common/requests/JoinGroupResponse.java | 5 +-
.../common/requests/LeaderAndIsrRequest.java | 9 ++-
.../common/requests/LeaderAndIsrResponse.java | 5 +-
.../common/requests/LeaveGroupRequest.java | 9 ++-
.../common/requests/LeaveGroupResponse.java | 5 +-
.../common/requests/ListGroupsRequest.java | 9 ++-
.../common/requests/ListGroupsResponse.java | 7 +-
.../common/requests/ListOffsetRequest.java | 13 ++--
.../common/requests/ListOffsetResponse.java | 5 +-
.../kafka/common/requests/MetadataRequest.java | 9 ++-
.../kafka/common/requests/MetadataResponse.java | 5 +-
.../common/requests/OffsetCommitRequest.java | 11 ++-
.../common/requests/OffsetCommitResponse.java | 5 +-
.../common/requests/OffsetFetchRequest.java | 9 ++-
.../common/requests/OffsetFetchResponse.java | 5 +-
.../kafka/common/requests/ProduceRequest.java | 9 ++-
.../kafka/common/requests/ProduceResponse.java | 5 +-
.../common/requests/SaslHandshakeRequest.java | 11 ++-
.../common/requests/SaslHandshakeResponse.java | 5 +-
.../common/requests/StopReplicaRequest.java | 9 ++-
.../common/requests/StopReplicaResponse.java | 7 +-
.../kafka/common/requests/SyncGroupRequest.java | 9 ++-
.../common/requests/SyncGroupResponse.java | 5 +-
.../common/requests/UpdateMetadataRequest.java | 10 ++-
.../common/requests/UpdateMetadataResponse.java | 5 +-
.../apache/kafka/clients/NetworkClientTest.java | 3 +-
.../kafka/clients/NodeApiVersionsTest.java | 18 +++--
.../kafka/common/protocol/ApiKeysTest.java | 5 ++
.../kafka/common/protocol/ProtoUtilsTest.java | 26 -------
.../common/requests/RequestResponseTest.java | 23 +++---
.../authenticator/SaslAuthenticatorTest.java | 5 +-
.../kafka/api/AuthorizerIntegrationTest.scala | 14 ++--
.../AbstractCreateTopicsRequestTest.scala | 5 +-
.../unit/kafka/server/ApiVersionsTest.scala | 6 +-
.../unit/kafka/server/FetchRequestTest.scala | 4 +-
.../unit/kafka/server/MetadataCacheTest.scala | 14 ++--
61 files changed, 253 insertions(+), 352 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/clients/ClientRequest.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/ClientRequest.java b/clients/src/main/java/org/apache/kafka/clients/ClientRequest.java
index a1973ad..564b391 100644
--- a/clients/src/main/java/org/apache/kafka/clients/ClientRequest.java
+++ b/clients/src/main/java/org/apache/kafka/clients/ClientRequest.java
@@ -75,7 +75,7 @@ public final class ClientRequest {
}
public RequestHeader makeHeader(short version) {
- return new RequestHeader(requestBuilder.apiKey().id, version, clientId, correlationId);
+ return new RequestHeader(apiKey().id, version, clientId, correlationId);
}
public AbstractRequest.Builder<?> requestBuilder() {
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
index a927910..79d467a 100644
--- a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
@@ -20,7 +20,6 @@ import org.apache.kafka.common.network.Selectable;
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.Struct;
import org.apache.kafka.common.requests.AbstractRequest;
import org.apache.kafka.common.requests.AbstractResponse;
@@ -314,12 +313,12 @@ public class NetworkClient implements KafkaClient {
String nodeId = clientRequest.destination();
RequestHeader header = clientRequest.makeHeader(request.version());
if (log.isDebugEnabled()) {
- int latestClientVersion = ProtoUtils.latestVersion(clientRequest.apiKey().id);
+ int latestClientVersion = clientRequest.apiKey().latestVersion();
if (header.apiVersion() == latestClientVersion) {
- log.trace("Sending {} {} to node {}.", ApiKeys.forId(header.apiKey()), request, nodeId);
+ log.trace("Sending {} {} to node {}.", clientRequest.apiKey(), request, nodeId);
} else {
log.debug("Using older server API v{} to send {} {} to node {}.",
- header.apiVersion(), ApiKeys.forId(header.apiKey()), request, nodeId);
+ header.apiVersion(), clientRequest.apiKey(), request, nodeId);
}
}
Send send = request.toSend(nodeId, header);
@@ -454,9 +453,8 @@ public class NetworkClient implements KafkaClient {
public static AbstractResponse parseResponse(ByteBuffer responseBuffer, RequestHeader requestHeader) {
ResponseHeader responseHeader = ResponseHeader.parse(responseBuffer);
// Always expect the response version id to be the same as the request version id
- short apiKey = requestHeader.apiKey();
- short apiVer = requestHeader.apiVersion();
- Struct responseBody = ProtoUtils.responseSchema(apiKey, apiVer).read(responseBuffer);
+ ApiKeys apiKey = ApiKeys.forId(requestHeader.apiKey());
+ Struct responseBody = apiKey.responseSchema(requestHeader.apiVersion()).read(responseBuffer);
correlate(requestHeader, responseHeader);
return AbstractResponse.getResponse(apiKey, responseBody);
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/clients/NodeApiVersions.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/NodeApiVersions.java b/clients/src/main/java/org/apache/kafka/clients/NodeApiVersions.java
index 906c226..0fcdf4d 100644
--- a/clients/src/main/java/org/apache/kafka/clients/NodeApiVersions.java
+++ b/clients/src/main/java/org/apache/kafka/clients/NodeApiVersions.java
@@ -14,7 +14,6 @@ package org.apache.kafka.clients;
import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.protocol.ApiKeys;
-import org.apache.kafka.common.protocol.ProtoUtils;
import org.apache.kafka.common.requests.ApiVersionsResponse.ApiVersion;
import org.apache.kafka.common.utils.Utils;
@@ -67,8 +66,7 @@ public class NodeApiVersions {
}
}
if (!exists) {
- apiVersions.add(new ApiVersion(apiKey.id, ProtoUtils.oldestVersion(apiKey.id),
- ProtoUtils.latestVersion(apiKey.id)));
+ apiVersions.add(new ApiVersion(apiKey));
}
}
return new NodeApiVersions(apiVersions);
@@ -77,16 +75,16 @@ public class NodeApiVersions {
public NodeApiVersions(Collection<ApiVersion> nodeApiVersions) {
this.nodeApiVersions = nodeApiVersions;
for (ApiVersion nodeApiVersion : nodeApiVersions) {
- int nodeApiKey = nodeApiVersion.apiKey;
// Newer brokers may support ApiKeys we don't know about, ignore them
- if (ApiKeys.hasId(nodeApiKey)) {
- short v = Utils.min(ProtoUtils.latestVersion(nodeApiKey), nodeApiVersion.maxVersion);
+ if (ApiKeys.hasId(nodeApiVersion.apiKey)) {
+ ApiKeys nodeApiKey = ApiKeys.forId(nodeApiVersion.apiKey);
+ short v = Utils.min(nodeApiKey.latestVersion(), nodeApiVersion.maxVersion);
if (v < nodeApiVersion.minVersion) {
- usableVersions.put(ApiKeys.forId(nodeApiKey), NODE_TOO_NEW);
- } else if (v < ProtoUtils.oldestVersion(nodeApiKey)) {
- usableVersions.put(ApiKeys.forId(nodeApiKey), NODE_TOO_OLD);
+ usableVersions.put(nodeApiKey, NODE_TOO_NEW);
+ } else if (v < nodeApiKey.oldestVersion()) {
+ usableVersions.put(nodeApiKey, NODE_TOO_OLD);
} else {
- usableVersions.put(ApiKeys.forId(nodeApiKey), v);
+ usableVersions.put(nodeApiKey, v);
}
}
}
@@ -101,10 +99,10 @@ public class NodeApiVersions {
throw new UnsupportedVersionException("The broker does not support " + apiKey);
else if (usableVersion == NODE_TOO_OLD)
throw new UnsupportedVersionException("The broker is too old to support " + apiKey +
- " version " + ProtoUtils.oldestVersion(apiKey.id));
+ " version " + apiKey.oldestVersion());
else if (usableVersion == NODE_TOO_NEW)
throw new UnsupportedVersionException("The broker is too new to support " + apiKey +
- " version " + ProtoUtils.latestVersion(apiKey.id));
+ " version " + apiKey.latestVersion());
else
return usableVersion;
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
index 655c27b..d67b940 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
@@ -605,9 +605,7 @@ public class Fetcher<K, V> implements SubscriptionState.Listener {
private RequestFuture<Map<TopicPartition, OffsetData>> sendListOffsetRequest(final Node node,
final Map<TopicPartition, Long> timestampsToSearch,
boolean requireTimestamp) {
- // If we need a timestamp in the response, the minimum RPC version we can send is v1. Otherwise, v0 is OK.
- short minVersion = requireTimestamp ? (short) 1 : (short) 0;
- ListOffsetRequest.Builder builder = ListOffsetRequest.Builder.forConsumer(minVersion)
+ ListOffsetRequest.Builder builder = ListOffsetRequest.Builder.forConsumer(requireTimestamp)
.setTargetTimes(timestampsToSearch);
log.trace("Sending ListOffsetRequest {} to broker {}", builder, node);
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java b/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java
index 239780c..50d89f1 100644
--- a/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java
+++ b/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java
@@ -16,6 +16,11 @@
*/
package org.apache.kafka.common.protocol;
+import org.apache.kafka.common.protocol.types.Schema;
+import org.apache.kafka.common.protocol.types.Struct;
+
+import java.nio.ByteBuffer;
+
/**
* Identifiers for all the Kafka APIs
*/
@@ -64,6 +69,8 @@ public enum ApiKeys {
public final String name;
ApiKeys(int id, String name) {
+ if (id < 0)
+ throw new IllegalArgumentException("id must not be negative, id: " + id);
this.id = (short) id;
this.name = name;
}
@@ -79,6 +86,46 @@ public enum ApiKeys {
return id >= MIN_API_KEY && id <= MAX_API_KEY;
}
+ public short latestVersion() {
+ if (id >= Protocol.CURR_VERSION.length)
+ throw new IllegalArgumentException("Latest version for API key " + this + " is not defined");
+ return Protocol.CURR_VERSION[id];
+ }
+
+ public short oldestVersion() {
+ if (id >= Protocol.MIN_VERSIONS.length)
+ throw new IllegalArgumentException("Oldest version for API key " + this + " is not defined");
+ return Protocol.MIN_VERSIONS[id];
+ }
+
+ public Schema requestSchema(short version) {
+ return schemaFor(Protocol.REQUESTS, version);
+ }
+
+ public Schema responseSchema(short version) {
+ return schemaFor(Protocol.RESPONSES, version);
+ }
+
+ public Struct parseRequest(short version, ByteBuffer buffer) {
+ return requestSchema(version).read(buffer);
+ }
+
+ public Struct parseResponse(short version, ByteBuffer buffer) {
+ return responseSchema(version).read(buffer);
+ }
+
+ private Schema schemaFor(Schema[][] schemas, short version) {
+ if (id > schemas.length)
+ throw new IllegalArgumentException("No schema available for API key " + this);
+ if (version < 0 || version > latestVersion())
+ throw new IllegalArgumentException("Invalid version for API key " + this + ": " + version);
+
+ Schema[] versions = schemas[id];
+ if (versions[version] == null)
+ throw new IllegalArgumentException("Unsupported version for API key " + this + ": " + version);
+ return versions[version];
+ }
+
private static String toHtml() {
final StringBuilder b = new StringBuilder();
b.append("<table class=\"data-table\"><tbody>\n");
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/common/protocol/ProtoUtils.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/ProtoUtils.java b/clients/src/main/java/org/apache/kafka/common/protocol/ProtoUtils.java
deleted file mode 100644
index ab2ebb1..0000000
--- a/clients/src/main/java/org/apache/kafka/common/protocol/ProtoUtils.java
+++ /dev/null
@@ -1,73 +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 org.apache.kafka.common.protocol;
-
-import java.nio.ByteBuffer;
-
-import org.apache.kafka.common.protocol.types.Schema;
-import org.apache.kafka.common.protocol.types.Struct;
-
-public class ProtoUtils {
-
- private static Schema schemaFor(Schema[][] schemas, int apiKey, int version) {
- if (apiKey < 0 || apiKey > schemas.length)
- throw new IllegalArgumentException("Invalid api key: " + apiKey);
- Schema[] versions = schemas[apiKey];
- if (version < 0 || version > latestVersion(apiKey))
- throw new IllegalArgumentException("Invalid version for API key " + apiKey + ": " + version);
- if (versions[version] == null)
- throw new IllegalArgumentException("Unsupported version for API key " + apiKey + ": " + version);
- return versions[version];
- }
-
- public static short latestVersion(int apiKey) {
- if (apiKey < 0 || apiKey >= Protocol.CURR_VERSION.length)
- throw new IllegalArgumentException("Invalid api key: " + apiKey);
- return Protocol.CURR_VERSION[apiKey];
- }
-
- public static short oldestVersion(int apiKey) {
- if (apiKey < 0 || apiKey >= Protocol.CURR_VERSION.length)
- throw new IllegalArgumentException("Invalid api key: " + apiKey);
- return Protocol.MIN_VERSIONS[apiKey];
- }
-
- public static Schema requestSchema(int apiKey, int version) {
- return schemaFor(Protocol.REQUESTS, apiKey, version);
- }
-
- public static Schema currentRequestSchema(int apiKey) {
- return requestSchema(apiKey, latestVersion(apiKey));
- }
-
- public static Schema responseSchema(int apiKey, int version) {
- return schemaFor(Protocol.RESPONSES, apiKey, version);
- }
-
- public static Schema currentResponseSchema(int apiKey) {
- return schemaFor(Protocol.RESPONSES, apiKey, latestVersion(apiKey));
- }
-
- public static Struct parseRequest(int apiKey, int version, ByteBuffer buffer) {
- return requestSchema(apiKey, version).read(buffer);
- }
-
- public static Struct parseResponse(int apiKey, int version, ByteBuffer buffer) {
- return responseSchema(apiKey, version).read(buffer);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java
index eea1916..e2403bc 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java
@@ -19,7 +19,6 @@ package org.apache.kafka.common.requests;
import org.apache.kafka.common.network.NetworkSend;
import org.apache.kafka.common.network.Send;
import org.apache.kafka.common.protocol.ApiKeys;
-import org.apache.kafka.common.protocol.ProtoUtils;
import org.apache.kafka.common.protocol.types.Struct;
import java.nio.ByteBuffer;
@@ -44,7 +43,7 @@ public abstract class AbstractRequest extends AbstractRequestResponse {
}
public short desiredOrLatestVersion() {
- return desiredVersion == null ? ProtoUtils.latestVersion(apiKey.id) : desiredVersion;
+ return desiredVersion == null ? apiKey.latestVersion() : desiredVersion;
}
public T build() {
@@ -90,7 +89,7 @@ public abstract class AbstractRequest extends AbstractRequestResponse {
*/
public static RequestAndSize getRequest(int requestId, short version, ByteBuffer buffer) {
ApiKeys apiKey = ApiKeys.forId(requestId);
- Struct struct = ProtoUtils.parseRequest(apiKey.id, version, buffer);
+ Struct struct = apiKey.parseRequest(version, buffer);
AbstractRequest request;
switch (apiKey) {
case PRODUCE:
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java
index a21e340..b018871 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java
@@ -47,8 +47,7 @@ public abstract class AbstractResponse extends AbstractRequestResponse {
protected abstract Struct toStruct(short version);
- public static AbstractResponse getResponse(int requestId, Struct struct) {
- ApiKeys apiKey = ApiKeys.forId(requestId);
+ public static AbstractResponse getResponse(ApiKeys apiKey, Struct struct) {
switch (apiKey) {
case PRODUCE:
return new ProduceResponse(struct);
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsRequest.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsRequest.java
index 7d40900..fde07bf 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsRequest.java
@@ -14,7 +14,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.Struct;
import java.nio.ByteBuffer;
@@ -48,7 +47,7 @@ public class ApiVersionsRequest extends AbstractRequest {
@Override
protected Struct toStruct() {
- return new Struct(ProtoUtils.requestSchema(ApiKeys.API_VERSIONS.id, version()));
+ return new Struct(ApiKeys.API_VERSIONS.requestSchema(version()));
}
@Override
@@ -59,12 +58,12 @@ public class ApiVersionsRequest extends AbstractRequest {
return new ApiVersionsResponse(Errors.forException(e), Collections.<ApiVersionsResponse.ApiVersion>emptyList());
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.API_VERSIONS.id)));
+ versionId, this.getClass().getSimpleName(), ApiKeys.API_VERSIONS.latestVersion()));
}
}
public static ApiVersionsRequest parse(ByteBuffer buffer, short version) {
- return new ApiVersionsRequest(ProtoUtils.parseRequest(ApiKeys.API_VERSIONS.id, version, buffer), version);
+ return new ApiVersionsRequest(ApiKeys.API_VERSIONS.parseRequest(version, buffer), version);
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java
index 0066c08..6fea979 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java
@@ -15,7 +15,6 @@ package org.apache.kafka.common.requests;
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.Struct;
import java.nio.ByteBuffer;
@@ -48,6 +47,10 @@ public class ApiVersionsResponse extends AbstractResponse {
public final short minVersion;
public final short maxVersion;
+ public ApiVersion(ApiKeys apiKey) {
+ this(apiKey.id, apiKey.oldestVersion(), apiKey.latestVersion());
+ }
+
public ApiVersion(short apiKey, short minVersion, short maxVersion) {
this.apiKey = apiKey;
this.minVersion = minVersion;
@@ -84,7 +87,7 @@ public class ApiVersionsResponse extends AbstractResponse {
@Override
protected Struct toStruct(short version) {
- Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.API_VERSIONS.id, version));
+ Struct struct = new Struct(ApiKeys.API_VERSIONS.responseSchema(version));
struct.set(ERROR_CODE_KEY_NAME, error.code());
List<Struct> apiVersionList = new ArrayList<>();
for (ApiVersion apiVersion : apiKeyToApiVersion.values()) {
@@ -120,13 +123,13 @@ public class ApiVersionsResponse extends AbstractResponse {
}
public static ApiVersionsResponse parse(ByteBuffer buffer, short version) {
- return new ApiVersionsResponse(ProtoUtils.responseSchema(ApiKeys.API_VERSIONS.id, version).read(buffer));
+ return new ApiVersionsResponse(ApiKeys.API_VERSIONS.responseSchema(version).read(buffer));
}
private static ApiVersionsResponse createApiVersionsResponse() {
List<ApiVersion> versionList = new ArrayList<>();
for (ApiKeys apiKey : ApiKeys.values()) {
- versionList.add(new ApiVersion(apiKey.id, ProtoUtils.oldestVersion(apiKey.id), ProtoUtils.latestVersion(apiKey.id)));
+ versionList.add(new ApiVersion(apiKey));
}
return new ApiVersionsResponse(Errors.NONE, versionList);
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/common/requests/ControlledShutdownRequest.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/ControlledShutdownRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/ControlledShutdownRequest.java
index 679e5dd..48f0a11 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/ControlledShutdownRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/ControlledShutdownRequest.java
@@ -15,7 +15,6 @@ package org.apache.kafka.common.requests;
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.Struct;
import java.nio.ByteBuffer;
@@ -69,7 +68,7 @@ public class ControlledShutdownRequest extends AbstractRequest {
return new ControlledShutdownResponse(Errors.forException(e), Collections.<TopicPartition>emptySet());
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.CONTROLLED_SHUTDOWN_KEY.id)));
+ versionId, this.getClass().getSimpleName(), ApiKeys.CONTROLLED_SHUTDOWN_KEY.latestVersion()));
}
}
@@ -77,14 +76,14 @@ public class ControlledShutdownRequest extends AbstractRequest {
return brokerId;
}
- public static ControlledShutdownRequest parse(ByteBuffer buffer, short versionId) {
+ public static ControlledShutdownRequest parse(ByteBuffer buffer, short version) {
return new ControlledShutdownRequest(
- ProtoUtils.parseRequest(ApiKeys.CONTROLLED_SHUTDOWN_KEY.id, versionId, buffer), versionId);
+ ApiKeys.CONTROLLED_SHUTDOWN_KEY.parseRequest(version, buffer), version);
}
@Override
protected Struct toStruct() {
- Struct struct = new Struct(ProtoUtils.requestSchema(ApiKeys.CONTROLLED_SHUTDOWN_KEY.id, version()));
+ Struct struct = new Struct(ApiKeys.CONTROLLED_SHUTDOWN_KEY.requestSchema(version()));
struct.set(BROKER_ID_KEY_NAME, brokerId);
return struct;
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/common/requests/ControlledShutdownResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/ControlledShutdownResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/ControlledShutdownResponse.java
index a2cac6c..a407c5a 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/ControlledShutdownResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/ControlledShutdownResponse.java
@@ -15,7 +15,6 @@ package org.apache.kafka.common.requests;
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.Struct;
import java.nio.ByteBuffer;
@@ -69,12 +68,12 @@ public class ControlledShutdownResponse extends AbstractResponse {
}
public static ControlledShutdownResponse parse(ByteBuffer buffer, short version) {
- return new ControlledShutdownResponse(ProtoUtils.parseResponse(ApiKeys.CONTROLLED_SHUTDOWN_KEY.id, version, buffer));
+ return new ControlledShutdownResponse(ApiKeys.CONTROLLED_SHUTDOWN_KEY.parseResponse(version, buffer));
}
@Override
protected Struct toStruct(short version) {
- Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.CONTROLLED_SHUTDOWN_KEY.id, version));
+ Struct struct = new Struct(ApiKeys.CONTROLLED_SHUTDOWN_KEY.responseSchema(version));
struct.set(ERROR_CODE_KEY_NAME, error.code());
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsRequest.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsRequest.java
index a8f8c5e..7e19687 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsRequest.java
@@ -19,7 +19,6 @@ package org.apache.kafka.common.requests;
import org.apache.kafka.common.errors.UnsupportedVersionException;
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.Struct;
import org.apache.kafka.common.utils.Utils;
@@ -227,7 +226,7 @@ public class CreateTopicsRequest extends AbstractRequest {
return new CreateTopicsResponse(topicErrors);
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.CREATE_TOPICS.id)));
+ versionId, this.getClass().getSimpleName(), ApiKeys.CREATE_TOPICS.latestVersion()));
}
}
@@ -247,8 +246,8 @@ public class CreateTopicsRequest extends AbstractRequest {
return this.duplicateTopics;
}
- public static CreateTopicsRequest parse(ByteBuffer buffer, short versionId) {
- return new CreateTopicsRequest(ProtoUtils.parseRequest(ApiKeys.CREATE_TOPICS.id, versionId, buffer), versionId);
+ public static CreateTopicsRequest parse(ByteBuffer buffer, short version) {
+ return new CreateTopicsRequest(ApiKeys.CREATE_TOPICS.parseRequest(version, buffer), version);
}
/**
@@ -257,7 +256,7 @@ public class CreateTopicsRequest extends AbstractRequest {
@Override
public Struct toStruct() {
short version = version();
- Struct struct = new Struct(ProtoUtils.requestSchema(ApiKeys.CREATE_TOPICS.id, version));
+ Struct struct = new Struct(ApiKeys.CREATE_TOPICS.requestSchema(version));
List<Struct> createTopicRequestStructs = new ArrayList<>(topics.size());
for (Map.Entry<String, TopicDetails> entry : topics.entrySet()) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsResponse.java
index 01b7c2b..f0986ff 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsResponse.java
@@ -19,7 +19,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.Struct;
import java.nio.ByteBuffer;
@@ -106,7 +105,7 @@ public class CreateTopicsResponse extends AbstractResponse {
@Override
protected Struct toStruct(short version) {
- Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.CREATE_TOPICS.id, version));
+ Struct struct = new Struct(ApiKeys.CREATE_TOPICS.responseSchema(version));
List<Struct> topicErrorsStructs = new ArrayList<>(errors.size());
for (Map.Entry<String, Error> topicError : errors.entrySet()) {
@@ -127,6 +126,6 @@ public class CreateTopicsResponse extends AbstractResponse {
}
public static CreateTopicsResponse parse(ByteBuffer buffer, short version) {
- return new CreateTopicsResponse(ProtoUtils.responseSchema(ApiKeys.CREATE_TOPICS.id, version).read(buffer));
+ return new CreateTopicsResponse(ApiKeys.CREATE_TOPICS.responseSchema(version).read(buffer));
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/common/requests/DeleteTopicsRequest.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/DeleteTopicsRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/DeleteTopicsRequest.java
index eea4aa9..1cb6eb8 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/DeleteTopicsRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/DeleteTopicsRequest.java
@@ -18,7 +18,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.Struct;
import org.apache.kafka.common.utils.Utils;
@@ -80,7 +79,7 @@ public class DeleteTopicsRequest extends AbstractRequest {
@Override
protected Struct toStruct() {
- Struct struct = new Struct(ProtoUtils.requestSchema(ApiKeys.DELETE_TOPICS.id, version()));
+ Struct struct = new Struct(ApiKeys.DELETE_TOPICS.requestSchema(version()));
struct.set(TOPICS_KEY_NAME, topics.toArray());
struct.set(TIMEOUT_KEY_NAME, timeout);
return struct;
@@ -97,7 +96,7 @@ public class DeleteTopicsRequest extends AbstractRequest {
return new DeleteTopicsResponse(topicErrors);
default:
throw new IllegalArgumentException(String.format("Version %d is not valid. Valid versions for %s are 0 to %d",
- version(), this.getClass().getSimpleName(), ProtoUtils.latestVersion(ApiKeys.DELETE_TOPICS.id)));
+ version(), this.getClass().getSimpleName(), ApiKeys.DELETE_TOPICS.latestVersion()));
}
}
@@ -109,8 +108,8 @@ public class DeleteTopicsRequest extends AbstractRequest {
return this.timeout;
}
- public static DeleteTopicsRequest parse(ByteBuffer buffer, short versionId) {
- return new DeleteTopicsRequest(ProtoUtils.parseRequest(ApiKeys.DELETE_TOPICS.id, versionId, buffer), versionId);
+ public static DeleteTopicsRequest parse(ByteBuffer buffer, short version) {
+ return new DeleteTopicsRequest(ApiKeys.DELETE_TOPICS.parseRequest(version, buffer), version);
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/common/requests/DeleteTopicsResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/DeleteTopicsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/DeleteTopicsResponse.java
index c47d098..dc651e1 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/DeleteTopicsResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/DeleteTopicsResponse.java
@@ -18,7 +18,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.Struct;
import java.nio.ByteBuffer;
@@ -61,7 +60,7 @@ public class DeleteTopicsResponse extends AbstractResponse {
@Override
protected Struct toStruct(short version) {
- Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.DELETE_TOPICS.id, version));
+ Struct struct = new Struct(ApiKeys.DELETE_TOPICS.responseSchema(version));
List<Struct> topicErrorCodeStructs = new ArrayList<>(errors.size());
for (Map.Entry<String, Errors> topicError : errors.entrySet()) {
Struct topicErrorCodeStruct = struct.instance(TOPIC_ERROR_CODES_KEY_NAME);
@@ -78,6 +77,6 @@ public class DeleteTopicsResponse extends AbstractResponse {
}
public static DeleteTopicsResponse parse(ByteBuffer buffer, short version) {
- return new DeleteTopicsResponse(ProtoUtils.responseSchema(ApiKeys.DELETE_TOPICS.id, version).read(buffer));
+ return new DeleteTopicsResponse(ApiKeys.DELETE_TOPICS.responseSchema(version).read(buffer));
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/common/requests/DescribeGroupsRequest.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/DescribeGroupsRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/DescribeGroupsRequest.java
index 0f13371..d5bde06 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/DescribeGroupsRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/DescribeGroupsRequest.java
@@ -14,7 +14,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.Struct;
import org.apache.kafka.common.utils.Utils;
@@ -64,25 +63,25 @@ public class DescribeGroupsRequest extends AbstractRequest {
@Override
protected Struct toStruct() {
- Struct struct = new Struct(ProtoUtils.requestSchema(ApiKeys.DESCRIBE_GROUPS.id, version()));
+ Struct struct = new Struct(ApiKeys.DESCRIBE_GROUPS.requestSchema(version()));
struct.set(GROUP_IDS_KEY_NAME, groupIds.toArray());
return struct;
}
@Override
public AbstractResponse getErrorResponse(Throwable e) {
- short versionId = version();
- switch (versionId) {
+ short version = version();
+ switch (version) {
case 0:
return DescribeGroupsResponse.fromError(Errors.forException(e), groupIds);
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.DESCRIBE_GROUPS.id)));
+ version, this.getClass().getSimpleName(), ApiKeys.DESCRIBE_GROUPS.latestVersion()));
}
}
- public static DescribeGroupsRequest parse(ByteBuffer buffer, short versionId) {
- return new DescribeGroupsRequest(ProtoUtils.parseRequest(ApiKeys.DESCRIBE_GROUPS.id, versionId, buffer), versionId);
+ public static DescribeGroupsRequest parse(ByteBuffer buffer, short version) {
+ return new DescribeGroupsRequest(ApiKeys.DESCRIBE_GROUPS.parseRequest(version, buffer), version);
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/common/requests/DescribeGroupsResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/DescribeGroupsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/DescribeGroupsResponse.java
index 0dde987..bab13ff 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/DescribeGroupsResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/DescribeGroupsResponse.java
@@ -14,7 +14,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.Struct;
import java.nio.ByteBuffer;
@@ -190,7 +189,7 @@ public class DescribeGroupsResponse extends AbstractResponse {
@Override
protected Struct toStruct(short version) {
- Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.DESCRIBE_GROUPS.id, version));
+ Struct struct = new Struct(ApiKeys.DESCRIBE_GROUPS.responseSchema(version));
List<Struct> groupStructs = new ArrayList<>();
for (Map.Entry<String, GroupMetadata> groupEntry : groups.entrySet()) {
@@ -219,7 +218,7 @@ public class DescribeGroupsResponse extends AbstractResponse {
return struct;
}
- public static DescribeGroupsResponse parse(ByteBuffer buffer, short versionId) {
- return new DescribeGroupsResponse(ProtoUtils.parseResponse(ApiKeys.DESCRIBE_GROUPS.id, versionId, buffer));
+ public static DescribeGroupsResponse parse(ByteBuffer buffer, short version) {
+ return new DescribeGroupsResponse(ApiKeys.DESCRIBE_GROUPS.parseResponse(version, buffer));
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java
index 55fd286..7418b89 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java
@@ -21,7 +21,6 @@ import java.util.Map;
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.Struct;
import org.apache.kafka.common.record.MemoryRecords;
import org.apache.kafka.common.utils.Utils;
@@ -219,14 +218,14 @@ public class FetchRequest extends AbstractRequest {
return replicaId >= 0;
}
- public static FetchRequest parse(ByteBuffer buffer, short versionId) {
- return new FetchRequest(ProtoUtils.parseRequest(ApiKeys.FETCH.id, versionId, buffer), versionId);
+ public static FetchRequest parse(ByteBuffer buffer, short version) {
+ return new FetchRequest(ApiKeys.FETCH.parseRequest(version, buffer), version);
}
@Override
protected Struct toStruct() {
short version = version();
- Struct struct = new Struct(ProtoUtils.requestSchema(ApiKeys.FETCH.id, version));
+ Struct struct = new Struct(ApiKeys.FETCH.requestSchema(version));
List<TopicAndPartitionData<PartitionData>> topicsData = TopicAndPartitionData.batchByTopic(fetchData);
struct.set(REPLICA_ID_KEY_NAME, replicaId);
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/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 dee6a5f..25631a5 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
@@ -22,7 +22,6 @@ import org.apache.kafka.common.network.MultiSend;
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.Struct;
import org.apache.kafka.common.protocol.types.Type;
@@ -158,7 +157,7 @@ public class FetchResponse extends AbstractResponse {
}
public static FetchResponse parse(ByteBuffer buffer, short version) {
- return new FetchResponse(ProtoUtils.responseSchema(ApiKeys.FETCH.id, version).read(buffer));
+ return new FetchResponse(ApiKeys.FETCH.responseSchema(version).read(buffer));
}
private static void addResponseData(Struct struct, int throttleTimeMs, String dest, List<Send> sends) {
@@ -212,7 +211,7 @@ public class FetchResponse extends AbstractResponse {
}
private static Struct toStruct(short version, LinkedHashMap<TopicPartition, PartitionData> responseData, int throttleTime) {
- Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.FETCH.id, version));
+ Struct struct = new Struct(ApiKeys.FETCH.responseSchema(version));
List<FetchRequest.TopicAndPartitionData<PartitionData>> topicsData = FetchRequest.TopicAndPartitionData.batchByTopic(responseData);
List<Struct> topicArray = new ArrayList<>();
for (FetchRequest.TopicAndPartitionData<PartitionData> topicEntry: topicsData) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/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 83d6cba..dcd7156 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
@@ -15,7 +15,6 @@ package org.apache.kafka.common.requests;
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.Struct;
import java.nio.ByteBuffer;
@@ -65,7 +64,7 @@ public class GroupCoordinatorRequest extends AbstractRequest {
return new GroupCoordinatorResponse(Errors.GROUP_COORDINATOR_NOT_AVAILABLE, Node.noNode());
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.GROUP_COORDINATOR.id)));
+ versionId, this.getClass().getSimpleName(), ApiKeys.GROUP_COORDINATOR.latestVersion()));
}
}
@@ -73,14 +72,13 @@ public class GroupCoordinatorRequest extends AbstractRequest {
return groupId;
}
- public static GroupCoordinatorRequest parse(ByteBuffer buffer, short versionId) {
- return new GroupCoordinatorRequest(ProtoUtils.parseRequest(ApiKeys.GROUP_COORDINATOR.id, versionId, buffer),
- versionId);
+ public static GroupCoordinatorRequest parse(ByteBuffer buffer, short version) {
+ return new GroupCoordinatorRequest(ApiKeys.GROUP_COORDINATOR.parseRequest(version, buffer), version);
}
@Override
protected Struct toStruct() {
- Struct struct = new Struct(ProtoUtils.requestSchema(ApiKeys.GROUP_COORDINATOR.id, version()));
+ Struct struct = new Struct(ApiKeys.GROUP_COORDINATOR.requestSchema(version()));
struct.set(GROUP_ID_KEY_NAME, groupId);
return struct;
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/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 c13cf3c..8a0f5cd 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
@@ -15,7 +15,6 @@ package org.apache.kafka.common.requests;
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.Struct;
import java.nio.ByteBuffer;
@@ -66,7 +65,7 @@ public class GroupCoordinatorResponse extends AbstractResponse {
@Override
protected Struct toStruct(short version) {
- Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.GROUP_COORDINATOR.id, version));
+ Struct struct = new Struct(ApiKeys.GROUP_COORDINATOR.responseSchema(version));
struct.set(ERROR_CODE_KEY_NAME, error.code());
Struct coordinator = struct.instance(COORDINATOR_KEY_NAME);
coordinator.set(NODE_ID_KEY_NAME, node.id());
@@ -77,6 +76,6 @@ public class GroupCoordinatorResponse extends AbstractResponse {
}
public static GroupCoordinatorResponse parse(ByteBuffer buffer, short version) {
- return new GroupCoordinatorResponse(ProtoUtils.parseResponse(ApiKeys.GROUP_COORDINATOR.id, version, buffer));
+ return new GroupCoordinatorResponse(ApiKeys.GROUP_COORDINATOR.parseResponse(version, buffer));
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/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 4440830..0b713de 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
@@ -14,7 +14,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.Struct;
import java.nio.ByteBuffer;
@@ -79,7 +78,7 @@ public class HeartbeatRequest extends AbstractRequest {
return new HeartbeatResponse(Errors.forException(e));
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.HEARTBEAT.id)));
+ versionId, this.getClass().getSimpleName(), ApiKeys.HEARTBEAT.latestVersion()));
}
}
@@ -95,13 +94,13 @@ public class HeartbeatRequest extends AbstractRequest {
return memberId;
}
- 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, short version) {
+ return new HeartbeatRequest(ApiKeys.HEARTBEAT.parseRequest(version, buffer), version);
}
@Override
protected Struct toStruct() {
- Struct struct = new Struct(ProtoUtils.requestSchema(ApiKeys.HEARTBEAT.id, version()));
+ Struct struct = new Struct(ApiKeys.HEARTBEAT.requestSchema(version()));
struct.set(GROUP_ID_KEY_NAME, groupId);
struct.set(GROUP_GENERATION_ID_KEY_NAME, groupGenerationId);
struct.set(MEMBER_ID_KEY_NAME, memberId);
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/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 4cca846..ac58a70 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
@@ -14,7 +14,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.Struct;
import java.nio.ByteBuffer;
@@ -49,12 +48,12 @@ public class HeartbeatResponse extends AbstractResponse {
@Override
protected Struct toStruct(short version) {
- Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.HEARTBEAT.id, version));
+ Struct struct = new Struct(ApiKeys.HEARTBEAT.responseSchema(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));
+ return new HeartbeatResponse(ApiKeys.HEARTBEAT.parseResponse(version, buffer));
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/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 37906a6..93f124a 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
@@ -14,7 +14,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.Struct;
import org.apache.kafka.common.utils.Utils;
@@ -161,7 +160,7 @@ public class JoinGroupRequest extends AbstractRequest {
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.JOIN_GROUP.id)));
+ versionId, this.getClass().getSimpleName(), ApiKeys.JOIN_GROUP.latestVersion()));
}
}
@@ -190,13 +189,13 @@ public class JoinGroupRequest extends AbstractRequest {
}
public static JoinGroupRequest parse(ByteBuffer buffer, short version) {
- return new JoinGroupRequest(ProtoUtils.parseRequest(ApiKeys.JOIN_GROUP.id, version, buffer), version);
+ return new JoinGroupRequest(ApiKeys.JOIN_GROUP.parseRequest(version, buffer), version);
}
@Override
protected Struct toStruct() {
short version = version();
- Struct struct = new Struct(ProtoUtils.requestSchema(ApiKeys.JOIN_GROUP.id, version));
+ Struct struct = new Struct(ApiKeys.JOIN_GROUP.requestSchema(version));
struct.set(GROUP_ID_KEY_NAME, groupId);
struct.set(SESSION_TIMEOUT_KEY_NAME, sessionTimeout);
if (version >= 1) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/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 d2a323b..82f6445 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
@@ -14,7 +14,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.Struct;
import java.nio.ByteBuffer;
@@ -117,12 +116,12 @@ public class JoinGroupResponse extends AbstractResponse {
}
public static JoinGroupResponse parse(ByteBuffer buffer, short version) {
- return new JoinGroupResponse(ProtoUtils.parseResponse(ApiKeys.JOIN_GROUP.id, version, buffer));
+ return new JoinGroupResponse(ApiKeys.JOIN_GROUP.parseResponse(version, buffer));
}
@Override
protected Struct toStruct(short version) {
- Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.JOIN_GROUP.id, version));
+ Struct struct = new Struct(ApiKeys.JOIN_GROUP.responseSchema(version));
struct.set(ERROR_CODE_KEY_NAME, error.code());
struct.set(GENERATION_ID_KEY_NAME, generationId);
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/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 c564b43..942590e 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
@@ -21,7 +21,6 @@ import org.apache.kafka.common.Node;
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.Struct;
import org.apache.kafka.common.utils.Utils;
@@ -147,7 +146,7 @@ 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 struct = new Struct(ApiKeys.LEADER_AND_ISR.requestSchema(version));
struct.set(CONTROLLER_ID_KEY_NAME, controllerId);
struct.set(CONTROLLER_EPOCH_KEY_NAME, controllerEpoch);
@@ -193,7 +192,7 @@ public class LeaderAndIsrRequest extends AbstractRequest {
return new LeaderAndIsrResponse(Errors.NONE, responses);
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.LEADER_AND_ISR.id)));
+ versionId, this.getClass().getSimpleName(), ApiKeys.LEADER_AND_ISR.latestVersion()));
}
}
@@ -213,8 +212,8 @@ public class LeaderAndIsrRequest extends AbstractRequest {
return liveLeaders;
}
- 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, short version) {
+ return new LeaderAndIsrRequest(ApiKeys.LEADER_AND_ISR.parseRequest(version, buffer), version);
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/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 2b02daf..3ec1d05 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
@@ -16,7 +16,6 @@ package org.apache.kafka.common.requests;
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.Struct;
import java.nio.ByteBuffer;
@@ -70,12 +69,12 @@ public class LeaderAndIsrResponse extends AbstractResponse {
}
public static LeaderAndIsrResponse parse(ByteBuffer buffer, short version) {
- return new LeaderAndIsrResponse(ProtoUtils.parseResponse(ApiKeys.LEADER_AND_ISR.id, version, buffer));
+ return new LeaderAndIsrResponse(ApiKeys.LEADER_AND_ISR.parseResponse(version, buffer));
}
@Override
protected Struct toStruct(short version) {
- Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.LEADER_AND_ISR.id, version));
+ Struct struct = new Struct(ApiKeys.LEADER_AND_ISR.responseSchema(version));
List<Struct> responseDatas = new ArrayList<>(responses.size());
for (Map.Entry<TopicPartition, Errors> response : responses.entrySet()) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/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 16622e4..f89675e 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
@@ -15,7 +15,6 @@ 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.Struct;
public class LeaveGroupRequest extends AbstractRequest {
@@ -71,7 +70,7 @@ public class LeaveGroupRequest extends AbstractRequest {
return new LeaveGroupResponse(Errors.forException(e));
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.LEAVE_GROUP.id)));
+ versionId, this.getClass().getSimpleName(), ApiKeys.LEAVE_GROUP.latestVersion()));
}
}
@@ -83,13 +82,13 @@ public class LeaveGroupRequest extends AbstractRequest {
return memberId;
}
- 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, short version) {
+ return new LeaveGroupRequest(ApiKeys.LEAVE_GROUP.parseRequest(version, buffer), version);
}
@Override
protected Struct toStruct() {
- Struct struct = new Struct(ProtoUtils.requestSchema(ApiKeys.LEAVE_GROUP.id, version()));
+ Struct struct = new Struct(ApiKeys.LEAVE_GROUP.requestSchema(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/015f1d73/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 a28816a..983027e 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
@@ -14,7 +14,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.Struct;
import java.nio.ByteBuffer;
@@ -48,13 +47,13 @@ public class LeaveGroupResponse extends AbstractResponse {
@Override
public Struct toStruct(short version) {
- Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.LEAVE_GROUP.id, version));
+ Struct struct = new Struct(ApiKeys.LEAVE_GROUP.responseSchema(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));
+ return new LeaveGroupResponse(ApiKeys.LEAVE_GROUP.parseResponse(versionId, buffer));
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/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 badb527..cca5399 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
@@ -14,7 +14,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.Struct;
import java.nio.ByteBuffer;
@@ -53,16 +52,16 @@ public class ListGroupsRequest extends AbstractRequest {
return new ListGroupsResponse(Errors.forException(e), Collections.<ListGroupsResponse.Group>emptyList());
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_GROUPS.id)));
+ versionId, this.getClass().getSimpleName(), ApiKeys.LIST_GROUPS.latestVersion()));
}
}
- 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, short version) {
+ return new ListGroupsRequest(ApiKeys.LIST_GROUPS.parseRequest(version, buffer), version);
}
@Override
protected Struct toStruct() {
- return new Struct(ProtoUtils.requestSchema(ApiKeys.LIST_GROUPS.id, version()));
+ return new Struct(ApiKeys.LIST_GROUPS.requestSchema(version()));
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/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 e05a4b1..ae54d33 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
@@ -14,7 +14,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.Struct;
import java.nio.ByteBuffer;
@@ -84,7 +83,7 @@ public class ListGroupsResponse extends AbstractResponse {
@Override
protected Struct toStruct(short version) {
- Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.LIST_GROUPS.id, version));
+ Struct struct = new Struct(ApiKeys.LIST_GROUPS.responseSchema(version));
struct.set(ERROR_CODE_KEY_NAME, error.code());
List<Struct> groupList = new ArrayList<>();
for (Group group : groups) {
@@ -101,8 +100,8 @@ public class ListGroupsResponse extends AbstractResponse {
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));
+ public static ListGroupsResponse parse(ByteBuffer buffer, short version) {
+ return new ListGroupsResponse(ApiKeys.LIST_GROUPS.parseResponse(version, buffer));
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/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 3e2ad7c..aea16de 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
@@ -20,7 +20,6 @@ import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.UnsupportedVersionException;
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.Struct;
import org.apache.kafka.common.utils.CollectionUtils;
import org.apache.kafka.common.utils.Utils;
@@ -68,7 +67,9 @@ public class ListOffsetRequest extends AbstractRequest {
return new Builder((short) 0, desiredVersion, replicaId);
}
- public static Builder forConsumer(short minVersion) {
+ public static Builder forConsumer(boolean requireTimestamp) {
+ // If we need a timestamp in the response, the minimum RPC version we can send is v1. Otherwise, v0 is OK.
+ short minVersion = requireTimestamp ? (short) 1 : (short) 0;
return new Builder(minVersion, null, CONSUMER_REPLICA_ID);
}
@@ -226,7 +227,7 @@ public class ListOffsetRequest extends AbstractRequest {
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)));
+ versionId, this.getClass().getSimpleName(), ApiKeys.LIST_OFFSETS.latestVersion()));
}
}
@@ -247,14 +248,14 @@ public class ListOffsetRequest extends AbstractRequest {
return duplicatePartitions;
}
- 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, short version) {
+ return new ListOffsetRequest(ApiKeys.LIST_OFFSETS.parseRequest(version, buffer), version);
}
@Override
protected Struct toStruct() {
short version = version();
- Struct struct = new Struct(ProtoUtils.requestSchema(ApiKeys.LIST_OFFSETS.id, version));
+ Struct struct = new Struct(ApiKeys.LIST_OFFSETS.requestSchema(version));
Map<TopicPartition, ?> targetTimes = partitionTimestamps == null ? offsetData : partitionTimestamps;
Map<String, Map<Integer, Object>> topicsData = CollectionUtils.groupDataByTopic(targetTimes);
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/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 cb3bafc..d4ab2d2 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
@@ -19,7 +19,6 @@ package org.apache.kafka.common.requests;
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.Struct;
import org.apache.kafka.common.utils.CollectionUtils;
import org.apache.kafka.common.utils.Utils;
@@ -146,12 +145,12 @@ public class ListOffsetResponse extends AbstractResponse {
}
public static ListOffsetResponse parse(ByteBuffer buffer, short version) {
- return new ListOffsetResponse(ProtoUtils.parseResponse(ApiKeys.LIST_OFFSETS.id, version, buffer));
+ return new ListOffsetResponse(ApiKeys.LIST_OFFSETS.parseResponse(version, buffer));
}
@Override
protected Struct toStruct(short version) {
- Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.LIST_OFFSETS.id, version));
+ Struct struct = new Struct(ApiKeys.LIST_OFFSETS.responseSchema(version));
Map<String, Map<Integer, PartitionData>> topicsData = CollectionUtils.groupDataByTopic(responseData);
List<Struct> topicArray = new ArrayList<>();
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/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 f31315f..db85c34 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
@@ -16,7 +16,6 @@ import org.apache.kafka.common.Node;
import org.apache.kafka.common.errors.UnsupportedVersionException;
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.Struct;
import org.apache.kafka.common.utils.Utils;
@@ -120,7 +119,7 @@ public class MetadataRequest extends AbstractRequest {
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)));
+ versionId, this.getClass().getSimpleName(), ApiKeys.METADATA.latestVersion()));
}
}
@@ -132,13 +131,13 @@ public class MetadataRequest extends AbstractRequest {
return topics;
}
- 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, short version) {
+ return new MetadataRequest(ApiKeys.METADATA.parseRequest(version, buffer), version);
}
@Override
protected Struct toStruct() {
- Struct struct = new Struct(ProtoUtils.requestSchema(ApiKeys.METADATA.id, version()));
+ Struct struct = new Struct(ApiKeys.METADATA.requestSchema(version()));
if (topics == null)
struct.set(TOPICS_KEY_NAME, null);
else
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/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 1532f67..fa79e92 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
@@ -17,7 +17,6 @@ import org.apache.kafka.common.Node;
import org.apache.kafka.common.PartitionInfo;
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.Struct;
import java.nio.ByteBuffer;
@@ -257,7 +256,7 @@ public class MetadataResponse extends AbstractResponse {
}
public static MetadataResponse parse(ByteBuffer buffer, short version) {
- return new MetadataResponse(ProtoUtils.parseResponse(ApiKeys.METADATA.id, version, buffer));
+ return new MetadataResponse(ApiKeys.METADATA.parseResponse(version, buffer));
}
public static class TopicMetadata {
@@ -337,7 +336,7 @@ public class MetadataResponse extends AbstractResponse {
@Override
protected Struct toStruct(short version) {
- Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.METADATA.id, version));
+ Struct struct = new Struct(ApiKeys.METADATA.responseSchema(version));
List<Struct> brokerArray = new ArrayList<>();
for (Node node : brokers) {
Struct broker = struct.instance(BROKERS_KEY_NAME);
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java
index bf14f10..556292c 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java
@@ -16,7 +16,6 @@ import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.UnsupportedVersionException;
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;
@@ -209,7 +208,7 @@ public class OffsetCommitRequest extends AbstractRequest {
@Override
public Struct toStruct() {
short version = version();
- Struct struct = new Struct(ProtoUtils.requestSchema(ApiKeys.OFFSET_COMMIT.id, version));
+ Struct struct = new Struct(ApiKeys.OFFSET_COMMIT.requestSchema(version));
struct.set(GROUP_ID_KEY_NAME, groupId);
Map<String, Map<Integer, PartitionData>> topicsData = CollectionUtils.groupDataByTopic(offsetData);
@@ -257,7 +256,7 @@ public class OffsetCommitRequest extends AbstractRequest {
return new OffsetCommitResponse(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.OFFSET_COMMIT.id)));
+ versionId, this.getClass().getSimpleName(), ApiKeys.OFFSET_COMMIT.latestVersion()));
}
}
@@ -281,8 +280,8 @@ public class OffsetCommitRequest extends AbstractRequest {
return offsetData;
}
- public static OffsetCommitRequest parse(ByteBuffer buffer, short versionId) {
- Schema schema = ProtoUtils.requestSchema(ApiKeys.OFFSET_COMMIT.id, versionId);
- return new OffsetCommitRequest(schema.read(buffer), versionId);
+ public static OffsetCommitRequest parse(ByteBuffer buffer, short version) {
+ Schema schema = ApiKeys.OFFSET_COMMIT.requestSchema(version);
+ return new OffsetCommitRequest(schema.read(buffer), version);
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitResponse.java
index b5709e2..8c0ff97 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitResponse.java
@@ -15,7 +15,6 @@ package org.apache.kafka.common.requests;
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.Struct;
import org.apache.kafka.common.utils.CollectionUtils;
@@ -75,7 +74,7 @@ public class OffsetCommitResponse extends AbstractResponse {
@Override
public Struct toStruct(short version) {
- Struct struct = new Struct(ProtoUtils.responseSchema(ApiKeys.OFFSET_COMMIT.id, version));
+ Struct struct = new Struct(ApiKeys.OFFSET_COMMIT.responseSchema(version));
Map<String, Map<Integer, Errors>> topicsData = CollectionUtils.groupDataByTopic(responseData);
List<Struct> topicArray = new ArrayList<>();
@@ -102,7 +101,7 @@ public class OffsetCommitResponse extends AbstractResponse {
}
public static OffsetCommitResponse parse(ByteBuffer buffer, short version) {
- return new OffsetCommitResponse(ProtoUtils.parseResponse(ApiKeys.OFFSET_COMMIT.id, version, buffer));
+ return new OffsetCommitResponse(ApiKeys.OFFSET_COMMIT.parseResponse(version, buffer));
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/015f1d73/clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java
index 2a550e5..653cbd8 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java
@@ -16,7 +16,6 @@ import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.UnsupportedVersionException;
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.Struct;
import org.apache.kafka.common.utils.CollectionUtils;
import org.apache.kafka.common.utils.Utils;
@@ -132,7 +131,7 @@ public class OffsetFetchRequest extends AbstractRequest {
return new OffsetFetchResponse(error, responsePartitions);
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.OFFSET_FETCH.id)));
+ versionId, this.getClass().getSimpleName(), ApiKeys.OFFSET_FETCH.latestVersion()));
}
}
@@ -149,8 +148,8 @@ public class OffsetFetchRequest extends AbstractRequest {
return partitions;
}
- public static OffsetFetchRequest parse(ByteBuffer buffer, short versionId) {
- return new OffsetFetchRequest(ProtoUtils.parseRequest(ApiKeys.OFFSET_FETCH.id, versionId, buffer), versionId);
+ public static OffsetFetchRequest parse(ByteBuffer buffer, short version) {
+ return new OffsetFetchRequest(ApiKeys.OFFSET_FETCH.parseRequest(version, buffer), version);
}
public boolean isAllPartitions() {
@@ -159,7 +158,7 @@ public class OffsetFetchRequest extends AbstractRequest {
@Override
protected Struct toStruct() {
- Struct struct = new Struct(ProtoUtils.requestSchema(ApiKeys.OFFSET_FETCH.id, version()));
+ Struct struct = new Struct(ApiKeys.OFFSET_FETCH.requestSchema(version()));
struct.set(GROUP_ID_KEY_NAME, groupId);
if (partitions != null) {
Map<String, List<Integer>> topicsData = CollectionUtils.groupDataByTopic(partitions);