kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From j...@apache.org
Subject kafka git commit: MINOR: Factor out some common group/transactional fields in request objects
Date Mon, 09 Oct 2017 20:22:49 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk 6d6080f13 -> f4ac68a40


MINOR: Factor out some common group/transactional fields in request objects

Author: Jason Gustafson <jason@confluent.io>

Reviewers: tedyu <yuzhihong@gmail.com>, Ismael Juma <ismael@juma.me.uk>

Closes #4047 from hachikuji/factor-out-some-common-fields


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/f4ac68a4
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/f4ac68a4
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/f4ac68a4

Branch: refs/heads/trunk
Commit: f4ac68a400a8289704a5888bf62f1d93e7aee99d
Parents: 6d6080f
Author: Jason Gustafson <jason@confluent.io>
Authored: Mon Oct 9 13:03:20 2017 -0700
Committer: Jason Gustafson <jason@confluent.io>
Committed: Mon Oct 9 13:21:04 2017 -0700

----------------------------------------------------------------------
 .../kafka/common/protocol/CommonFields.java     | 13 +++++-
 .../kafka/common/protocol/types/Struct.java     | 14 ++++++
 .../common/requests/AddOffsetsToTxnRequest.java | 37 +++++++---------
 .../requests/AddPartitionsToTxnRequest.java     | 27 ++++++------
 .../common/requests/DescribeGroupsResponse.java | 16 +++----
 .../kafka/common/requests/EndTxnRequest.java    | 27 ++++++------
 .../kafka/common/requests/HeartbeatRequest.java | 28 ++++++------
 .../common/requests/InitProducerIdResponse.java | 22 ++++------
 .../kafka/common/requests/JoinGroupRequest.java | 20 ++++-----
 .../common/requests/JoinGroupResponse.java      | 27 ++++++------
 .../common/requests/LeaveGroupRequest.java      | 19 ++++-----
 .../common/requests/ListGroupsResponse.java     |  8 ++--
 .../common/requests/OffsetCommitRequest.java    | 45 ++++++++------------
 .../common/requests/OffsetFetchRequest.java     | 11 +++--
 .../kafka/common/requests/SyncGroupRequest.java | 32 +++++++-------
 .../common/requests/TxnOffsetCommitRequest.java | 34 +++++++--------
 16 files changed, 182 insertions(+), 198 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/f4ac68a4/clients/src/main/java/org/apache/kafka/common/protocol/CommonFields.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/CommonFields.java b/clients/src/main/java/org/apache/kafka/common/protocol/CommonFields.java
index e12cde4..472a791 100644
--- a/clients/src/main/java/org/apache/kafka/common/protocol/CommonFields.java
+++ b/clients/src/main/java/org/apache/kafka/common/protocol/CommonFields.java
@@ -27,7 +27,18 @@ public class CommonFields {
     public static final Field.Int16 ERROR_CODE = new Field.Int16("error_code", "Response error code");
     public static final Field.NullableStr ERROR_MESSAGE = new Field.NullableStr("error_message", "Response error message");
 
-    // ACL Apis
+    // Group APIs
+    public static final Field.Str GROUP_ID = new Field.Str("group_id", "The unique group identifier");
+    public static final Field.Int32 GENERATION_ID = new Field.Int32("generation_id", "The generation of the group.");
+    public static final Field.Str MEMBER_ID = new Field.Str("member_id", "The member id assigned by the group " +
+            "coordinator or null if joining for the first time.");
+
+    // Transactional APIs
+    public static final Field.Str TRANSACTIONAL_ID = new Field.Str("transactional_id", "The transactional id corresponding to the transaction.");
+    public static final Field.Int64 PRODUCER_ID = new Field.Int64("producer_id", "Current producer id in use by the transactional id.");
+    public static final Field.Int16 PRODUCER_EPOCH = new Field.Int16("producer_epoch", "Current epoch associated with the producer id.");
+
+    // ACL APIs
     public static final Field.Int8 RESOURCE_TYPE = new Field.Int8("resource_type", "The resource type");
     public static final Field.Str RESOURCE_NAME = new Field.Str("resource_name", "The resource name");
     public static final Field.NullableStr RESOURCE_NAME_FILTER = new Field.NullableStr("resource_name", "The resource name filter");

http://git-wip-us.apache.org/repos/asf/kafka/blob/f4ac68a4/clients/src/main/java/org/apache/kafka/common/protocol/types/Struct.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/types/Struct.java b/clients/src/main/java/org/apache/kafka/common/protocol/types/Struct.java
index 1cbbcb3..b825201 100644
--- a/clients/src/main/java/org/apache/kafka/common/protocol/types/Struct.java
+++ b/clients/src/main/java/org/apache/kafka/common/protocol/types/Struct.java
@@ -83,6 +83,10 @@ public class Struct {
         return getInt(field.name);
     }
 
+    public Long get(Field.Int64 field) {
+        return getLong(field.name);
+    }
+
     public Short get(Field.Int16 field) {
         return getShort(field.name);
     }
@@ -113,6 +117,12 @@ public class Struct {
         return alternative;
     }
 
+    public String getOrElse(Field.Str field, String alternative) {
+        if (hasField(field.name))
+            return getString(field.name);
+        return alternative;
+    }
+
     /**
      * Get the record value for the field with the given name by doing a hash table lookup (slower!)
      *
@@ -270,6 +280,10 @@ public class Struct {
         return set(def.name, value);
     }
 
+    public Struct set(Field.Int64 def, long value) {
+        return set(def.name, value);
+    }
+
     public Struct set(Field.Int16 def, short value) {
         return set(def.name, value);
     }

http://git-wip-us.apache.org/repos/asf/kafka/blob/f4ac68a4/clients/src/main/java/org/apache/kafka/common/requests/AddOffsetsToTxnRequest.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AddOffsetsToTxnRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/AddOffsetsToTxnRequest.java
index e3e4d79..f6a1722 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/AddOffsetsToTxnRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/AddOffsetsToTxnRequest.java
@@ -18,27 +18,22 @@ 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.types.Field;
 import org.apache.kafka.common.protocol.types.Schema;
 import org.apache.kafka.common.protocol.types.Struct;
 
 import java.nio.ByteBuffer;
 
-import static org.apache.kafka.common.protocol.types.Type.INT16;
-import static org.apache.kafka.common.protocol.types.Type.INT64;
-import static org.apache.kafka.common.protocol.types.Type.STRING;
+import static org.apache.kafka.common.protocol.CommonFields.GROUP_ID;
+import static org.apache.kafka.common.protocol.CommonFields.PRODUCER_EPOCH;
+import static org.apache.kafka.common.protocol.CommonFields.PRODUCER_ID;
+import static org.apache.kafka.common.protocol.CommonFields.TRANSACTIONAL_ID;
 
 public class AddOffsetsToTxnRequest extends AbstractRequest {
-    private static final String TRANSACTIONAL_ID_KEY_NAME = "transactional_id";
-    private static final String PRODUCER_ID_KEY_NAME = "producer_id";
-    private static final String EPOCH_KEY_NAME = "producer_epoch";
-    private static final String CONSUMER_GROUP_ID_KEY_NAME = "consumer_group_id";
-
     private static final Schema ADD_OFFSETS_TO_TXN_REQUEST_V0 = new Schema(
-            new Field(TRANSACTIONAL_ID_KEY_NAME, STRING, "The transactional id corresponding to the transaction."),
-            new Field(PRODUCER_ID_KEY_NAME, INT64, "Current producer id in use by the transactional id."),
-            new Field(EPOCH_KEY_NAME, INT16, "Current epoch associated with the producer id."),
-            new Field(CONSUMER_GROUP_ID_KEY_NAME, STRING, "Consumer group id whose offsets should be included in the transaction."));
+            TRANSACTIONAL_ID,
+            PRODUCER_ID,
+            PRODUCER_EPOCH,
+            GROUP_ID);
 
     public static Schema[] schemaVersions() {
         return new Schema[]{ADD_OFFSETS_TO_TXN_REQUEST_V0};
@@ -95,10 +90,10 @@ public class AddOffsetsToTxnRequest extends AbstractRequest {
 
     public AddOffsetsToTxnRequest(Struct struct, short version) {
         super(version);
-        this.transactionalId = struct.getString(TRANSACTIONAL_ID_KEY_NAME);
-        this.producerId = struct.getLong(PRODUCER_ID_KEY_NAME);
-        this.producerEpoch = struct.getShort(EPOCH_KEY_NAME);
-        this.consumerGroupId = struct.getString(CONSUMER_GROUP_ID_KEY_NAME);
+        this.transactionalId = struct.get(TRANSACTIONAL_ID);
+        this.producerId = struct.get(PRODUCER_ID);
+        this.producerEpoch = struct.get(PRODUCER_EPOCH);
+        this.consumerGroupId = struct.get(GROUP_ID);
     }
 
     public String transactionalId() {
@@ -120,10 +115,10 @@ public class AddOffsetsToTxnRequest extends AbstractRequest {
     @Override
     protected Struct toStruct() {
         Struct struct = new Struct(ApiKeys.ADD_OFFSETS_TO_TXN.requestSchema(version()));
-        struct.set(TRANSACTIONAL_ID_KEY_NAME, transactionalId);
-        struct.set(PRODUCER_ID_KEY_NAME, producerId);
-        struct.set(EPOCH_KEY_NAME, producerEpoch);
-        struct.set(CONSUMER_GROUP_ID_KEY_NAME, consumerGroupId);
+        struct.set(TRANSACTIONAL_ID, transactionalId);
+        struct.set(PRODUCER_ID, producerId);
+        struct.set(PRODUCER_EPOCH, producerEpoch);
+        struct.set(GROUP_ID, consumerGroupId);
         return struct;
     }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/f4ac68a4/clients/src/main/java/org/apache/kafka/common/requests/AddPartitionsToTxnRequest.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AddPartitionsToTxnRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/AddPartitionsToTxnRequest.java
index c195e24..0ca32be 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/AddPartitionsToTxnRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/AddPartitionsToTxnRequest.java
@@ -31,23 +31,20 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import static org.apache.kafka.common.protocol.CommonFields.PRODUCER_EPOCH;
+import static org.apache.kafka.common.protocol.CommonFields.PRODUCER_ID;
 import static org.apache.kafka.common.protocol.CommonFields.TOPIC_NAME;
-import static org.apache.kafka.common.protocol.types.Type.INT16;
+import static org.apache.kafka.common.protocol.CommonFields.TRANSACTIONAL_ID;
 import static org.apache.kafka.common.protocol.types.Type.INT32;
-import static org.apache.kafka.common.protocol.types.Type.INT64;
-import static org.apache.kafka.common.protocol.types.Type.STRING;
 
 public class AddPartitionsToTxnRequest extends AbstractRequest {
-    private static final String TRANSACTIONAL_ID_KEY_NAME = "transactional_id";
-    private static final String PRODUCER_ID_KEY_NAME = "producer_id";
-    private static final String PRODUCER_EPOCH_KEY_NAME = "producer_epoch";
     private static final String TOPICS_KEY_NAME = "topics";
     private static final String PARTITIONS_KEY_NAME = "partitions";
 
     private static final Schema ADD_PARTITIONS_TO_TXN_REQUEST_V0 = new Schema(
-            new Field(TRANSACTIONAL_ID_KEY_NAME, STRING, "The transactional id corresponding to the transaction."),
-            new Field(PRODUCER_ID_KEY_NAME, INT64, "Current producer id in use by the transactional id."),
-            new Field(PRODUCER_EPOCH_KEY_NAME, INT16, "Current epoch associated with the producer id."),
+            TRANSACTIONAL_ID,
+            PRODUCER_ID,
+            PRODUCER_EPOCH,
             new Field(TOPICS_KEY_NAME, new ArrayOf(new Schema(
                     TOPIC_NAME,
                     new Field(PARTITIONS_KEY_NAME, new ArrayOf(INT32)))),
@@ -109,9 +106,9 @@ public class AddPartitionsToTxnRequest extends AbstractRequest {
 
     public AddPartitionsToTxnRequest(Struct struct, short version) {
         super(version);
-        this.transactionalId = struct.getString(TRANSACTIONAL_ID_KEY_NAME);
-        this.producerId = struct.getLong(PRODUCER_ID_KEY_NAME);
-        this.producerEpoch = struct.getShort(PRODUCER_EPOCH_KEY_NAME);
+        this.transactionalId = struct.get(TRANSACTIONAL_ID);
+        this.producerId = struct.get(PRODUCER_ID);
+        this.producerEpoch = struct.get(PRODUCER_EPOCH);
 
         List<TopicPartition> partitions = new ArrayList<>();
         Object[] topicPartitionsArray = struct.getArray(TOPICS_KEY_NAME);
@@ -144,9 +141,9 @@ public class AddPartitionsToTxnRequest extends AbstractRequest {
     @Override
     protected Struct toStruct() {
         Struct struct = new Struct(ApiKeys.ADD_PARTITIONS_TO_TXN.requestSchema(version()));
-        struct.set(TRANSACTIONAL_ID_KEY_NAME, transactionalId);
-        struct.set(PRODUCER_ID_KEY_NAME, producerId);
-        struct.set(PRODUCER_EPOCH_KEY_NAME, producerEpoch);
+        struct.set(TRANSACTIONAL_ID, transactionalId);
+        struct.set(PRODUCER_ID, producerId);
+        struct.set(PRODUCER_EPOCH, producerEpoch);
 
         Map<String, List<Integer>> mappedPartitions = CollectionUtils.groupDataByTopic(partitions);
         Object[] partitionsArray = new Object[mappedPartitions.size()];

http://git-wip-us.apache.org/repos/asf/kafka/blob/f4ac68a4/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 313e113..174b14b 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
@@ -31,6 +31,8 @@ import java.util.List;
 import java.util.Map;
 
 import static org.apache.kafka.common.protocol.CommonFields.ERROR_CODE;
+import static org.apache.kafka.common.protocol.CommonFields.GROUP_ID;
+import static org.apache.kafka.common.protocol.CommonFields.MEMBER_ID;
 import static org.apache.kafka.common.protocol.CommonFields.THROTTLE_TIME_MS;
 import static org.apache.kafka.common.protocol.types.Type.BYTES;
 import static org.apache.kafka.common.protocol.types.Type.STRING;
@@ -39,20 +41,18 @@ public class DescribeGroupsResponse extends AbstractResponse {
 
     private static final String GROUPS_KEY_NAME = "groups";
 
-    private static final String GROUP_ID_KEY_NAME = "group_id";
     private static final String GROUP_STATE_KEY_NAME = "state";
     private static final String PROTOCOL_TYPE_KEY_NAME = "protocol_type";
     private static final String PROTOCOL_KEY_NAME = "protocol";
 
     private static final String MEMBERS_KEY_NAME = "members";
-    private static final String MEMBER_ID_KEY_NAME = "member_id";
     private static final String CLIENT_ID_KEY_NAME = "client_id";
     private static final String CLIENT_HOST_KEY_NAME = "client_host";
     private static final String MEMBER_METADATA_KEY_NAME = "member_metadata";
     private static final String MEMBER_ASSIGNMENT_KEY_NAME = "member_assignment";
 
     private static final Schema DESCRIBE_GROUPS_RESPONSE_MEMBER_V0 = new Schema(
-            new Field(MEMBER_ID_KEY_NAME, STRING, "The memberId assigned by the coordinator"),
+            MEMBER_ID,
             new Field(CLIENT_ID_KEY_NAME, STRING, "The client id used in the member's latest join group request"),
             new Field(CLIENT_HOST_KEY_NAME, STRING, "The client host used in the request session corresponding to the " +
                     "member's join group."),
@@ -63,7 +63,7 @@ public class DescribeGroupsResponse extends AbstractResponse {
 
     private static final Schema DESCRIBE_GROUPS_RESPONSE_GROUP_METADATA_V0 = new Schema(
             ERROR_CODE,
-            new Field(GROUP_ID_KEY_NAME, STRING),
+            GROUP_ID,
             new Field(GROUP_STATE_KEY_NAME, STRING, "The current state of the group (one of: Dead, Stable, CompletingRebalance, " +
                     "PreparingRebalance, or empty if there is no active group)"),
             new Field(PROTOCOL_TYPE_KEY_NAME, STRING, "The current group protocol type (will be empty if there is no active group)"),
@@ -112,7 +112,7 @@ public class DescribeGroupsResponse extends AbstractResponse {
         for (Object groupObj : struct.getArray(GROUPS_KEY_NAME)) {
             Struct groupStruct = (Struct) groupObj;
 
-            String groupId = groupStruct.getString(GROUP_ID_KEY_NAME);
+            String groupId = groupStruct.get(GROUP_ID);
             Errors error = Errors.forCode(groupStruct.get(ERROR_CODE));
             String state = groupStruct.getString(GROUP_STATE_KEY_NAME);
             String protocolType = groupStruct.getString(PROTOCOL_TYPE_KEY_NAME);
@@ -121,7 +121,7 @@ public class DescribeGroupsResponse extends AbstractResponse {
             List<GroupMember> members = new ArrayList<>();
             for (Object memberObj : groupStruct.getArray(MEMBERS_KEY_NAME)) {
                 Struct memberStruct = (Struct) memberObj;
-                String memberId = memberStruct.getString(MEMBER_ID_KEY_NAME);
+                String memberId = memberStruct.get(MEMBER_ID);
                 String clientId = memberStruct.getString(CLIENT_ID_KEY_NAME);
                 String clientHost = memberStruct.getString(CLIENT_HOST_KEY_NAME);
                 ByteBuffer memberMetadata = memberStruct.getBytes(MEMBER_METADATA_KEY_NAME);
@@ -259,7 +259,7 @@ public class DescribeGroupsResponse extends AbstractResponse {
         for (Map.Entry<String, GroupMetadata> groupEntry : groups.entrySet()) {
             Struct groupStruct = struct.instance(GROUPS_KEY_NAME);
             GroupMetadata group = groupEntry.getValue();
-            groupStruct.set(GROUP_ID_KEY_NAME, groupEntry.getKey());
+            groupStruct.set(GROUP_ID, groupEntry.getKey());
             groupStruct.set(ERROR_CODE, group.error.code());
             groupStruct.set(GROUP_STATE_KEY_NAME, group.state);
             groupStruct.set(PROTOCOL_TYPE_KEY_NAME, group.protocolType);
@@ -267,7 +267,7 @@ public class DescribeGroupsResponse extends AbstractResponse {
             List<Struct> membersList = new ArrayList<>();
             for (GroupMember member : group.members) {
                 Struct memberStruct = groupStruct.instance(MEMBERS_KEY_NAME);
-                memberStruct.set(MEMBER_ID_KEY_NAME, member.memberId);
+                memberStruct.set(MEMBER_ID, member.memberId);
                 memberStruct.set(CLIENT_ID_KEY_NAME, member.clientId);
                 memberStruct.set(CLIENT_HOST_KEY_NAME, member.clientHost);
                 memberStruct.set(MEMBER_METADATA_KEY_NAME, member.memberMetadata);

http://git-wip-us.apache.org/repos/asf/kafka/blob/f4ac68a4/clients/src/main/java/org/apache/kafka/common/requests/EndTxnRequest.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/EndTxnRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/EndTxnRequest.java
index 243e9f5..9118d6a 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/EndTxnRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/EndTxnRequest.java
@@ -24,21 +24,18 @@ import org.apache.kafka.common.protocol.types.Struct;
 
 import java.nio.ByteBuffer;
 
+import static org.apache.kafka.common.protocol.CommonFields.PRODUCER_EPOCH;
+import static org.apache.kafka.common.protocol.CommonFields.PRODUCER_ID;
+import static org.apache.kafka.common.protocol.CommonFields.TRANSACTIONAL_ID;
 import static org.apache.kafka.common.protocol.types.Type.BOOLEAN;
-import static org.apache.kafka.common.protocol.types.Type.INT16;
-import static org.apache.kafka.common.protocol.types.Type.INT64;
-import static org.apache.kafka.common.protocol.types.Type.STRING;
 
 public class EndTxnRequest extends AbstractRequest {
-    private static final String TRANSACTIONAL_ID_KEY_NAME = "transactional_id";
-    private static final String PRODUCER_ID_KEY_NAME = "producer_id";
-    private static final String PRODUCER_EPOCH_KEY_NAME = "producer_epoch";
     private static final String TRANSACTION_RESULT_KEY_NAME = "transaction_result";
 
     private static final Schema END_TXN_REQUEST_V0 = new Schema(
-            new Field(TRANSACTIONAL_ID_KEY_NAME, STRING, "The transactional id corresponding to the transaction."),
-            new Field(PRODUCER_ID_KEY_NAME, INT64, "Current producer id in use by the transactional id."),
-            new Field(PRODUCER_EPOCH_KEY_NAME, INT16, "Current epoch associated with the producer id."),
+            TRANSACTIONAL_ID,
+            PRODUCER_ID,
+            PRODUCER_EPOCH,
             new Field(TRANSACTION_RESULT_KEY_NAME, BOOLEAN, "The result of the transaction (0 = ABORT, 1 = COMMIT)"));
 
     public static Schema[] schemaVersions() {
@@ -96,9 +93,9 @@ public class EndTxnRequest extends AbstractRequest {
 
     public EndTxnRequest(Struct struct, short version) {
         super(version);
-        this.transactionalId = struct.getString(TRANSACTIONAL_ID_KEY_NAME);
-        this.producerId = struct.getLong(PRODUCER_ID_KEY_NAME);
-        this.producerEpoch = struct.getShort(PRODUCER_EPOCH_KEY_NAME);
+        this.transactionalId = struct.get(TRANSACTIONAL_ID);
+        this.producerId = struct.get(PRODUCER_ID);
+        this.producerEpoch = struct.get(PRODUCER_EPOCH);
         this.result = TransactionResult.forId(struct.getBoolean(TRANSACTION_RESULT_KEY_NAME));
     }
 
@@ -121,9 +118,9 @@ public class EndTxnRequest extends AbstractRequest {
     @Override
     protected Struct toStruct() {
         Struct struct = new Struct(ApiKeys.END_TXN.requestSchema(version()));
-        struct.set(TRANSACTIONAL_ID_KEY_NAME, transactionalId);
-        struct.set(PRODUCER_ID_KEY_NAME, producerId);
-        struct.set(PRODUCER_EPOCH_KEY_NAME, producerEpoch);
+        struct.set(TRANSACTIONAL_ID, transactionalId);
+        struct.set(PRODUCER_ID, producerId);
+        struct.set(PRODUCER_EPOCH, producerEpoch);
         struct.set(TRANSACTION_RESULT_KEY_NAME, result.id);
         return struct;
     }

http://git-wip-us.apache.org/repos/asf/kafka/blob/f4ac68a4/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 00a806f..7d84918 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
@@ -18,24 +18,20 @@ 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.types.Field;
 import org.apache.kafka.common.protocol.types.Schema;
 import org.apache.kafka.common.protocol.types.Struct;
 
 import java.nio.ByteBuffer;
 
-import static org.apache.kafka.common.protocol.types.Type.INT32;
-import static org.apache.kafka.common.protocol.types.Type.STRING;
+import static org.apache.kafka.common.protocol.CommonFields.GENERATION_ID;
+import static org.apache.kafka.common.protocol.CommonFields.GROUP_ID;
+import static org.apache.kafka.common.protocol.CommonFields.MEMBER_ID;
 
 public class HeartbeatRequest extends AbstractRequest {
-    private static final String GROUP_ID_KEY_NAME = "group_id";
-    private static final String GROUP_GENERATION_ID_KEY_NAME = "group_generation_id";
-    private static final String MEMBER_ID_KEY_NAME = "member_id";
-
     private static final Schema HEARTBEAT_REQUEST_V0 = new Schema(
-            new Field(GROUP_ID_KEY_NAME, STRING, "The group id."),
-            new Field(GROUP_GENERATION_ID_KEY_NAME, INT32, "The generation of the group."),
-            new Field(MEMBER_ID_KEY_NAME, STRING, "The member id assigned by the group coordinator."));
+            GROUP_ID,
+            GENERATION_ID,
+            MEMBER_ID);
 
     /* v1 request is the same as v0. Throttle time has been added to response */
     private static final Schema HEARTBEAT_REQUEST_V1 = HEARTBEAT_REQUEST_V0;
@@ -86,9 +82,9 @@ public class HeartbeatRequest extends AbstractRequest {
 
     public HeartbeatRequest(Struct struct, short version) {
         super(version);
-        groupId = struct.getString(GROUP_ID_KEY_NAME);
-        groupGenerationId = struct.getInt(GROUP_GENERATION_ID_KEY_NAME);
-        memberId = struct.getString(MEMBER_ID_KEY_NAME);
+        groupId = struct.get(GROUP_ID);
+        groupGenerationId = struct.get(GENERATION_ID);
+        memberId = struct.get(MEMBER_ID);
     }
 
     @Override
@@ -124,9 +120,9 @@ public class HeartbeatRequest extends AbstractRequest {
     @Override
     protected Struct toStruct() {
         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);
+        struct.set(GROUP_ID, groupId);
+        struct.set(GENERATION_ID, groupGenerationId);
+        struct.set(MEMBER_ID, memberId);
         return struct;
     }
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/f4ac68a4/clients/src/main/java/org/apache/kafka/common/requests/InitProducerIdResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/InitProducerIdResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/InitProducerIdResponse.java
index 9ecb21f..7a988ca 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/InitProducerIdResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/InitProducerIdResponse.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.types.Field;
 import org.apache.kafka.common.protocol.types.Schema;
 import org.apache.kafka.common.protocol.types.Struct;
 import org.apache.kafka.common.record.RecordBatch;
@@ -27,9 +26,9 @@ import java.nio.ByteBuffer;
 import java.util.Map;
 
 import static org.apache.kafka.common.protocol.CommonFields.ERROR_CODE;
+import static org.apache.kafka.common.protocol.CommonFields.PRODUCER_EPOCH;
+import static org.apache.kafka.common.protocol.CommonFields.PRODUCER_ID;
 import static org.apache.kafka.common.protocol.CommonFields.THROTTLE_TIME_MS;
-import static org.apache.kafka.common.protocol.types.Type.INT16;
-import static org.apache.kafka.common.protocol.types.Type.INT64;
 
 public class InitProducerIdResponse extends AbstractResponse {
     // Possible error codes:
@@ -39,16 +38,11 @@ public class InitProducerIdResponse extends AbstractResponse {
     //   TransactionalIdAuthorizationFailed
     //   ClusterAuthorizationFailed
 
-    private static final String PRODUCER_ID_KEY_NAME = "producer_id";
-    private static final String EPOCH_KEY_NAME = "producer_epoch";
-
     private static final Schema INIT_PRODUCER_ID_RESPONSE_V0 = new Schema(
             THROTTLE_TIME_MS,
             ERROR_CODE,
-            new Field(PRODUCER_ID_KEY_NAME, INT64, "The producer id for the input transactional id. If the input " +
-                    "id was empty, then this is used only for ensuring idempotence of messages."),
-            new Field(EPOCH_KEY_NAME, INT16, "The epoch for the producer id. Will always be 0 if no transactional " +
-                    "id was specified in the request."));
+            PRODUCER_ID,
+            PRODUCER_EPOCH);
 
     public static Schema[] schemaVersions() {
         return new Schema[]{INIT_PRODUCER_ID_RESPONSE_V0};
@@ -69,8 +63,8 @@ public class InitProducerIdResponse extends AbstractResponse {
     public InitProducerIdResponse(Struct struct) {
         this.throttleTimeMs = struct.get(THROTTLE_TIME_MS);
         this.error = Errors.forCode(struct.get(ERROR_CODE));
-        this.producerId = struct.getLong(PRODUCER_ID_KEY_NAME);
-        this.epoch = struct.getShort(EPOCH_KEY_NAME);
+        this.producerId = struct.get(PRODUCER_ID);
+        this.epoch = struct.get(PRODUCER_EPOCH);
     }
 
     public InitProducerIdResponse(int throttleTimeMs, Errors errors) {
@@ -102,8 +96,8 @@ public class InitProducerIdResponse extends AbstractResponse {
     protected Struct toStruct(short version) {
         Struct struct = new Struct(ApiKeys.INIT_PRODUCER_ID.responseSchema(version));
         struct.set(THROTTLE_TIME_MS, throttleTimeMs);
-        struct.set(PRODUCER_ID_KEY_NAME, producerId);
-        struct.set(EPOCH_KEY_NAME, epoch);
+        struct.set(PRODUCER_ID, producerId);
+        struct.set(PRODUCER_EPOCH, epoch);
         struct.set(ERROR_CODE, error.code());
         return struct;
     }

http://git-wip-us.apache.org/repos/asf/kafka/blob/f4ac68a4/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 b2ff133..a7b62a9 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
@@ -29,15 +29,15 @@ import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 
+import static org.apache.kafka.common.protocol.CommonFields.GROUP_ID;
+import static org.apache.kafka.common.protocol.CommonFields.MEMBER_ID;
 import static org.apache.kafka.common.protocol.types.Type.BYTES;
 import static org.apache.kafka.common.protocol.types.Type.INT32;
 import static org.apache.kafka.common.protocol.types.Type.STRING;
 
 public class JoinGroupRequest extends AbstractRequest {
-    private static final String GROUP_ID_KEY_NAME = "group_id";
     private static final String SESSION_TIMEOUT_KEY_NAME = "session_timeout";
     private static final String REBALANCE_TIMEOUT_KEY_NAME = "rebalance_timeout";
-    private static final String MEMBER_ID_KEY_NAME = "member_id";
     private static final String PROTOCOL_TYPE_KEY_NAME = "protocol_type";
     private static final String GROUP_PROTOCOLS_KEY_NAME = "group_protocols";
     private static final String PROTOCOL_NAME_KEY_NAME = "protocol_name";
@@ -49,21 +49,21 @@ public class JoinGroupRequest extends AbstractRequest {
             new Field(PROTOCOL_METADATA_KEY_NAME, BYTES));
 
     private static final Schema JOIN_GROUP_REQUEST_V0 = new Schema(
-            new Field(GROUP_ID_KEY_NAME, STRING, "The group id."),
+            GROUP_ID,
             new Field(SESSION_TIMEOUT_KEY_NAME, INT32, "The coordinator considers the consumer dead if it receives " +
                     "no heartbeat after this timeout in ms."),
-            new Field(MEMBER_ID_KEY_NAME, STRING, "The assigned consumer id or an empty string for a new consumer."),
+            MEMBER_ID,
             new Field(PROTOCOL_TYPE_KEY_NAME, STRING, "Unique name for class of protocols implemented by group"),
             new Field(GROUP_PROTOCOLS_KEY_NAME, new ArrayOf(JOIN_GROUP_REQUEST_PROTOCOL_V0), "List of protocols " +
                     "that the member supports"));
 
     private static final Schema JOIN_GROUP_REQUEST_V1 = new Schema(
-            new Field(GROUP_ID_KEY_NAME, STRING, "The group id."),
+            GROUP_ID,
             new Field(SESSION_TIMEOUT_KEY_NAME, INT32, "The coordinator considers the consumer dead if it receives no " +
                     "heartbeat after this timeout in ms."),
             new Field(REBALANCE_TIMEOUT_KEY_NAME, INT32, "The maximum time that the coordinator will wait for each " +
                     "member to rejoin when rebalancing the group"),
-            new Field(MEMBER_ID_KEY_NAME, STRING, "The assigned consumer id or an empty string for a new consumer."),
+            MEMBER_ID,
             new Field(PROTOCOL_TYPE_KEY_NAME, STRING, "Unique name for class of protocols implemented by group"),
             new Field(GROUP_PROTOCOLS_KEY_NAME, new ArrayOf(JOIN_GROUP_REQUEST_PROTOCOL_V0), "List of protocols " +
                     "that the member supports"));
@@ -166,7 +166,7 @@ public class JoinGroupRequest extends AbstractRequest {
     public JoinGroupRequest(Struct struct, short versionId) {
         super(versionId);
 
-        groupId = struct.getString(GROUP_ID_KEY_NAME);
+        groupId = struct.get(GROUP_ID);
         sessionTimeout = struct.getInt(SESSION_TIMEOUT_KEY_NAME);
 
         if (struct.hasField(REBALANCE_TIMEOUT_KEY_NAME))
@@ -176,7 +176,7 @@ public class JoinGroupRequest extends AbstractRequest {
             // v0 had no rebalance timeout but used session timeout implicitly
             rebalanceTimeout = sessionTimeout;
 
-        memberId = struct.getString(MEMBER_ID_KEY_NAME);
+        memberId = struct.get(MEMBER_ID);
         protocolType = struct.getString(PROTOCOL_TYPE_KEY_NAME);
 
         groupProtocols = new ArrayList<>();
@@ -249,12 +249,12 @@ public class JoinGroupRequest extends AbstractRequest {
     protected Struct toStruct() {
         short version = version();
         Struct struct = new Struct(ApiKeys.JOIN_GROUP.requestSchema(version));
-        struct.set(GROUP_ID_KEY_NAME, groupId);
+        struct.set(GROUP_ID, groupId);
         struct.set(SESSION_TIMEOUT_KEY_NAME, sessionTimeout);
         if (version >= 1) {
             struct.set(REBALANCE_TIMEOUT_KEY_NAME, rebalanceTimeout);
         }
-        struct.set(MEMBER_ID_KEY_NAME, memberId);
+        struct.set(MEMBER_ID, memberId);
         struct.set(PROTOCOL_TYPE_KEY_NAME, protocolType);
         List<Struct> groupProtocolsList = new ArrayList<>(groupProtocols.size());
         for (ProtocolMetadata protocol : groupProtocols) {

http://git-wip-us.apache.org/repos/asf/kafka/blob/f4ac68a4/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 d9f987b..4bcd6e6 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
@@ -31,31 +31,30 @@ import java.util.List;
 import java.util.Map;
 
 import static org.apache.kafka.common.protocol.CommonFields.ERROR_CODE;
+import static org.apache.kafka.common.protocol.CommonFields.GENERATION_ID;
+import static org.apache.kafka.common.protocol.CommonFields.MEMBER_ID;
 import static org.apache.kafka.common.protocol.CommonFields.THROTTLE_TIME_MS;
 import static org.apache.kafka.common.protocol.types.Type.BYTES;
-import static org.apache.kafka.common.protocol.types.Type.INT32;
 import static org.apache.kafka.common.protocol.types.Type.STRING;
 
 public class JoinGroupResponse extends AbstractResponse {
 
-    private static final String GENERATION_ID_KEY_NAME = "generation_id";
     private static final String GROUP_PROTOCOL_KEY_NAME = "group_protocol";
     private static final String LEADER_ID_KEY_NAME = "leader_id";
-    private static final String MEMBER_ID_KEY_NAME = "member_id";
     private static final String MEMBERS_KEY_NAME = "members";
 
     private static final String MEMBER_METADATA_KEY_NAME = "member_metadata";
 
     private static final Schema JOIN_GROUP_RESPONSE_MEMBER_V0 = new Schema(
-            new Field(MEMBER_ID_KEY_NAME, STRING),
+            MEMBER_ID,
             new Field(MEMBER_METADATA_KEY_NAME, BYTES));
 
     private static final Schema JOIN_GROUP_RESPONSE_V0 = new Schema(
             ERROR_CODE,
-            new Field(GENERATION_ID_KEY_NAME, INT32, "The generation of the consumer group."),
+            GENERATION_ID,
             new Field(GROUP_PROTOCOL_KEY_NAME, STRING, "The group protocol selected by the coordinator"),
             new Field(LEADER_ID_KEY_NAME, STRING, "The leader of the group"),
-            new Field(MEMBER_ID_KEY_NAME, STRING, "The consumer id assigned by the group coordinator."),
+            MEMBER_ID,
             new Field(MEMBERS_KEY_NAME, new ArrayOf(JOIN_GROUP_RESPONSE_MEMBER_V0)));
 
     private static final Schema JOIN_GROUP_RESPONSE_V1 = JOIN_GROUP_RESPONSE_V0;
@@ -63,10 +62,10 @@ public class JoinGroupResponse extends AbstractResponse {
     private static final Schema JOIN_GROUP_RESPONSE_V2 = new Schema(
             THROTTLE_TIME_MS,
             ERROR_CODE,
-            new Field(GENERATION_ID_KEY_NAME, INT32, "The generation of the consumer group."),
+            GENERATION_ID,
             new Field(GROUP_PROTOCOL_KEY_NAME, STRING, "The group protocol selected by the coordinator"),
             new Field(LEADER_ID_KEY_NAME, STRING, "The leader of the group"),
-            new Field(MEMBER_ID_KEY_NAME, STRING, "The consumer id assigned by the group coordinator."),
+            MEMBER_ID,
             new Field(MEMBERS_KEY_NAME, new ArrayOf(JOIN_GROUP_RESPONSE_MEMBER_V0)));
 
 
@@ -129,14 +128,14 @@ public class JoinGroupResponse extends AbstractResponse {
 
         for (Object memberDataObj : struct.getArray(MEMBERS_KEY_NAME)) {
             Struct memberData = (Struct) memberDataObj;
-            String memberId = memberData.getString(MEMBER_ID_KEY_NAME);
+            String memberId = memberData.get(MEMBER_ID);
             ByteBuffer memberMetadata = memberData.getBytes(MEMBER_METADATA_KEY_NAME);
             members.put(memberId, memberMetadata);
         }
         error = Errors.forCode(struct.get(ERROR_CODE));
-        generationId = struct.getInt(GENERATION_ID_KEY_NAME);
+        generationId = struct.get(GENERATION_ID);
         groupProtocol = struct.getString(GROUP_PROTOCOL_KEY_NAME);
-        memberId = struct.getString(MEMBER_ID_KEY_NAME);
+        memberId = struct.get(MEMBER_ID);
         leaderId = struct.getString(LEADER_ID_KEY_NAME);
     }
 
@@ -187,15 +186,15 @@ public class JoinGroupResponse extends AbstractResponse {
         struct.setIfExists(THROTTLE_TIME_MS, throttleTimeMs);
 
         struct.set(ERROR_CODE, error.code());
-        struct.set(GENERATION_ID_KEY_NAME, generationId);
+        struct.set(GENERATION_ID, generationId);
         struct.set(GROUP_PROTOCOL_KEY_NAME, groupProtocol);
-        struct.set(MEMBER_ID_KEY_NAME, memberId);
+        struct.set(MEMBER_ID, memberId);
         struct.set(LEADER_ID_KEY_NAME, leaderId);
 
         List<Struct> memberArray = new ArrayList<>();
         for (Map.Entry<String, ByteBuffer> entries : members.entrySet()) {
             Struct memberData = struct.instance(MEMBERS_KEY_NAME);
-            memberData.set(MEMBER_ID_KEY_NAME, entries.getKey());
+            memberData.set(MEMBER_ID, entries.getKey());
             memberData.set(MEMBER_METADATA_KEY_NAME, entries.getValue());
             memberArray.add(memberData);
         }

http://git-wip-us.apache.org/repos/asf/kafka/blob/f4ac68a4/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 661eb7f..b0d0ad6 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
@@ -18,21 +18,18 @@ 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.types.Field;
 import org.apache.kafka.common.protocol.types.Schema;
 import org.apache.kafka.common.protocol.types.Struct;
 
 import java.nio.ByteBuffer;
 
-import static org.apache.kafka.common.protocol.types.Type.STRING;
+import static org.apache.kafka.common.protocol.CommonFields.GROUP_ID;
+import static org.apache.kafka.common.protocol.CommonFields.MEMBER_ID;
 
 public class LeaveGroupRequest extends AbstractRequest {
-    private static final String GROUP_ID_KEY_NAME = "group_id";
-    private static final String MEMBER_ID_KEY_NAME = "member_id";
-
     private static final Schema LEAVE_GROUP_REQUEST_V0 = new Schema(
-            new Field(GROUP_ID_KEY_NAME, STRING, "The group id."),
-            new Field(MEMBER_ID_KEY_NAME, STRING, "The member id assigned by the group coordinator."));
+            GROUP_ID,
+            MEMBER_ID);
 
     /* v1 request is the same as v0. Throttle time has been added to response */
     private static final Schema LEAVE_GROUP_REQUEST_V1 = LEAVE_GROUP_REQUEST_V0;
@@ -78,8 +75,8 @@ public class LeaveGroupRequest extends AbstractRequest {
 
     public LeaveGroupRequest(Struct struct, short version) {
         super(version);
-        groupId = struct.getString(GROUP_ID_KEY_NAME);
-        memberId = struct.getString(MEMBER_ID_KEY_NAME);
+        groupId = struct.get(GROUP_ID);
+        memberId = struct.get(MEMBER_ID);
     }
 
     @Override
@@ -111,8 +108,8 @@ public class LeaveGroupRequest extends AbstractRequest {
     @Override
     protected Struct toStruct() {
         Struct struct = new Struct(ApiKeys.LEAVE_GROUP.requestSchema(version()));
-        struct.set(GROUP_ID_KEY_NAME, groupId);
-        struct.set(MEMBER_ID_KEY_NAME, memberId);
+        struct.set(GROUP_ID, groupId);
+        struct.set(MEMBER_ID, memberId);
         return struct;
     }
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/f4ac68a4/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 afc5ebd..9c82ae0 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
@@ -29,17 +29,17 @@ import java.util.List;
 import java.util.Map;
 
 import static org.apache.kafka.common.protocol.CommonFields.ERROR_CODE;
+import static org.apache.kafka.common.protocol.CommonFields.GROUP_ID;
 import static org.apache.kafka.common.protocol.CommonFields.THROTTLE_TIME_MS;
 import static org.apache.kafka.common.protocol.types.Type.STRING;
 
 public class ListGroupsResponse extends AbstractResponse {
 
     private static final String GROUPS_KEY_NAME = "groups";
-    private static final String GROUP_ID_KEY_NAME = "group_id";
     private static final String PROTOCOL_TYPE_KEY_NAME = "protocol_type";
 
     private static final Schema LIST_GROUPS_RESPONSE_GROUP_V0 = new Schema(
-            new Field(GROUP_ID_KEY_NAME, STRING),
+            GROUP_ID,
             new Field(PROTOCOL_TYPE_KEY_NAME, STRING));
     private static final Schema LIST_GROUPS_RESPONSE_V0 = new Schema(
             ERROR_CODE,
@@ -80,7 +80,7 @@ public class ListGroupsResponse extends AbstractResponse {
         this.groups = new ArrayList<>();
         for (Object groupObj : struct.getArray(GROUPS_KEY_NAME)) {
             Struct groupStruct = (Struct) groupObj;
-            String groupId = groupStruct.getString(GROUP_ID_KEY_NAME);
+            String groupId = groupStruct.get(GROUP_ID);
             String protocolType = groupStruct.getString(PROTOCOL_TYPE_KEY_NAME);
             this.groups.add(new Group(groupId, protocolType));
         }
@@ -130,7 +130,7 @@ public class ListGroupsResponse extends AbstractResponse {
         List<Struct> groupList = new ArrayList<>();
         for (Group group : groups) {
             Struct groupStruct = struct.instance(GROUPS_KEY_NAME);
-            groupStruct.set(GROUP_ID_KEY_NAME, group.groupId);
+            groupStruct.set(GROUP_ID, group.groupId);
             groupStruct.set(PROTOCOL_TYPE_KEY_NAME, group.protocolType);
             groupList.add(groupStruct);
         }

http://git-wip-us.apache.org/repos/asf/kafka/blob/f4ac68a4/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 696d967..4686c3b 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
@@ -32,20 +32,18 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import static org.apache.kafka.common.protocol.CommonFields.GENERATION_ID;
+import static org.apache.kafka.common.protocol.CommonFields.GROUP_ID;
+import static org.apache.kafka.common.protocol.CommonFields.MEMBER_ID;
 import static org.apache.kafka.common.protocol.CommonFields.PARTITION_ID;
 import static org.apache.kafka.common.protocol.CommonFields.TOPIC_NAME;
-import static org.apache.kafka.common.protocol.types.Type.INT32;
 import static org.apache.kafka.common.protocol.types.Type.INT64;
 import static org.apache.kafka.common.protocol.types.Type.NULLABLE_STRING;
-import static org.apache.kafka.common.protocol.types.Type.STRING;
 
 /**
  * This wrapper supports both v0 and v1 of OffsetCommitRequest.
  */
 public class OffsetCommitRequest extends AbstractRequest {
-    private static final String GROUP_ID_KEY_NAME = "group_id";
-    private static final String GENERATION_ID_KEY_NAME = "group_generation_id";
-    private static final String MEMBER_ID_KEY_NAME = "member_id";
     private static final String TOPICS_KEY_NAME = "topics";
     private static final String RETENTION_TIME_KEY_NAME = "retention_time";
 
@@ -89,19 +87,19 @@ public class OffsetCommitRequest extends AbstractRequest {
             new Field(PARTITIONS_KEY_NAME, new ArrayOf(OFFSET_COMMIT_REQUEST_PARTITION_V2), "Partitions to commit offsets."));
 
     private static final Schema OFFSET_COMMIT_REQUEST_V0 = new Schema(
-            new Field(GROUP_ID_KEY_NAME, STRING, "The group id."),
+            GROUP_ID,
             new Field(TOPICS_KEY_NAME, new ArrayOf(OFFSET_COMMIT_REQUEST_TOPIC_V0), "Topics to commit offsets."));
 
     private static final Schema OFFSET_COMMIT_REQUEST_V1 = new Schema(
-            new Field(GROUP_ID_KEY_NAME, STRING, "The group id."),
-            new Field(GENERATION_ID_KEY_NAME, INT32, "The generation of the group."),
-            new Field(MEMBER_ID_KEY_NAME, STRING, "The member id assigned by the group coordinator."),
+            GROUP_ID,
+            GENERATION_ID,
+            MEMBER_ID,
             new Field(TOPICS_KEY_NAME, new ArrayOf(OFFSET_COMMIT_REQUEST_TOPIC_V1), "Topics to commit offsets."));
 
     private static final Schema OFFSET_COMMIT_REQUEST_V2 = new Schema(
-            new Field(GROUP_ID_KEY_NAME, STRING, "The group id."),
-            new Field(GENERATION_ID_KEY_NAME, INT32, "The generation of the consumer group."),
-            new Field(MEMBER_ID_KEY_NAME, STRING, "The consumer id assigned by the group coordinator."),
+            GROUP_ID,
+            GENERATION_ID,
+            MEMBER_ID,
             new Field(RETENTION_TIME_KEY_NAME, INT64, "Time period in ms to retain the offset."),
             new Field(TOPICS_KEY_NAME, new ArrayOf(OFFSET_COMMIT_REQUEST_TOPIC_V2), "Topics to commit offsets."));
 
@@ -229,18 +227,11 @@ public class OffsetCommitRequest extends AbstractRequest {
     public OffsetCommitRequest(Struct struct, short versionId) {
         super(versionId);
 
-        groupId = struct.getString(GROUP_ID_KEY_NAME);
-        // This field only exists in v1.
-        if (struct.hasField(GENERATION_ID_KEY_NAME))
-            generationId = struct.getInt(GENERATION_ID_KEY_NAME);
-        else
-            generationId = DEFAULT_GENERATION_ID;
+        groupId = struct.get(GROUP_ID);
 
-        // This field only exists in v1.
-        if (struct.hasField(MEMBER_ID_KEY_NAME))
-            memberId = struct.getString(MEMBER_ID_KEY_NAME);
-        else
-            memberId = DEFAULT_MEMBER_ID;
+        // These fields only exists in v1.
+        generationId = struct.getOrElse(GENERATION_ID, DEFAULT_GENERATION_ID);
+        memberId = struct.getOrElse(MEMBER_ID, DEFAULT_MEMBER_ID);
 
         // This field only exists in v2
         if (struct.hasField(RETENTION_TIME_KEY_NAME))
@@ -274,7 +265,7 @@ public class OffsetCommitRequest extends AbstractRequest {
     public Struct toStruct() {
         short version = version();
         Struct struct = new Struct(ApiKeys.OFFSET_COMMIT.requestSchema(version));
-        struct.set(GROUP_ID_KEY_NAME, groupId);
+        struct.set(GROUP_ID, groupId);
 
         Map<String, Map<Integer, PartitionData>> topicsData = CollectionUtils.groupDataByTopic(offsetData);
         List<Struct> topicArray = new ArrayList<>();
@@ -297,10 +288,8 @@ public class OffsetCommitRequest extends AbstractRequest {
             topicArray.add(topicData);
         }
         struct.set(TOPICS_KEY_NAME, topicArray.toArray());
-        if (struct.hasField(GENERATION_ID_KEY_NAME))
-            struct.set(GENERATION_ID_KEY_NAME, generationId);
-        if (struct.hasField(MEMBER_ID_KEY_NAME))
-            struct.set(MEMBER_ID_KEY_NAME, memberId);
+        struct.setIfExists(GENERATION_ID, generationId);
+        struct.setIfExists(MEMBER_ID, memberId);
         if (struct.hasField(RETENTION_TIME_KEY_NAME))
             struct.set(RETENTION_TIME_KEY_NAME, retentionTime);
         return struct;

http://git-wip-us.apache.org/repos/asf/kafka/blob/f4ac68a4/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 6d8b959..0db1c50 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
@@ -33,12 +33,11 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import static org.apache.kafka.common.protocol.CommonFields.GROUP_ID;
 import static org.apache.kafka.common.protocol.CommonFields.PARTITION_ID;
 import static org.apache.kafka.common.protocol.CommonFields.TOPIC_NAME;
-import static org.apache.kafka.common.protocol.types.Type.STRING;
 
 public class OffsetFetchRequest extends AbstractRequest {
-    private static final String GROUP_ID_KEY_NAME = "group_id";
     private static final String TOPICS_KEY_NAME = "topics";
 
     // topic level field names
@@ -63,13 +62,13 @@ public class OffsetFetchRequest extends AbstractRequest {
             new Field(PARTITIONS_KEY_NAME, new ArrayOf(OFFSET_FETCH_REQUEST_PARTITION_V0), "Partitions to fetch offsets."));
 
     private static final Schema OFFSET_FETCH_REQUEST_V0 = new Schema(
-            new Field(GROUP_ID_KEY_NAME, STRING, "The consumer group id."),
+            GROUP_ID,
             new Field(TOPICS_KEY_NAME, new ArrayOf(OFFSET_FETCH_REQUEST_TOPIC_V0), "Topics to fetch offsets."));
 
     private static final Schema OFFSET_FETCH_REQUEST_V1 = OFFSET_FETCH_REQUEST_V0;
 
     private static final Schema OFFSET_FETCH_REQUEST_V2 = new Schema(
-            new Field(GROUP_ID_KEY_NAME, STRING, "The consumer group id."),
+            GROUP_ID,
             new Field(TOPICS_KEY_NAME, ArrayOf.nullable(OFFSET_FETCH_REQUEST_TOPIC_V0), "Topics to fetch offsets. If the " +
                     "topic array is null fetch offsets for all topics."));
 
@@ -153,7 +152,7 @@ public class OffsetFetchRequest extends AbstractRequest {
             partitions = null;
 
 
-        groupId = struct.getString(GROUP_ID_KEY_NAME);
+        groupId = struct.get(GROUP_ID);
     }
 
     public OffsetFetchResponse getErrorResponse(Errors error) {
@@ -210,7 +209,7 @@ public class OffsetFetchRequest extends AbstractRequest {
     @Override
     protected Struct toStruct() {
         Struct struct = new Struct(ApiKeys.OFFSET_FETCH.requestSchema(version()));
-        struct.set(GROUP_ID_KEY_NAME, groupId);
+        struct.set(GROUP_ID, groupId);
         if (partitions != null) {
             Map<String, List<Integer>> topicsData = CollectionUtils.groupDataByTopic(partitions);
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/f4ac68a4/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 4ff9fcd..14ed262 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
@@ -30,24 +30,22 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import static org.apache.kafka.common.protocol.CommonFields.GENERATION_ID;
+import static org.apache.kafka.common.protocol.CommonFields.GROUP_ID;
+import static org.apache.kafka.common.protocol.CommonFields.MEMBER_ID;
 import static org.apache.kafka.common.protocol.types.Type.BYTES;
-import static org.apache.kafka.common.protocol.types.Type.INT32;
-import static org.apache.kafka.common.protocol.types.Type.STRING;
 
 public class SyncGroupRequest extends AbstractRequest {
-    private static final String GROUP_ID_KEY_NAME = "group_id";
-    private static final String GENERATION_ID_KEY_NAME = "generation_id";
-    private static final String MEMBER_ID_KEY_NAME = "member_id";
     private static final String MEMBER_ASSIGNMENT_KEY_NAME = "member_assignment";
     private static final String GROUP_ASSIGNMENT_KEY_NAME = "group_assignment";
 
     private static final Schema SYNC_GROUP_REQUEST_MEMBER_V0 = new Schema(
-            new Field(MEMBER_ID_KEY_NAME, STRING),
+            MEMBER_ID,
             new Field(MEMBER_ASSIGNMENT_KEY_NAME, BYTES));
     private static final Schema SYNC_GROUP_REQUEST_V0 = new Schema(
-            new Field(GROUP_ID_KEY_NAME, STRING),
-            new Field(GENERATION_ID_KEY_NAME, INT32),
-            new Field(MEMBER_ID_KEY_NAME, STRING),
+            GROUP_ID,
+            GENERATION_ID,
+            MEMBER_ID,
             new Field(GROUP_ASSIGNMENT_KEY_NAME, new ArrayOf(SYNC_GROUP_REQUEST_MEMBER_V0)));
 
     /* v1 request is the same as v0. Throttle time has been added to response */
@@ -106,15 +104,15 @@ public class SyncGroupRequest extends AbstractRequest {
 
     public SyncGroupRequest(Struct struct, short version) {
         super(version);
-        this.groupId = struct.getString(GROUP_ID_KEY_NAME);
-        this.generationId = struct.getInt(GENERATION_ID_KEY_NAME);
-        this.memberId = struct.getString(MEMBER_ID_KEY_NAME);
+        this.groupId = struct.get(GROUP_ID);
+        this.generationId = struct.get(GENERATION_ID);
+        this.memberId = struct.get(MEMBER_ID);
 
         groupAssignment = new HashMap<>();
 
         for (Object memberDataObj : struct.getArray(GROUP_ASSIGNMENT_KEY_NAME)) {
             Struct memberData = (Struct) memberDataObj;
-            String memberId = memberData.getString(MEMBER_ID_KEY_NAME);
+            String memberId = memberData.get(MEMBER_ID);
             ByteBuffer memberMetadata = memberData.getBytes(MEMBER_ASSIGNMENT_KEY_NAME);
             groupAssignment.put(memberId, memberMetadata);
         }
@@ -162,14 +160,14 @@ public class SyncGroupRequest extends AbstractRequest {
     @Override
     protected Struct toStruct() {
         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);
+        struct.set(GROUP_ID, groupId);
+        struct.set(GENERATION_ID, generationId);
+        struct.set(MEMBER_ID, memberId);
 
         List<Struct> memberArray = new ArrayList<>();
         for (Map.Entry<String, ByteBuffer> entries: groupAssignment.entrySet()) {
             Struct memberData = struct.instance(GROUP_ASSIGNMENT_KEY_NAME);
-            memberData.set(MEMBER_ID_KEY_NAME, entries.getKey());
+            memberData.set(MEMBER_ID, entries.getKey());
             memberData.set(MEMBER_ASSIGNMENT_KEY_NAME, entries.getValue());
             memberArray.add(memberData);
         }

http://git-wip-us.apache.org/repos/asf/kafka/blob/f4ac68a4/clients/src/main/java/org/apache/kafka/common/requests/TxnOffsetCommitRequest.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/TxnOffsetCommitRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/TxnOffsetCommitRequest.java
index d384192..9787c2d 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/TxnOffsetCommitRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/TxnOffsetCommitRequest.java
@@ -29,18 +29,16 @@ import java.nio.ByteBuffer;
 import java.util.HashMap;
 import java.util.Map;
 
+import static org.apache.kafka.common.protocol.CommonFields.GROUP_ID;
 import static org.apache.kafka.common.protocol.CommonFields.PARTITION_ID;
+import static org.apache.kafka.common.protocol.CommonFields.PRODUCER_EPOCH;
+import static org.apache.kafka.common.protocol.CommonFields.PRODUCER_ID;
 import static org.apache.kafka.common.protocol.CommonFields.TOPIC_NAME;
-import static org.apache.kafka.common.protocol.types.Type.INT16;
+import static org.apache.kafka.common.protocol.CommonFields.TRANSACTIONAL_ID;
 import static org.apache.kafka.common.protocol.types.Type.INT64;
 import static org.apache.kafka.common.protocol.types.Type.NULLABLE_STRING;
-import static org.apache.kafka.common.protocol.types.Type.STRING;
 
 public class TxnOffsetCommitRequest extends AbstractRequest {
-    private static final String TRANSACTIONAL_ID_KEY_NAME = "transactional_id";
-    private static final String CONSUMER_GROUP_ID_KEY_NAME = "consumer_group_id";
-    private static final String PRODUCER_ID_KEY_NAME = "producer_id";
-    private static final String PRODUCER_EPOCH_KEY_NAME = "producer_epoch";
     private static final String TOPICS_KEY_NAME = "topics";
     private static final String PARTITIONS_KEY_NAME = "partitions";
     private static final String OFFSET_KEY_NAME = "offset";
@@ -52,10 +50,10 @@ public class TxnOffsetCommitRequest extends AbstractRequest {
             new Field(METADATA_KEY_NAME, NULLABLE_STRING));
 
     private static final Schema TXN_OFFSET_COMMIT_REQUEST_V0 = new Schema(
-            new Field(TRANSACTIONAL_ID_KEY_NAME, STRING, "The transactional id corresponding to the transaction."),
-            new Field(CONSUMER_GROUP_ID_KEY_NAME, STRING, "Id of the associated consumer group to commit offsets for."),
-            new Field(PRODUCER_ID_KEY_NAME, INT64, "Current producer id in use by the transactional id."),
-            new Field(PRODUCER_EPOCH_KEY_NAME, INT16, "Current epoch associated with the producer id."),
+            TRANSACTIONAL_ID,
+            GROUP_ID,
+            PRODUCER_ID,
+            PRODUCER_EPOCH,
             new Field(TOPICS_KEY_NAME, new ArrayOf(new Schema(
                     TOPIC_NAME,
                     new Field(PARTITIONS_KEY_NAME, new ArrayOf(TXN_OFFSET_COMMIT_PARTITION_OFFSET_METADATA_REQUEST_V0)))),
@@ -127,10 +125,10 @@ public class TxnOffsetCommitRequest extends AbstractRequest {
 
     public TxnOffsetCommitRequest(Struct struct, short version) {
         super(version);
-        this.transactionalId = struct.getString(TRANSACTIONAL_ID_KEY_NAME);
-        this.consumerGroupId = struct.getString(CONSUMER_GROUP_ID_KEY_NAME);
-        this.producerId = struct.getLong(PRODUCER_ID_KEY_NAME);
-        this.producerEpoch = struct.getShort(PRODUCER_EPOCH_KEY_NAME);
+        this.transactionalId = struct.get(TRANSACTIONAL_ID);
+        this.consumerGroupId = struct.get(GROUP_ID);
+        this.producerId = struct.get(PRODUCER_ID);
+        this.producerEpoch = struct.get(PRODUCER_EPOCH);
 
         Map<TopicPartition, CommittedOffset> offsets = new HashMap<>();
         Object[] topicPartitionsArray = struct.getArray(TOPICS_KEY_NAME);
@@ -171,10 +169,10 @@ public class TxnOffsetCommitRequest extends AbstractRequest {
     @Override
     protected Struct toStruct() {
         Struct struct = new Struct(ApiKeys.TXN_OFFSET_COMMIT.requestSchema(version()));
-        struct.set(TRANSACTIONAL_ID_KEY_NAME, transactionalId);
-        struct.set(CONSUMER_GROUP_ID_KEY_NAME, consumerGroupId);
-        struct.set(PRODUCER_ID_KEY_NAME, producerId);
-        struct.set(PRODUCER_EPOCH_KEY_NAME, producerEpoch);
+        struct.set(TRANSACTIONAL_ID, transactionalId);
+        struct.set(GROUP_ID, consumerGroupId);
+        struct.set(PRODUCER_ID, producerId);
+        struct.set(PRODUCER_EPOCH, producerEpoch);
 
         Map<String, Map<Integer, CommittedOffset>> mappedPartitionOffsets = CollectionUtils.groupDataByTopic(offsets);
         Object[] partitionsArray = new Object[mappedPartitionOffsets.size()];


Mime
View raw message