kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From j...@apache.org
Subject [5/5] kafka git commit: MINOR: Use an explicit `Errors` object when possible instead of a numeric error code
Date Fri, 10 Feb 2017 05:21:53 GMT
MINOR: Use an explicit `Errors` object when possible instead of a numeric error code

Author: Vahid Hashemian <vahidhashemian@us.ibm.com>

Reviewers: Ismael Juma <ismael@juma.me.uk>, Jason Gustafson <jason@confluent.io>

Closes #2475 from vahidhashemian/minor/use_explicit_Errors_type_when_possible


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

Branch: refs/heads/trunk
Commit: 9898d665d1ab201405d66c70e3ea9710d9dcecd7
Parents: a15fcea
Author: Vahid Hashemian <vahidhashemian@us.ibm.com>
Authored: Thu Feb 9 21:03:46 2017 -0800
Committer: Jason Gustafson <jason@confluent.io>
Committed: Thu Feb 9 21:03:46 2017 -0800

----------------------------------------------------------------------
 .../org/apache/kafka/clients/NetworkClient.java |   4 +-
 .../consumer/internals/AbstractCoordinator.java |  10 +-
 .../consumer/internals/ConsumerCoordinator.java |   4 +-
 .../clients/consumer/internals/Fetcher.java     |   4 +-
 .../common/requests/ApiVersionsRequest.java     |   3 +-
 .../common/requests/ApiVersionsResponse.java    |  18 +-
 .../requests/ControlledShutdownRequest.java     |   2 +-
 .../requests/ControlledShutdownResponse.java    |  15 +-
 .../common/requests/CreateTopicsResponse.java   |   4 +-
 .../common/requests/DeleteTopicsResponse.java   |   4 +-
 .../common/requests/DescribeGroupsResponse.java |  18 +-
 .../kafka/common/requests/FetchRequest.java     |   2 +-
 .../kafka/common/requests/FetchResponse.java    |  17 +-
 .../requests/GroupCoordinatorRequest.java       |   2 +-
 .../requests/GroupCoordinatorResponse.java      |  17 +-
 .../kafka/common/requests/HeartbeatRequest.java |   2 +-
 .../common/requests/HeartbeatResponse.java      |  18 +-
 .../kafka/common/requests/JoinGroupRequest.java |   2 +-
 .../common/requests/JoinGroupResponse.java      |  19 +-
 .../common/requests/LeaderAndIsrRequest.java    |   6 +-
 .../common/requests/LeaderAndIsrResponse.java   |  32 +-
 .../common/requests/LeaveGroupRequest.java      |   2 +-
 .../common/requests/LeaveGroupResponse.java     |  15 +-
 .../common/requests/ListGroupsRequest.java      |   3 +-
 .../common/requests/ListGroupsResponse.java     |  16 +-
 .../common/requests/ListOffsetRequest.java      |   4 +-
 .../common/requests/ListOffsetResponse.java     |  23 +-
 .../common/requests/OffsetCommitRequest.java    |   8 +-
 .../common/requests/OffsetCommitResponse.java   |  23 +-
 .../common/requests/SaslHandshakeRequest.java   |   2 +-
 .../common/requests/SaslHandshakeResponse.java  |  15 +-
 .../common/requests/StopReplicaRequest.java     |   6 +-
 .../common/requests/StopReplicaResponse.java    |  32 +-
 .../kafka/common/requests/SyncGroupRequest.java |   2 +-
 .../common/requests/SyncGroupResponse.java      |  15 +-
 .../common/requests/UpdateMetadataRequest.java  |   2 +-
 .../common/requests/UpdateMetadataResponse.java |  15 +-
 .../authenticator/SaslClientAuthenticator.java  |   4 +-
 .../authenticator/SaslServerAuthenticator.java  |   4 +-
 .../clients/consumer/KafkaConsumerTest.java     |  56 ++--
 .../internals/AbstractCoordinatorTest.java      |   8 +-
 .../internals/ConsumerCoordinatorTest.java      | 320 +++++++++----------
 .../internals/ConsumerNetworkClientTest.java    |  16 +-
 .../clients/consumer/internals/FetcherTest.java |  48 +--
 .../clients/producer/internals/SenderTest.java  |  21 +-
 .../common/requests/RequestResponseTest.java    |  46 +--
 .../authenticator/SaslAuthenticatorTest.java    |   6 +-
 .../distributed/WorkerCoordinatorTest.java      |  40 +--
 .../main/scala/kafka/admin/AdminClient.scala    |   8 +-
 .../kafka/admin/ConsumerGroupCommand.scala      |   4 +-
 .../kafka/api/ControlledShutdownRequest.scala   |   4 +-
 .../kafka/api/ControlledShutdownResponse.scala  |   8 +-
 .../src/main/scala/kafka/api/FetchRequest.scala |   2 +-
 .../main/scala/kafka/api/FetchResponse.scala    |   8 +-
 .../kafka/api/GroupCoordinatorRequest.scala     |   2 +-
 .../kafka/api/GroupCoordinatorResponse.scala    |  10 +-
 .../scala/kafka/api/OffsetCommitRequest.scala   |   4 +-
 .../scala/kafka/api/OffsetCommitResponse.scala  |  10 +-
 .../scala/kafka/api/OffsetFetchRequest.scala    |   4 +-
 .../scala/kafka/api/OffsetFetchResponse.scala   |  16 +-
 .../main/scala/kafka/api/OffsetRequest.scala    |   2 +-
 .../main/scala/kafka/api/OffsetResponse.scala   |  10 +-
 .../main/scala/kafka/api/ProducerRequest.scala  |   2 +-
 .../main/scala/kafka/api/ProducerResponse.scala |   8 +-
 .../main/scala/kafka/api/TopicMetadata.scala    |  20 +-
 .../scala/kafka/api/TopicMetadataRequest.scala  |   2 +-
 .../main/scala/kafka/client/ClientUtils.scala   |   2 +-
 .../main/scala/kafka/common/ErrorMapping.scala  |  20 +-
 .../kafka/common/OffsetMetadataAndError.scala   |  22 +-
 .../kafka/consumer/ConsumerFetcherThread.scala  |   5 +-
 .../scala/kafka/consumer/SimpleConsumer.scala   |   5 +-
 .../consumer/ZookeeperConsumerConnector.scala   |  20 +-
 .../kafka/controller/TopicDeletionManager.scala |   4 +-
 .../kafka/coordinator/GroupCoordinator.scala    | 138 ++++----
 .../coordinator/GroupMetadataManager.scala      |  14 +-
 .../kafka/coordinator/MemberMetadata.scala      |   4 +-
 .../scala/kafka/javaapi/FetchResponse.scala     |   4 +-
 .../javaapi/GroupCoordinatorResponse.scala      |   4 +-
 .../kafka/javaapi/OffsetCommitResponse.scala    |   6 +-
 .../scala/kafka/javaapi/OffsetResponse.scala    |   4 +-
 .../scala/kafka/javaapi/TopicMetadata.scala     |   9 +-
 .../kafka/producer/BrokerPartitionInfo.scala    |  12 +-
 .../producer/async/DefaultEventHandler.scala    |   6 +-
 .../kafka/security/auth/ResourceType.scala      |   8 +-
 .../kafka/server/AbstractFetcherThread.scala    |   4 +-
 .../src/main/scala/kafka/server/KafkaApis.scala | 112 +++----
 .../main/scala/kafka/server/KafkaServer.scala   |   8 +-
 .../kafka/server/ReplicaFetcherThread.scala     |   6 +-
 .../scala/kafka/server/ReplicaManager.scala     |  40 +--
 .../kafka/tools/ConsumerOffsetChecker.scala     |   4 +-
 .../kafka/tools/ReplicaVerificationTool.scala   |   2 +-
 .../kafka/api/AuthorizerIntegrationTest.scala   |  38 +--
 .../ReplicaFetcherThreadFatalErrorTest.scala    |   2 +-
 .../tools/ReplicaVerificationToolTest.scala     |   2 +-
 .../scala/other/kafka/TestOffsetManager.scala   |   4 +-
 .../api/RequestResponseSerializationTest.scala  |  28 +-
 .../GroupCoordinatorResponseTest.scala          | 300 ++++++++---------
 .../coordinator/GroupMetadataManagerTest.scala  |  36 +--
 .../integration/BaseTopicMetadataTest.scala     |  26 +-
 .../kafka/integration/PrimitiveApiTest.scala    |   4 +-
 .../unit/kafka/producer/AsyncProducerTest.scala |   6 +-
 .../unit/kafka/producer/SyncProducerTest.scala  |  18 +-
 .../AbstractCreateTopicsRequestTest.scala       |   2 +-
 .../server/AbstractFetcherThreadTest.scala      |   2 +-
 .../kafka/server/ApiVersionsRequestTest.scala   |   2 +-
 .../kafka/server/DeleteTopicsRequestTest.scala  |   2 +-
 .../unit/kafka/server/FetchRequestTest.scala    |   8 +-
 .../unit/kafka/server/LeaderElectionTest.scala  |   2 +-
 .../scala/unit/kafka/server/LogOffsetTest.scala |   2 +-
 .../unit/kafka/server/OffsetCommitTest.scala    |  50 +--
 .../unit/kafka/server/ReplicaManagerTest.scala  |   8 +-
 .../server/SaslApiVersionsRequestTest.scala     |   4 +-
 112 files changed, 1054 insertions(+), 1034 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 3a75288..890bf56 100644
--- a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
@@ -544,9 +544,9 @@ public class NetworkClient implements KafkaClient {
     private void handleApiVersionsResponse(List<ClientResponse> responses,
                                            InFlightRequest req, long now, ApiVersionsResponse apiVersionsResponse) {
         final String node = req.destination;
-        if (apiVersionsResponse.errorCode() != Errors.NONE.code()) {
+        if (apiVersionsResponse.error() != Errors.NONE) {
             log.warn("Node {} got error {} when making an ApiVersionsRequest.  Disconnecting.",
-                    node, Errors.forCode(apiVersionsResponse.errorCode()));
+                    node, apiVersionsResponse.error());
             this.selector.close(node);
             processDisconnection(responses, node, now);
             return;

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
index 2fdf802..350a84b 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
@@ -420,7 +420,7 @@ public abstract class AbstractCoordinator implements Closeable {
     private class JoinGroupResponseHandler extends CoordinatorResponseHandler<JoinGroupResponse, ByteBuffer> {
         @Override
         public void handle(JoinGroupResponse joinResponse, RequestFuture<ByteBuffer> future) {
-            Errors error = Errors.forCode(joinResponse.errorCode());
+            Errors error = joinResponse.error();
             if (error == Errors.NONE) {
                 log.debug("Received successful JoinGroup response for group {}: {}", groupId, joinResponse);
                 sensors.joinLatency.record(response.requestLatencyMs());
@@ -509,7 +509,7 @@ public abstract class AbstractCoordinator implements Closeable {
         @Override
         public void handle(SyncGroupResponse syncResponse,
                            RequestFuture<ByteBuffer> future) {
-            Errors error = Errors.forCode(syncResponse.errorCode());
+            Errors error = syncResponse.error();
             if (error == Errors.NONE) {
                 sensors.syncLatency.record(response.requestLatencyMs());
                 future.complete(syncResponse.memberAssignment());
@@ -562,7 +562,7 @@ public abstract class AbstractCoordinator implements Closeable {
             // use MAX_VALUE - node.id as the coordinator id to mimic separate connections
             // for the coordinator in the underlying network client layer
             // TODO: this needs to be better handled in KAFKA-1935
-            Errors error = Errors.forCode(groupCoordinatorResponse.errorCode());
+            Errors error = groupCoordinatorResponse.error();
             clearFindCoordinatorFuture();
             if (error == Errors.NONE) {
                 synchronized (AbstractCoordinator.this) {
@@ -688,7 +688,7 @@ public abstract class AbstractCoordinator implements Closeable {
     private class LeaveGroupResponseHandler extends CoordinatorResponseHandler<LeaveGroupResponse, Void> {
         @Override
         public void handle(LeaveGroupResponse leaveResponse, RequestFuture<Void> future) {
-            Errors error = Errors.forCode(leaveResponse.errorCode());
+            Errors error = leaveResponse.error();
             if (error == Errors.NONE) {
                 log.debug("LeaveGroup request for group {} returned successfully", groupId);
                 future.complete(null);
@@ -712,7 +712,7 @@ public abstract class AbstractCoordinator implements Closeable {
         @Override
         public void handle(HeartbeatResponse heartbeatResponse, RequestFuture<Void> future) {
             sensors.heartbeatLatency.record(response.requestLatencyMs());
-            Errors error = Errors.forCode(heartbeatResponse.errorCode());
+            Errors error = heartbeatResponse.error();
             if (error == Errors.NONE) {
                 log.debug("Received successful Heartbeat response for group {}", groupId);
                 future.complete(null);

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
index 8669527..0dc073e 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
@@ -722,12 +722,12 @@ public final class ConsumerCoordinator extends AbstractCoordinator {
             sensors.commitLatency.record(response.requestLatencyMs());
             Set<String> unauthorizedTopics = new HashSet<>();
 
-            for (Map.Entry<TopicPartition, Short> entry : commitResponse.responseData().entrySet()) {
+            for (Map.Entry<TopicPartition, Errors> entry : commitResponse.responseData().entrySet()) {
                 TopicPartition tp = entry.getKey();
                 OffsetAndMetadata offsetAndMetadata = this.offsets.get(tp);
                 long offset = offsetAndMetadata.offset();
 
-                Errors error = Errors.forCode(entry.getValue());
+                Errors error = entry.getValue();
                 if (error == Errors.NONE) {
                     log.debug("Group {} committed offset {} for partition {}", groupId, offset, tp);
                     if (subscriptions.isAssigned(tp))

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 6a13d46..02f34e5 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
@@ -637,7 +637,7 @@ public class Fetcher<K, V> implements SubscriptionState.Listener {
         for (Map.Entry<TopicPartition, Long> entry : timestampsToSearch.entrySet()) {
             TopicPartition topicPartition = entry.getKey();
             ListOffsetResponse.PartitionData partitionData = listOffsetResponse.responseData().get(topicPartition);
-            Errors error = Errors.forCode(partitionData.errorCode);
+            Errors error = partitionData.error;
             if (error == Errors.NONE) {
                 if (partitionData.offsets != null) {
                     // Handle v0 response
@@ -750,7 +750,7 @@ public class Fetcher<K, V> implements SubscriptionState.Listener {
         int bytes = 0;
         int recordsCount = 0;
         PartitionRecords<K, V> parsedRecords = null;
-        Errors error = Errors.forCode(partition.errorCode);
+        Errors error = partition.error;
 
         try {
             if (!subscriptions.isFetchable(tp)) {

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 17e6d5e..fe7c348 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
@@ -51,8 +51,7 @@ public class ApiVersionsRequest extends AbstractRequest {
         short versionId = version();
         switch (versionId) {
             case 0:
-                short errorCode = Errors.forException(e).code();
-                return new ApiVersionsResponse(errorCode, Collections.<ApiVersionsResponse.ApiVersion>emptyList());
+                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)));

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 81be9c3..7d8bcc5 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
@@ -42,7 +42,7 @@ public class ApiVersionsResponse extends AbstractResponse {
      *
      * UNSUPPORTED_VERSION (33)
      */
-    private final short errorCode;
+    private final Errors error;
     private final Map<Short, ApiVersion> apiKeyToApiVersion;
 
     public static final class ApiVersion {
@@ -66,9 +66,9 @@ public class ApiVersionsResponse extends AbstractResponse {
         }
     }
 
-    public ApiVersionsResponse(short errorCode, List<ApiVersion> apiVersions) {
+    public ApiVersionsResponse(Errors error, List<ApiVersion> apiVersions) {
         super(new Struct(CURRENT_SCHEMA));
-        struct.set(ERROR_CODE_KEY_NAME, errorCode);
+        struct.set(ERROR_CODE_KEY_NAME, error.code());
         List<Struct> apiVersionList = new ArrayList<>();
         for (ApiVersion apiVersion : apiVersions) {
             Struct apiVersionStruct = struct.instance(API_VERSIONS_KEY_NAME);
@@ -78,13 +78,13 @@ public class ApiVersionsResponse extends AbstractResponse {
             apiVersionList.add(apiVersionStruct);
         }
         struct.set(API_VERSIONS_KEY_NAME, apiVersionList.toArray());
-        this.errorCode = errorCode;
+        this.error = error;
         this.apiKeyToApiVersion = buildApiKeyToApiVersion(apiVersions);
     }
 
     public ApiVersionsResponse(Struct struct) {
         super(struct);
-        this.errorCode = struct.getShort(ERROR_CODE_KEY_NAME);
+        this.error = Errors.forCode(struct.getShort(ERROR_CODE_KEY_NAME));
         List<ApiVersion> tempApiVersions = new ArrayList<>();
         for (Object apiVersionsObj : struct.getArray(API_VERSIONS_KEY_NAME)) {
             Struct apiVersionStruct = (Struct) apiVersionsObj;
@@ -104,8 +104,8 @@ public class ApiVersionsResponse extends AbstractResponse {
         return apiKeyToApiVersion.get(apiKey);
     }
 
-    public short errorCode() {
-        return errorCode;
+    public Errors error() {
+        return error;
     }
 
     public static ApiVersionsResponse parse(ByteBuffer buffer) {
@@ -113,7 +113,7 @@ public class ApiVersionsResponse extends AbstractResponse {
     }
 
     public static ApiVersionsResponse fromError(Errors error) {
-        return new ApiVersionsResponse(error.code(), Collections.<ApiVersion>emptyList());
+        return new ApiVersionsResponse(error, Collections.<ApiVersion>emptyList());
     }
 
     private static ApiVersionsResponse createApiVersionsResponse() {
@@ -121,7 +121,7 @@ public class ApiVersionsResponse extends AbstractResponse {
         for (ApiKeys apiKey : ApiKeys.values()) {
             versionList.add(new ApiVersion(apiKey.id, ProtoUtils.oldestVersion(apiKey.id), ProtoUtils.latestVersion(apiKey.id)));
         }
-        return new ApiVersionsResponse(Errors.NONE.code(), versionList);
+        return new ApiVersionsResponse(Errors.NONE, versionList);
     }
 
     private Map<Short, ApiVersion> buildApiKeyToApiVersion(List<ApiVersion> apiVersions) {

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 337ccfc..8f44e5c 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
@@ -68,7 +68,7 @@ public class ControlledShutdownRequest extends AbstractRequest {
                 throw new IllegalArgumentException("Version 0 is not supported. It is only supported by " +
                         "the Scala request class for controlled shutdown");
             case 1:
-                return new ControlledShutdownResponse(Errors.forException(e).code(), Collections.<TopicPartition>emptySet());
+                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)));

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 1996f82..b3922f9 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
@@ -14,6 +14,7 @@ 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.Schema;
 import org.apache.kafka.common.protocol.types.Struct;
@@ -41,14 +42,14 @@ public class ControlledShutdownResponse extends AbstractResponse {
      * BROKER_NOT_AVAILABLE(8)
      * STALE_CONTROLLER_EPOCH(11)
      */
-    private final short errorCode;
+    private final Errors error;
 
     private final Set<TopicPartition> partitionsRemaining;
 
-    public ControlledShutdownResponse(short errorCode, Set<TopicPartition> partitionsRemaining) {
+    public ControlledShutdownResponse(Errors error, Set<TopicPartition> partitionsRemaining) {
         super(new Struct(CURRENT_SCHEMA));
 
-        struct.set(ERROR_CODE_KEY_NAME, errorCode);
+        struct.set(ERROR_CODE_KEY_NAME, error.code());
 
         List<Struct> partitionsRemainingList = new ArrayList<>(partitionsRemaining.size());
         for (TopicPartition topicPartition : partitionsRemaining) {
@@ -59,13 +60,13 @@ public class ControlledShutdownResponse extends AbstractResponse {
         }
         struct.set(PARTITIONS_REMAINING_KEY_NAME, partitionsRemainingList.toArray());
 
-        this.errorCode = errorCode;
+        this.error = error;
         this.partitionsRemaining = partitionsRemaining;
     }
 
     public ControlledShutdownResponse(Struct struct) {
         super(struct);
-        errorCode = struct.getShort(ERROR_CODE_KEY_NAME);
+        error = Errors.forCode(struct.getShort(ERROR_CODE_KEY_NAME));
         Set<TopicPartition> partitions = new HashSet<>();
         for (Object topicPartitionObj : struct.getArray(PARTITIONS_REMAINING_KEY_NAME)) {
             Struct topicPartition = (Struct) topicPartitionObj;
@@ -76,8 +77,8 @@ public class ControlledShutdownResponse extends AbstractResponse {
         partitionsRemaining = partitions;
     }
 
-    public short errorCode() {
-        return errorCode;
+    public Errors error() {
+        return error;
     }
 
     public Set<TopicPartition> partitionsRemaining() {

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 9807283..b09795f 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
@@ -113,11 +113,11 @@ public class CreateTopicsResponse extends AbstractResponse {
         for (Object topicErrorStructObj : topicErrorStructs) {
             Struct topicErrorCodeStruct = (Struct) topicErrorStructObj;
             String topic = topicErrorCodeStruct.getString(TOPIC_KEY_NAME);
-            short errorCode = topicErrorCodeStruct.getShort(ERROR_CODE_KEY_NAME);
+            Errors error = Errors.forCode(topicErrorCodeStruct.getShort(ERROR_CODE_KEY_NAME));
             String errorMessage = null;
             if (topicErrorCodeStruct.hasField(ERROR_MESSAGE_KEY_NAME))
                 errorMessage = topicErrorCodeStruct.getString(ERROR_MESSAGE_KEY_NAME);
-            errors.put(topic, new Error(Errors.forCode(errorCode), errorMessage));
+            errors.put(topic, new Error(error, errorMessage));
         }
 
         this.errors = errors;

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 ed6a63d..5c8b3d5 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
@@ -67,8 +67,8 @@ public class DeleteTopicsResponse extends AbstractResponse {
         for (Object topicErrorCodeStructObj : topicErrorCodesStructs) {
             Struct topicErrorCodeStruct = (Struct) topicErrorCodeStructObj;
             String topic = topicErrorCodeStruct.getString(TOPIC_KEY_NAME);
-            short errorCode = topicErrorCodeStruct.getShort(ERROR_CODE_KEY_NAME);
-            errors.put(topic, Errors.forCode(errorCode));
+            Errors error = Errors.forCode(topicErrorCodeStruct.getShort(ERROR_CODE_KEY_NAME));
+            errors.put(topic, error);
         }
 
         this.errors = errors;

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 2eff628..56b387e 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
@@ -67,7 +67,7 @@ public class DescribeGroupsResponse extends AbstractResponse {
             Struct groupStruct = struct.instance(GROUPS_KEY_NAME);
             GroupMetadata group = groupEntry.getValue();
             groupStruct.set(GROUP_ID_KEY_NAME, groupEntry.getKey());
-            groupStruct.set(ERROR_CODE_KEY_NAME, group.errorCode);
+            groupStruct.set(ERROR_CODE_KEY_NAME, group.error.code());
             groupStruct.set(GROUP_STATE_KEY_NAME, group.state);
             groupStruct.set(PROTOCOL_TYPE_KEY_NAME, group.protocolType);
             groupStruct.set(PROTOCOL_KEY_NAME, group.protocol);
@@ -95,7 +95,7 @@ public class DescribeGroupsResponse extends AbstractResponse {
             Struct groupStruct = (Struct) groupObj;
 
             String groupId = groupStruct.getString(GROUP_ID_KEY_NAME);
-            short errorCode = groupStruct.getShort(ERROR_CODE_KEY_NAME);
+            Errors error = Errors.forCode(groupStruct.getShort(ERROR_CODE_KEY_NAME));
             String state = groupStruct.getString(GROUP_STATE_KEY_NAME);
             String protocolType = groupStruct.getString(PROTOCOL_TYPE_KEY_NAME);
             String protocol = groupStruct.getString(PROTOCOL_KEY_NAME);
@@ -111,7 +111,7 @@ public class DescribeGroupsResponse extends AbstractResponse {
                 members.add(new GroupMember(memberId, clientId, clientHost,
                         memberMetadata, memberAssignment));
             }
-            this.groups.put(groupId, new GroupMetadata(errorCode, state, protocolType, protocol, members));
+            this.groups.put(groupId, new GroupMetadata(error, state, protocolType, protocol, members));
         }
     }
 
@@ -121,26 +121,26 @@ public class DescribeGroupsResponse extends AbstractResponse {
 
 
     public static class GroupMetadata {
-        private final short errorCode;
+        private final Errors error;
         private final String state;
         private final String protocolType;
         private final String protocol;
         private final List<GroupMember> members;
 
-        public GroupMetadata(short errorCode,
+        public GroupMetadata(Errors error,
                              String state,
                              String protocolType,
                              String protocol,
                              List<GroupMember> members) {
-            this.errorCode = errorCode;
+            this.error = error;
             this.state = state;
             this.protocolType = protocolType;
             this.protocol = protocol;
             this.members = members;
         }
 
-        public short errorCode() {
-            return errorCode;
+        public Errors error() {
+            return error;
         }
 
         public String state() {
@@ -161,7 +161,7 @@ public class DescribeGroupsResponse extends AbstractResponse {
 
         public static GroupMetadata forError(Errors error) {
             return new DescribeGroupsResponse.GroupMetadata(
-                    error.code(),
+                    error,
                     DescribeGroupsResponse.UNKNOWN_STATE,
                     DescribeGroupsResponse.UNKNOWN_PROTOCOL_TYPE,
                     DescribeGroupsResponse.UNKNOWN_PROTOCOL,

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 5700e9e..4f270e1 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
@@ -212,7 +212,7 @@ public class FetchRequest extends AbstractRequest {
         LinkedHashMap<TopicPartition, FetchResponse.PartitionData> responseData = new LinkedHashMap<>();
 
         for (Map.Entry<TopicPartition, PartitionData> entry: fetchData.entrySet()) {
-            FetchResponse.PartitionData partitionResponse = new FetchResponse.PartitionData(Errors.forException(e).code(),
+            FetchResponse.PartitionData partitionResponse = new FetchResponse.PartitionData(Errors.forException(e),
                     FetchResponse.INVALID_HIGHWATERMARK, MemoryRecords.EMPTY);
 
             responseData.put(entry.getKey(), partitionResponse);

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 965b207..64bd3d3 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
@@ -21,6 +21,7 @@ import org.apache.kafka.common.network.ByteBufferSend;
 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.Schema;
 import org.apache.kafka.common.protocol.types.Struct;
@@ -37,7 +38,7 @@ import java.util.Map;
  * This wrapper supports all versions of the Fetch API
  */
 public class FetchResponse extends AbstractResponse {
-    
+
     private static final Schema CURRENT_SCHEMA = ProtoUtils.currentResponseSchema(ApiKeys.FETCH.id);
     private static final String RESPONSES_KEY_NAME = "responses";
 
@@ -73,19 +74,19 @@ public class FetchResponse extends AbstractResponse {
     private final int throttleTime;
 
     public static final class PartitionData {
-        public final short errorCode;
+        public final Errors error;
         public final long highWatermark;
         public final Records records;
 
-        public PartitionData(short errorCode, long highWatermark, Records records) {
-            this.errorCode = errorCode;
+        public PartitionData(Errors error, long highWatermark, Records records) {
+            this.error = error;
             this.highWatermark = highWatermark;
             this.records = records;
         }
 
         @Override
         public String toString() {
-            return "(errorCode=" + errorCode + ", highWaterMark=" + highWatermark +
+            return "(error=" + error.toString() + ", highWaterMark=" + highWatermark +
                     ", records=" + records + ")";
         }
     }
@@ -128,10 +129,10 @@ public class FetchResponse extends AbstractResponse {
                 Struct partitionResponse = (Struct) partitionResponseObj;
                 Struct partitionResponseHeader = partitionResponse.getStruct(PARTITION_HEADER_KEY_NAME);
                 int partition = partitionResponseHeader.getInt(PARTITION_KEY_NAME);
-                short errorCode = partitionResponseHeader.getShort(ERROR_CODE_KEY_NAME);
+                Errors error = Errors.forCode(partitionResponseHeader.getShort(ERROR_CODE_KEY_NAME));
                 long highWatermark = partitionResponseHeader.getLong(HIGH_WATERMARK_KEY_NAME);
                 Records records = partitionResponse.getRecords(RECORD_SET_KEY_NAME);
-                PartitionData partitionData = new PartitionData(errorCode, highWatermark, records);
+                PartitionData partitionData = new PartitionData(error, highWatermark, records);
                 responseData.put(new TopicPartition(topic, partition), partitionData);
             }
         }
@@ -237,7 +238,7 @@ public class FetchResponse extends AbstractResponse {
                 Struct partitionData = topicData.instance(PARTITIONS_KEY_NAME);
                 Struct partitionDataHeader = partitionData.instance(PARTITION_HEADER_KEY_NAME);
                 partitionDataHeader.set(PARTITION_KEY_NAME, partitionEntry.getKey());
-                partitionDataHeader.set(ERROR_CODE_KEY_NAME, fetchPartitionData.errorCode);
+                partitionDataHeader.set(ERROR_CODE_KEY_NAME, fetchPartitionData.error.code());
                 partitionDataHeader.set(HIGH_WATERMARK_KEY_NAME, fetchPartitionData.highWatermark);
                 partitionData.set(PARTITION_HEADER_KEY_NAME, partitionDataHeader);
                 partitionData.set(RECORD_SET_KEY_NAME, fetchPartitionData.records);

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 d8ccdf6..ed56f39 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
@@ -65,7 +65,7 @@ public class GroupCoordinatorRequest extends AbstractRequest {
         short versionId = version();
         switch (versionId) {
             case 0:
-                return new GroupCoordinatorResponse(Errors.GROUP_COORDINATOR_NOT_AVAILABLE.code(), Node.noNode());
+                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)));

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 1f447f7..fc3d358 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
@@ -14,6 +14,7 @@ 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.Schema;
 import org.apache.kafka.common.protocol.types.Struct;
@@ -21,7 +22,7 @@ import org.apache.kafka.common.protocol.types.Struct;
 import java.nio.ByteBuffer;
 
 public class GroupCoordinatorResponse extends AbstractResponse {
-    
+
     private static final Schema CURRENT_SCHEMA = ProtoUtils.currentResponseSchema(ApiKeys.GROUP_COORDINATOR.id);
     private static final String ERROR_CODE_KEY_NAME = "error_code";
     private static final String COORDINATOR_KEY_NAME = "coordinator";
@@ -40,24 +41,24 @@ public class GroupCoordinatorResponse extends AbstractResponse {
     private static final String HOST_KEY_NAME = "host";
     private static final String PORT_KEY_NAME = "port";
 
-    private final short errorCode;
+    private final Errors error;
     private final Node node;
 
-    public GroupCoordinatorResponse(short errorCode, Node node) {
+    public GroupCoordinatorResponse(Errors error, Node node) {
         super(new Struct(CURRENT_SCHEMA));
-        struct.set(ERROR_CODE_KEY_NAME, errorCode);
+        struct.set(ERROR_CODE_KEY_NAME, error.code());
         Struct coordinator = struct.instance(COORDINATOR_KEY_NAME);
         coordinator.set(NODE_ID_KEY_NAME, node.id());
         coordinator.set(HOST_KEY_NAME, node.host());
         coordinator.set(PORT_KEY_NAME, node.port());
         struct.set(COORDINATOR_KEY_NAME, coordinator);
-        this.errorCode = errorCode;
+        this.error = error;
         this.node = node;
     }
 
     public GroupCoordinatorResponse(Struct struct) {
         super(struct);
-        errorCode = struct.getShort(ERROR_CODE_KEY_NAME);
+        error = Errors.forCode(struct.getShort(ERROR_CODE_KEY_NAME));
         Struct broker = (Struct) struct.get(COORDINATOR_KEY_NAME);
         int nodeId = broker.getInt(NODE_ID_KEY_NAME);
         String host = broker.getString(HOST_KEY_NAME);
@@ -65,8 +66,8 @@ public class GroupCoordinatorResponse extends AbstractResponse {
         node = new Node(nodeId, host, port);
     }
 
-    public short errorCode() {
-        return errorCode;
+    public Errors error() {
+        return error;
     }
 
     public Node node() {

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 0e5c17a..7e79c8a 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
@@ -80,7 +80,7 @@ public class HeartbeatRequest extends AbstractRequest {
         short versionId = version();
         switch (versionId) {
             case 0:
-                return new HeartbeatResponse(Errors.forException(e).code());
+                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)));

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 72f0175..f36dec4 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
@@ -13,6 +13,7 @@
 package org.apache.kafka.common.requests;
 
 import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.protocol.ProtoUtils;
 import org.apache.kafka.common.protocol.types.Schema;
 import org.apache.kafka.common.protocol.types.Struct;
@@ -20,7 +21,7 @@ import org.apache.kafka.common.protocol.types.Struct;
 import java.nio.ByteBuffer;
 
 public class HeartbeatResponse extends AbstractResponse {
-    
+
     private static final Schema CURRENT_SCHEMA = ProtoUtils.currentResponseSchema(ApiKeys.HEARTBEAT.id);
     private static final String ERROR_CODE_KEY_NAME = "error_code";
 
@@ -35,20 +36,21 @@ public class HeartbeatResponse extends AbstractResponse {
      * GROUP_AUTHORIZATION_FAILED (30)
      */
 
-    private final short errorCode;
-    public HeartbeatResponse(short errorCode) {
+    private final Errors error;
+
+    public HeartbeatResponse(Errors error) {
         super(new Struct(CURRENT_SCHEMA));
-        struct.set(ERROR_CODE_KEY_NAME, errorCode);
-        this.errorCode = errorCode;
+        struct.set(ERROR_CODE_KEY_NAME, error.code());
+        this.error = error;
     }
 
     public HeartbeatResponse(Struct struct) {
         super(struct);
-        errorCode = struct.getShort(ERROR_CODE_KEY_NAME);
+        error = Errors.forCode(struct.getShort(ERROR_CODE_KEY_NAME));
     }
 
-    public short errorCode() {
-        return errorCode;
+    public Errors error() {
+        return error;
     }
 
     public static HeartbeatResponse parse(ByteBuffer buffer) {

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 3f00ed1..ad0cdd0 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
@@ -170,7 +170,7 @@ public class JoinGroupRequest extends AbstractRequest {
             case 1:
                 return new JoinGroupResponse(
                         versionId,
-                        Errors.forException(e).code(),
+                        Errors.forException(e),
                         JoinGroupResponse.UNKNOWN_GENERATION_ID,
                         JoinGroupResponse.UNKNOWN_PROTOCOL,
                         JoinGroupResponse.UNKNOWN_MEMBER_ID, // memberId

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 8fd77ce..bc9366a 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
@@ -13,6 +13,7 @@
 package org.apache.kafka.common.requests;
 
 import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.protocol.ProtoUtils;
 import org.apache.kafka.common.protocol.types.Schema;
 import org.apache.kafka.common.protocol.types.Struct;
@@ -53,24 +54,24 @@ public class JoinGroupResponse extends AbstractResponse {
     public static final int UNKNOWN_GENERATION_ID = -1;
     public static final String UNKNOWN_MEMBER_ID = "";
 
-    private final short errorCode;
+    private final Errors error;
     private final int generationId;
     private final String groupProtocol;
     private final String memberId;
     private final String leaderId;
     private final Map<String, ByteBuffer> members;
 
-    public JoinGroupResponse(short errorCode,
+    public JoinGroupResponse(Errors error,
                              int generationId,
                              String groupProtocol,
                              String memberId,
                              String leaderId,
                              Map<String, ByteBuffer> groupMembers) {
-        this(CURRENT_VERSION, errorCode, generationId, groupProtocol, memberId, leaderId, groupMembers);
+        this(CURRENT_VERSION, error, generationId, groupProtocol, memberId, leaderId, groupMembers);
     }
 
     public JoinGroupResponse(int version,
-                             short errorCode,
+                             Errors error,
                              int generationId,
                              String groupProtocol,
                              String memberId,
@@ -78,7 +79,7 @@ public class JoinGroupResponse extends AbstractResponse {
                              Map<String, ByteBuffer> groupMembers) {
         super(new Struct(ProtoUtils.responseSchema(ApiKeys.JOIN_GROUP.id, version)));
 
-        struct.set(ERROR_CODE_KEY_NAME, errorCode);
+        struct.set(ERROR_CODE_KEY_NAME, error.code());
         struct.set(GENERATION_ID_KEY_NAME, generationId);
         struct.set(GROUP_PROTOCOL_KEY_NAME, groupProtocol);
         struct.set(MEMBER_ID_KEY_NAME, memberId);
@@ -93,7 +94,7 @@ public class JoinGroupResponse extends AbstractResponse {
         }
         struct.set(MEMBERS_KEY_NAME, memberArray.toArray());
 
-        this.errorCode = errorCode;
+        this.error = error;
         this.generationId = generationId;
         this.groupProtocol = groupProtocol;
         this.memberId = memberId;
@@ -111,15 +112,15 @@ public class JoinGroupResponse extends AbstractResponse {
             ByteBuffer memberMetadata = memberData.getBytes(MEMBER_METADATA_KEY_NAME);
             members.put(memberId, memberMetadata);
         }
-        errorCode = struct.getShort(ERROR_CODE_KEY_NAME);
+        error = Errors.forCode(struct.getShort(ERROR_CODE_KEY_NAME));
         generationId = struct.getInt(GENERATION_ID_KEY_NAME);
         groupProtocol = struct.getString(GROUP_PROTOCOL_KEY_NAME);
         memberId = struct.getString(MEMBER_ID_KEY_NAME);
         leaderId = struct.getString(LEADER_ID_KEY_NAME);
     }
 
-    public short errorCode() {
-        return errorCode;
+    public Errors error() {
+        return error;
     }
 
     public int generationId() {

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 1f09a12..fde184a 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
@@ -178,15 +178,15 @@ public class LeaderAndIsrRequest extends AbstractRequest {
 
     @Override
     public AbstractResponse getErrorResponse(Throwable e) {
-        Map<TopicPartition, Short> responses = new HashMap<>(partitionStates.size());
+        Map<TopicPartition, Errors> responses = new HashMap<>(partitionStates.size());
         for (TopicPartition partition : partitionStates.keySet()) {
-            responses.put(partition, Errors.forException(e).code());
+            responses.put(partition, Errors.forException(e));
         }
 
         short versionId = version();
         switch (versionId) {
             case 0:
-                return new LeaderAndIsrResponse(Errors.NONE.code(), responses);
+                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)));

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 a754def..4d0a05d 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
@@ -41,34 +41,32 @@ public class LeaderAndIsrResponse extends AbstractResponse {
      *
      * STALE_CONTROLLER_EPOCH (11)
      */
-    private final short errorCode;
+    private final Errors error;
 
-    private final Map<TopicPartition, Short> responses;
+    private final Map<TopicPartition, Errors> responses;
 
-    public LeaderAndIsrResponse(Map<TopicPartition, Short> responses) {
-        this(Errors.NONE.code(), responses);
+    public LeaderAndIsrResponse(Map<TopicPartition, Errors> responses) {
+        this(Errors.NONE, responses);
     }
 
-    public LeaderAndIsrResponse(short errorCode, Map<TopicPartition, Short> responses) {
+    public LeaderAndIsrResponse(Errors error, Map<TopicPartition, Errors> responses) {
         super(new Struct(CURRENT_SCHEMA));
 
-        struct.set(ERROR_CODE_KEY_NAME, errorCode);
-
         List<Struct> responseDatas = new ArrayList<>(responses.size());
-        for (Map.Entry<TopicPartition, Short> response : responses.entrySet()) {
+        for (Map.Entry<TopicPartition, Errors> response : responses.entrySet()) {
             Struct partitionData = struct.instance(PARTITIONS_KEY_NAME);
             TopicPartition partition = response.getKey();
             partitionData.set(PARTITIONS_TOPIC_KEY_NAME, partition.topic());
             partitionData.set(PARTITIONS_PARTITION_KEY_NAME, partition.partition());
-            partitionData.set(PARTITIONS_ERROR_CODE_KEY_NAME, response.getValue());
+            partitionData.set(PARTITIONS_ERROR_CODE_KEY_NAME, response.getValue().code());
             responseDatas.add(partitionData);
         }
 
         struct.set(PARTITIONS_KEY_NAME, responseDatas.toArray());
-        struct.set(ERROR_CODE_KEY_NAME, errorCode);
+        struct.set(ERROR_CODE_KEY_NAME, error.code());
 
         this.responses = responses;
-        this.errorCode = errorCode;
+        this.error = error;
     }
 
     public LeaderAndIsrResponse(Struct struct) {
@@ -79,19 +77,19 @@ public class LeaderAndIsrResponse extends AbstractResponse {
             Struct responseData = (Struct) responseDataObj;
             String topic = responseData.getString(PARTITIONS_TOPIC_KEY_NAME);
             int partition = responseData.getInt(PARTITIONS_PARTITION_KEY_NAME);
-            short errorCode = responseData.getShort(PARTITIONS_ERROR_CODE_KEY_NAME);
-            responses.put(new TopicPartition(topic, partition), errorCode);
+            Errors error = Errors.forCode(responseData.getShort(PARTITIONS_ERROR_CODE_KEY_NAME));
+            responses.put(new TopicPartition(topic, partition), error);
         }
 
-        errorCode = struct.getShort(ERROR_CODE_KEY_NAME);
+        error = Errors.forCode(struct.getShort(ERROR_CODE_KEY_NAME));
     }
 
-    public Map<TopicPartition, Short> responses() {
+    public Map<TopicPartition, Errors> responses() {
         return responses;
     }
 
-    public short errorCode() {
-        return errorCode;
+    public Errors error() {
+        return error;
     }
 
     public static LeaderAndIsrResponse parse(ByteBuffer buffer, int version) {

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 573ebc8..2a7b70e 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
@@ -71,7 +71,7 @@ public class LeaveGroupRequest extends AbstractRequest {
         short versionId = version();
         switch (versionId) {
             case 0:
-                return new LeaveGroupResponse(Errors.forException(e).code());
+                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)));

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 9c7998b..bd1c84d 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,6 +14,7 @@ package org.apache.kafka.common.requests;
 
 import java.nio.ByteBuffer;
 import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.protocol.ProtoUtils;
 import org.apache.kafka.common.protocol.types.Schema;
 import org.apache.kafka.common.protocol.types.Struct;
@@ -33,20 +34,20 @@ public class LeaveGroupResponse extends AbstractResponse {
      * GROUP_AUTHORIZATION_FAILED (30)
      */
 
-    private final short errorCode;
-    public LeaveGroupResponse(short errorCode) {
+    private final Errors error;
+    public LeaveGroupResponse(Errors error) {
         super(new Struct(CURRENT_SCHEMA));
-        struct.set(ERROR_CODE_KEY_NAME, errorCode);
-        this.errorCode = errorCode;
+        struct.set(ERROR_CODE_KEY_NAME, error.code());
+        this.error = error;
     }
 
     public LeaveGroupResponse(Struct struct) {
         super(struct);
-        errorCode = struct.getShort(ERROR_CODE_KEY_NAME);
+        error = Errors.forCode(struct.getShort(ERROR_CODE_KEY_NAME));
     }
 
-    public short errorCode() {
-        return errorCode;
+    public Errors error() {
+        return error;
     }
 
     public static LeaveGroupResponse parse(ByteBuffer buffer) {

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 8d0a1af..235f4e4 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
@@ -51,8 +51,7 @@ public class ListGroupsRequest extends AbstractRequest {
         short versionId = version();
         switch (versionId) {
             case 0:
-                short errorCode = Errors.forException(e).code();
-                return new ListGroupsResponse(errorCode, Collections.<ListGroupsResponse.Group>emptyList());
+                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)));

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 98573f8..f421064 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
@@ -39,12 +39,12 @@ public class ListGroupsResponse extends AbstractResponse {
      * AUTHORIZATION_FAILED (29)
      */
 
-    private final short errorCode;
+    private final Errors error;
     private final List<Group> groups;
 
-    public ListGroupsResponse(short errorCode, List<Group> groups) {
+    public ListGroupsResponse(Errors error, List<Group> groups) {
         super(new Struct(CURRENT_SCHEMA));
-        struct.set(ERROR_CODE_KEY_NAME, errorCode);
+        struct.set(ERROR_CODE_KEY_NAME, error.code());
         List<Struct> groupList = new ArrayList<>();
         for (Group group : groups) {
             Struct groupStruct = struct.instance(GROUPS_KEY_NAME);
@@ -53,13 +53,13 @@ public class ListGroupsResponse extends AbstractResponse {
             groupList.add(groupStruct);
         }
         struct.set(GROUPS_KEY_NAME, groupList.toArray());
-        this.errorCode = errorCode;
+        this.error = error;
         this.groups = groups;
     }
 
     public ListGroupsResponse(Struct struct) {
         super(struct);
-        this.errorCode = struct.getShort(ERROR_CODE_KEY_NAME);
+        this.error = Errors.forCode(struct.getShort(ERROR_CODE_KEY_NAME));
         this.groups = new ArrayList<>();
         for (Object groupObj : struct.getArray(GROUPS_KEY_NAME)) {
             Struct groupStruct = (Struct) groupObj;
@@ -73,8 +73,8 @@ public class ListGroupsResponse extends AbstractResponse {
         return groups;
     }
 
-    public short errorCode() {
-        return errorCode;
+    public Errors error() {
+        return error;
     }
 
     public static class Group {
@@ -101,7 +101,7 @@ public class ListGroupsResponse extends AbstractResponse {
     }
 
     public static ListGroupsResponse fromError(Errors error) {
-        return new ListGroupsResponse(error.code(), Collections.<Group>emptyList());
+        return new ListGroupsResponse(error, Collections.<Group>emptyList());
     }
 
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 79251ed..6214a56 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
@@ -241,12 +241,12 @@ public class ListOffsetRequest extends AbstractRequest {
         short versionId = version();
         if (versionId == 0) {
             for (Map.Entry<TopicPartition, PartitionData> entry : offsetData.entrySet()) {
-                ListOffsetResponse.PartitionData partitionResponse = new ListOffsetResponse.PartitionData(Errors.forException(e).code(), new ArrayList<Long>());
+                ListOffsetResponse.PartitionData partitionResponse = new ListOffsetResponse.PartitionData(Errors.forException(e), new ArrayList<Long>());
                 responseData.put(entry.getKey(), partitionResponse);
             }
         } else {
             for (Map.Entry<TopicPartition, Long> entry : partitionTimestamps.entrySet()) {
-                ListOffsetResponse.PartitionData partitionResponse = new ListOffsetResponse.PartitionData(Errors.forException(e).code(), -1L, -1L);
+                ListOffsetResponse.PartitionData partitionResponse = new ListOffsetResponse.PartitionData(Errors.forException(e), -1L, -1L);
                 responseData.put(entry.getKey(), partitionResponse);
             }
         }

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 2eddf1e..b815a53 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
@@ -18,6 +18,7 @@ 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.Schema;
 import org.apache.kafka.common.protocol.types.Struct;
@@ -33,7 +34,7 @@ import java.util.Map;
 public class ListOffsetResponse extends AbstractResponse {
     public static final long UNKNOWN_TIMESTAMP = -1L;
     public static final long UNKNOWN_OFFSET = -1L;
-    
+
     private static final Schema CURRENT_SCHEMA = ProtoUtils.currentResponseSchema(ApiKeys.LIST_OFFSETS.id);
     private static final String RESPONSES_KEY_NAME = "responses";
 
@@ -63,7 +64,7 @@ public class ListOffsetResponse extends AbstractResponse {
     private final Map<TopicPartition, PartitionData> responseData;
 
     public static final class PartitionData {
-        public final short errorCode;
+        public final Errors error;
         // The offsets list is only used in ListOffsetResponse v0.
         @Deprecated
         public final List<Long> offsets;
@@ -74,8 +75,8 @@ public class ListOffsetResponse extends AbstractResponse {
          * Constructor for ListOffsetResponse v0
          */
         @Deprecated
-        public PartitionData(short errorCode, List<Long> offsets) {
-            this.errorCode = errorCode;
+        public PartitionData(Errors error, List<Long> offsets) {
+            this.error = error;
             this.offsets = offsets;
             this.timestamp = null;
             this.offset = null;
@@ -84,8 +85,8 @@ public class ListOffsetResponse extends AbstractResponse {
         /**
          * Constructor for ListOffsetResponse v1
          */
-        public PartitionData(short errorCode, long timestamp, long offset) {
-            this.errorCode = errorCode;
+        public PartitionData(Errors error, long timestamp, long offset) {
+            this.error = error;
             this.timestamp = timestamp;
             this.offset = offset;
             this.offsets = null;
@@ -95,7 +96,7 @@ public class ListOffsetResponse extends AbstractResponse {
         public String toString() {
             StringBuilder bld = new StringBuilder();
             bld.append("PartitionData{").
-                append("errorCode: ").append((int) errorCode).
+                append("errorCode: ").append((int) error.code()).
                 append(", timestamp: ").append(timestamp).
                 append(", offset: ").append(offset).
                 append(", offsets: ");
@@ -130,7 +131,7 @@ public class ListOffsetResponse extends AbstractResponse {
                 PartitionData offsetPartitionData = partitionEntry.getValue();
                 Struct partitionData = topicData.instance(PARTITIONS_KEY_NAME);
                 partitionData.set(PARTITION_KEY_NAME, partitionEntry.getKey());
-                partitionData.set(ERROR_CODE_KEY_NAME, offsetPartitionData.errorCode);
+                partitionData.set(ERROR_CODE_KEY_NAME, offsetPartitionData.error.code());
                 if (version == 0)
                     partitionData.set(OFFSETS_KEY_NAME, offsetPartitionData.offsets.toArray());
                 else {
@@ -155,18 +156,18 @@ public class ListOffsetResponse extends AbstractResponse {
             for (Object partitionResponseObj : topicResponse.getArray(PARTITIONS_KEY_NAME)) {
                 Struct partitionResponse = (Struct) partitionResponseObj;
                 int partition = partitionResponse.getInt(PARTITION_KEY_NAME);
-                short errorCode = partitionResponse.getShort(ERROR_CODE_KEY_NAME);
+                Errors error = Errors.forCode(partitionResponse.getShort(ERROR_CODE_KEY_NAME));
                 PartitionData partitionData;
                 if (partitionResponse.hasField(OFFSETS_KEY_NAME)) {
                     Object[] offsets = partitionResponse.getArray(OFFSETS_KEY_NAME);
                     List<Long> offsetsList = new ArrayList<Long>();
                     for (Object offset : offsets)
                         offsetsList.add((Long) offset);
-                    partitionData = new PartitionData(errorCode, offsetsList);
+                    partitionData = new PartitionData(error, offsetsList);
                 } else {
                     long timestamp = partitionResponse.getLong(TIMESTAMP_KEY_NAME);
                     long offset = partitionResponse.getLong(OFFSET_KEY_NAME);
-                    partitionData = new PartitionData(errorCode, timestamp, offset);
+                    partitionData = new PartitionData(error, timestamp, offset);
                 }
                 responseData.put(new TopicPartition(topic, partition), partitionData);
             }

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 361fd15..6dd1197 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
@@ -3,9 +3,9 @@
  * 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.
@@ -285,9 +285,9 @@ public class OffsetCommitRequest extends AbstractRequest {
 
     @Override
     public AbstractResponse getErrorResponse(Throwable e) {
-        Map<TopicPartition, Short> responseData = new HashMap<>();
+        Map<TopicPartition, Errors> responseData = new HashMap<>();
         for (Map.Entry<TopicPartition, PartitionData> entry: offsetData.entrySet()) {
-            responseData.put(entry.getKey(), Errors.forException(e).code());
+            responseData.put(entry.getKey(), Errors.forException(e));
         }
 
         short versionId = version();

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 abb260e..8a00c6b 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
@@ -14,6 +14,7 @@ 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.Schema;
 import org.apache.kafka.common.protocol.types.Struct;
@@ -26,7 +27,7 @@ import java.util.List;
 import java.util.Map;
 
 public class OffsetCommitResponse extends AbstractResponse {
-    
+
     private static final Schema CURRENT_SCHEMA = ProtoUtils.currentResponseSchema(ApiKeys.OFFSET_COMMIT.id);
     private static final String RESPONSES_KEY_NAME = "responses";
 
@@ -54,22 +55,22 @@ public class OffsetCommitResponse extends AbstractResponse {
      * GROUP_AUTHORIZATION_FAILED (30)
      */
 
-    private final Map<TopicPartition, Short> responseData;
+    private final Map<TopicPartition, Errors> responseData;
 
-    public OffsetCommitResponse(Map<TopicPartition, Short> responseData) {
+    public OffsetCommitResponse(Map<TopicPartition, Errors> responseData) {
         super(new Struct(CURRENT_SCHEMA));
 
-        Map<String, Map<Integer, Short>> topicsData = CollectionUtils.groupDataByTopic(responseData);
+        Map<String, Map<Integer, Errors>> topicsData = CollectionUtils.groupDataByTopic(responseData);
 
         List<Struct> topicArray = new ArrayList<Struct>();
-        for (Map.Entry<String, Map<Integer, Short>> entries: topicsData.entrySet()) {
+        for (Map.Entry<String, Map<Integer, Errors>> entries: topicsData.entrySet()) {
             Struct topicData = struct.instance(RESPONSES_KEY_NAME);
             topicData.set(TOPIC_KEY_NAME, entries.getKey());
             List<Struct> partitionArray = new ArrayList<Struct>();
-            for (Map.Entry<Integer, Short> partitionEntry : entries.getValue().entrySet()) {
+            for (Map.Entry<Integer, Errors> partitionEntry : entries.getValue().entrySet()) {
                 Struct partitionData = topicData.instance(PARTITIONS_KEY_NAME);
                 partitionData.set(PARTITION_KEY_NAME, partitionEntry.getKey());
-                partitionData.set(ERROR_CODE_KEY_NAME, partitionEntry.getValue());
+                partitionData.set(ERROR_CODE_KEY_NAME, partitionEntry.getValue().code());
                 partitionArray.add(partitionData);
             }
             topicData.set(PARTITIONS_KEY_NAME, partitionArray.toArray());
@@ -81,20 +82,20 @@ public class OffsetCommitResponse extends AbstractResponse {
 
     public OffsetCommitResponse(Struct struct) {
         super(struct);
-        responseData = new HashMap<TopicPartition, Short>();
+        responseData = new HashMap<TopicPartition, Errors>();
         for (Object topicResponseObj : struct.getArray(RESPONSES_KEY_NAME)) {
             Struct topicResponse = (Struct) topicResponseObj;
             String topic = topicResponse.getString(TOPIC_KEY_NAME);
             for (Object partitionResponseObj : topicResponse.getArray(PARTITIONS_KEY_NAME)) {
                 Struct partitionResponse = (Struct) partitionResponseObj;
                 int partition = partitionResponse.getInt(PARTITION_KEY_NAME);
-                short errorCode = partitionResponse.getShort(ERROR_CODE_KEY_NAME);
-                responseData.put(new TopicPartition(topic, partition), errorCode);
+                Errors error = Errors.forCode(partitionResponse.getShort(ERROR_CODE_KEY_NAME));
+                responseData.put(new TopicPartition(topic, partition), error);
             }
         }
     }
 
-    public Map<TopicPartition, Short> responseData() {
+    public Map<TopicPartition, Errors> responseData() {
         return responseData;
     }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 81bc249..d244f0a 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
@@ -66,7 +66,7 @@ public class SaslHandshakeRequest extends AbstractRequest {
         switch (versionId) {
             case 0:
                 List<String> enabledMechanisms = Collections.emptyList();
-                return new SaslHandshakeResponse(Errors.forException(e).code(), enabledMechanisms);
+                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)));

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 6d7f734..f50c5be 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
@@ -24,6 +24,7 @@ import java.util.Collection;
 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.Schema;
 import org.apache.kafka.common.protocol.types.Struct;
@@ -45,20 +46,20 @@ public class SaslHandshakeResponse extends AbstractResponse {
      *   UNSUPPORTED_SASL_MECHANISM(33): Client mechanism not enabled in server
      *   ILLEGAL_SASL_STATE(34) : Invalid request during SASL handshake
      */
-    private final short errorCode;
+    private final Errors error;
     private final List<String> enabledMechanisms;
 
-    public SaslHandshakeResponse(short errorCode, Collection<String> enabledMechanisms) {
+    public SaslHandshakeResponse(Errors error, Collection<String> enabledMechanisms) {
         super(new Struct(CURRENT_SCHEMA));
-        struct.set(ERROR_CODE_KEY_NAME, errorCode);
+        struct.set(ERROR_CODE_KEY_NAME, error.code());
         struct.set(ENABLED_MECHANISMS_KEY_NAME, enabledMechanisms.toArray());
-        this.errorCode = errorCode;
+        this.error = error;
         this.enabledMechanisms = new ArrayList<>(enabledMechanisms);
     }
 
     public SaslHandshakeResponse(Struct struct) {
         super(struct);
-        errorCode = struct.getShort(ERROR_CODE_KEY_NAME);
+        error = Errors.forCode(struct.getShort(ERROR_CODE_KEY_NAME));
         Object[] mechanisms = struct.getArray(ENABLED_MECHANISMS_KEY_NAME);
         ArrayList<String> enabledMechanisms = new ArrayList<>();
         for (Object mechanism : mechanisms)
@@ -66,8 +67,8 @@ public class SaslHandshakeResponse extends AbstractResponse {
         this.enabledMechanisms = enabledMechanisms;
     }
 
-    public short errorCode() {
-        return errorCode;
+    public Errors error() {
+        return error;
     }
 
     public List<String> enabledMechanisms() {

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 d687d99..ff2638b 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
@@ -117,15 +117,15 @@ public class StopReplicaRequest extends AbstractRequest {
 
     @Override
     public AbstractResponse getErrorResponse(Throwable e) {
-        Map<TopicPartition, Short> responses = new HashMap<>(partitions.size());
+        Map<TopicPartition, Errors> responses = new HashMap<>(partitions.size());
         for (TopicPartition partition : partitions) {
-            responses.put(partition, Errors.forException(e).code());
+            responses.put(partition, Errors.forException(e));
         }
 
         short versionId = version();
         switch (versionId) {
             case 0:
-                return new StopReplicaResponse(Errors.NONE.code(), responses);
+                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)));

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 92d9e58..b39fb19 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
@@ -36,8 +36,8 @@ public class StopReplicaResponse extends AbstractResponse {
     private static final String PARTITIONS_PARTITION_KEY_NAME = "partition";
     private static final String PARTITIONS_ERROR_CODE_KEY_NAME = "error_code";
 
-    private final Map<TopicPartition, Short> responses;
-    private final short errorCode;
+    private final Map<TopicPartition, Errors> responses;
+    private final Errors error;
 
     /**
      * Possible error code:
@@ -45,30 +45,28 @@ public class StopReplicaResponse extends AbstractResponse {
      * STALE_CONTROLLER_EPOCH (11)
      */
 
-    public StopReplicaResponse(Map<TopicPartition, Short> responses) {
-        this(Errors.NONE.code(), responses);
+    public StopReplicaResponse(Map<TopicPartition, Errors> responses) {
+        this(Errors.NONE, responses);
     }
 
-    public StopReplicaResponse(short errorCode, Map<TopicPartition, Short> responses) {
+    public StopReplicaResponse(Errors error, Map<TopicPartition, Errors> responses) {
         super(new Struct(CURRENT_SCHEMA));
 
-        struct.set(ERROR_CODE_KEY_NAME, errorCode);
-
         List<Struct> responseDatas = new ArrayList<>(responses.size());
-        for (Map.Entry<TopicPartition, Short> response : responses.entrySet()) {
+        for (Map.Entry<TopicPartition, Errors> response : responses.entrySet()) {
             Struct partitionData = struct.instance(PARTITIONS_KEY_NAME);
             TopicPartition partition = response.getKey();
             partitionData.set(PARTITIONS_TOPIC_KEY_NAME, partition.topic());
             partitionData.set(PARTITIONS_PARTITION_KEY_NAME, partition.partition());
-            partitionData.set(PARTITIONS_ERROR_CODE_KEY_NAME, response.getValue());
+            partitionData.set(PARTITIONS_ERROR_CODE_KEY_NAME, response.getValue().code());
             responseDatas.add(partitionData);
         }
 
         struct.set(PARTITIONS_KEY_NAME, responseDatas.toArray());
-        struct.set(ERROR_CODE_KEY_NAME, errorCode);
+        struct.set(ERROR_CODE_KEY_NAME, error.code());
 
         this.responses = responses;
-        this.errorCode = errorCode;
+        this.error = error;
     }
 
     public StopReplicaResponse(Struct struct) {
@@ -79,19 +77,19 @@ public class StopReplicaResponse extends AbstractResponse {
             Struct responseData = (Struct) responseDataObj;
             String topic = responseData.getString(PARTITIONS_TOPIC_KEY_NAME);
             int partition = responseData.getInt(PARTITIONS_PARTITION_KEY_NAME);
-            short errorCode = responseData.getShort(PARTITIONS_ERROR_CODE_KEY_NAME);
-            responses.put(new TopicPartition(topic, partition), errorCode);
+            Errors error = Errors.forCode(responseData.getShort(PARTITIONS_ERROR_CODE_KEY_NAME));
+            responses.put(new TopicPartition(topic, partition), error);
         }
 
-        errorCode = struct.getShort(ERROR_CODE_KEY_NAME);
+        error = Errors.forCode(struct.getShort(ERROR_CODE_KEY_NAME));
     }
 
-    public Map<TopicPartition, Short> responses() {
+    public Map<TopicPartition, Errors> responses() {
         return responses;
     }
 
-    public short errorCode() {
-        return errorCode;
+    public Errors error() {
+        return error;
     }
 
     public static StopReplicaResponse parse(ByteBuffer buffer, int versionId) {

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 55b7308..937bf98 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
@@ -117,7 +117,7 @@ public class SyncGroupRequest extends AbstractRequest {
         switch (versionId) {
             case 0:
                 return new SyncGroupResponse(
-                        Errors.forException(e).code(),
+                        Errors.forException(e),
                         ByteBuffer.wrap(new byte[]{}));
             default:
                 throw new IllegalArgumentException(String.format("Version %d is not valid. Valid versions for %s are 0 to %d",

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 f459656..e598975 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
@@ -17,6 +17,7 @@
 package org.apache.kafka.common.requests;
 
 import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.protocol.ProtoUtils;
 import org.apache.kafka.common.protocol.types.Schema;
 import org.apache.kafka.common.protocol.types.Struct;
@@ -40,28 +41,28 @@ public class SyncGroupResponse extends AbstractResponse {
      * GROUP_AUTHORIZATION_FAILED (30)
      */
 
-    private final short errorCode;
+    private final Errors error;
     private final ByteBuffer memberState;
 
-    public SyncGroupResponse(short errorCode, ByteBuffer memberState) {
+    public SyncGroupResponse(Errors error, ByteBuffer memberState) {
         super(new Struct(CURRENT_SCHEMA));
 
-        struct.set(ERROR_CODE_KEY_NAME, errorCode);
+        struct.set(ERROR_CODE_KEY_NAME, error.code());
         struct.set(MEMBER_ASSIGNMENT_KEY_NAME, memberState);
 
-        this.errorCode = errorCode;
+        this.error = error;
         this.memberState = memberState;
     }
 
     public SyncGroupResponse(Struct struct) {
         super(struct);
 
-        this.errorCode = struct.getShort(ERROR_CODE_KEY_NAME);
+        this.error = Errors.forCode(struct.getShort(ERROR_CODE_KEY_NAME));
         this.memberState = struct.getBytes(MEMBER_ASSIGNMENT_KEY_NAME);
     }
 
-    public short errorCode() {
-        return errorCode;
+    public Errors error() {
+        return error;
     }
 
     public ByteBuffer memberAssignment() {

http://git-wip-us.apache.org/repos/asf/kafka/blob/9898d665/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 95e5683..ef680ff 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
@@ -280,7 +280,7 @@ public class UpdateMetadataRequest extends AbstractRequest {
     public AbstractResponse getErrorResponse(Throwable e) {
         short versionId = version();
         if (versionId <= 3)
-            return new UpdateMetadataResponse(Errors.forException(e).code());
+            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)));


Mime
View raw message