kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ij...@apache.org
Subject [8/8] kafka git commit: MINOR: Move request/response schemas to the corresponding object representation
Date Tue, 19 Sep 2017 04:13:29 GMT
MINOR: Move request/response schemas to the corresponding object representation

This refactor achieves the following:

1. Breaks up the increasingly unmanageable `Protocol` class and moves schemas closer to their
actual usage.
2. Removes the need for redundant field identifiers maintained separately in `Protocol` and
the respective request/response objects.
3. Provides a better mechanism for sharing common fields between different schemas (e.g. topics,
partitions, error codes, etc.).
4. Adds convenience helpers to `Struct` for common patterns (such as setting a field only
if it exists).

Author: Jason Gustafson <jason@confluent.io>

Reviewers: Ismael Juma <ismael@juma.me.uk>

Closes #3813 from hachikuji/protocol-schema-refactor


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

Branch: refs/heads/trunk
Commit: 0cf7708007b01faac5012d939f3c50db274f858d
Parents: a64fe2e
Author: Jason Gustafson <jason@confluent.io>
Authored: Tue Sep 19 05:12:55 2017 +0100
Committer: Ismael Juma <ismael@juma.me.uk>
Committed: Tue Sep 19 05:12:55 2017 +0100

----------------------------------------------------------------------
 checkstyle/import-control.xml                   |    1 +
 .../org/apache/kafka/clients/NetworkClient.java |    5 +-
 .../kafka/clients/consumer/StickyAssignor.java  |   22 +-
 .../apache/kafka/common/protocol/ApiKeys.java   |  249 ++-
 .../kafka/common/protocol/CommonFields.java     |   40 +
 .../kafka/common/protocol/ProtoUtils.java       |    6 +-
 .../apache/kafka/common/protocol/Protocol.java  | 2095 +-----------------
 .../kafka/common/protocol/types/BoundField.java |   37 +
 .../kafka/common/protocol/types/Field.java      |   72 +-
 .../kafka/common/protocol/types/Schema.java     |   46 +-
 .../kafka/common/protocol/types/Struct.java     |  129 +-
 .../kafka/common/requests/AbstractRequest.java  |    4 +-
 .../kafka/common/requests/AbstractResponse.java |    5 +-
 .../common/requests/AddOffsetsToTxnRequest.java |   16 +
 .../requests/AddOffsetsToTxnResponse.java       |   20 +-
 .../requests/AddPartitionsToTxnRequest.java     |   35 +-
 .../requests/AddPartitionsToTxnResponse.java    |   37 +-
 .../common/requests/AlterConfigsRequest.java    |   26 +
 .../common/requests/AlterConfigsResponse.java   |   27 +-
 .../common/requests/AlterReplicaDirRequest.java |   23 +-
 .../requests/AlterReplicaDirResponse.java       |   40 +-
 .../apache/kafka/common/requests/ApiError.java  |   20 +-
 .../common/requests/ApiVersionsRequest.java     |   10 +
 .../common/requests/ApiVersionsResponse.java    |   72 +-
 .../requests/ControlledShutdownRequest.java     |   20 +-
 .../requests/ControlledShutdownResponse.java    |   41 +-
 .../common/requests/CreateAclsRequest.java      |   31 +-
 .../common/requests/CreateAclsResponse.java     |   29 +-
 .../common/requests/CreateTopicsRequest.java    |   76 +-
 .../common/requests/CreateTopicsResponse.java   |   42 +-
 .../common/requests/DeleteAclsRequest.java      |   22 +
 .../common/requests/DeleteAclsResponse.java     |   55 +-
 .../common/requests/DeleteRecordsRequest.java   |   36 +-
 .../common/requests/DeleteRecordsResponse.java  |   47 +-
 .../common/requests/DeleteTopicsRequest.java    |   19 +
 .../common/requests/DeleteTopicsResponse.java   |   39 +-
 .../common/requests/DescribeAclsRequest.java    |   20 +
 .../common/requests/DescribeAclsResponse.java   |   52 +-
 .../common/requests/DescribeConfigsRequest.java |   18 +
 .../requests/DescribeConfigsResponse.java       |   65 +-
 .../common/requests/DescribeGroupsRequest.java  |   17 +
 .../common/requests/DescribeGroupsResponse.java |   48 +-
 .../common/requests/DescribeLogDirsRequest.java |   20 +-
 .../requests/DescribeLogDirsResponse.java       |   52 +-
 .../kafka/common/requests/EndTxnRequest.java    |   17 +
 .../kafka/common/requests/EndTxnResponse.java   |   20 +-
 .../kafka/common/requests/FetchRequest.java     |  108 +-
 .../kafka/common/requests/FetchResponse.java    |  132 +-
 .../common/requests/FindCoordinatorRequest.java |   17 +
 .../requests/FindCoordinatorResponse.java       |   57 +-
 .../kafka/common/requests/HeartbeatRequest.java |   17 +
 .../common/requests/HeartbeatResponse.java      |   23 +-
 .../common/requests/InitProducerIdRequest.java  |   13 +
 .../common/requests/InitProducerIdResponse.java |   29 +-
 .../kafka/common/requests/JoinGroupRequest.java |   39 +
 .../common/requests/JoinGroupResponse.java      |   70 +-
 .../common/requests/LeaderAndIsrRequest.java    |   64 +-
 .../common/requests/LeaderAndIsrResponse.java   |   42 +-
 .../common/requests/LeaveGroupRequest.java      |   18 +-
 .../common/requests/LeaveGroupResponse.java     |   23 +-
 .../common/requests/ListGroupsRequest.java      |   14 +-
 .../common/requests/ListGroupsResponse.java     |   47 +-
 .../common/requests/ListOffsetRequest.java      |   58 +-
 .../common/requests/ListOffsetResponse.java     |   65 +-
 .../kafka/common/requests/MetadataRequest.java  |   34 +
 .../kafka/common/requests/MetadataResponse.java |  126 +-
 .../common/requests/OffsetCommitRequest.java    |   73 +-
 .../common/requests/OffsetCommitResponse.java   |   56 +-
 .../common/requests/OffsetFetchRequest.java     |   54 +-
 .../common/requests/OffsetFetchResponse.java    |   79 +-
 .../requests/OffsetsForLeaderEpochRequest.java  |   47 +-
 .../requests/OffsetsForLeaderEpochResponse.java |   56 +-
 .../kafka/common/requests/ProduceRequest.java   |   71 +-
 .../kafka/common/requests/ProduceResponse.java  |   80 +-
 .../kafka/common/requests/RequestContext.java   |    3 +-
 .../kafka/common/requests/RequestHeader.java    |   31 +-
 .../kafka/common/requests/RequestUtils.java     |   59 +-
 .../kafka/common/requests/ResponseHeader.java   |   19 +-
 .../requests/SaslAuthenticateRequest.java       |   15 +-
 .../requests/SaslAuthenticateResponse.java      |   30 +-
 .../common/requests/SaslHandshakeRequest.java   |   22 +-
 .../common/requests/SaslHandshakeResponse.java  |   28 +-
 .../common/requests/StopReplicaRequest.java     |   31 +-
 .../common/requests/StopReplicaResponse.java    |   40 +-
 .../kafka/common/requests/SyncGroupRequest.java |   33 +-
 .../common/requests/SyncGroupResponse.java      |   29 +-
 .../common/requests/TxnOffsetCommitRequest.java |   47 +-
 .../requests/TxnOffsetCommitResponse.java       |   50 +-
 .../common/requests/UpdateMetadataRequest.java  |  174 +-
 .../common/requests/UpdateMetadataResponse.java |   21 +-
 .../common/requests/WriteTxnMarkersRequest.java |   40 +-
 .../requests/WriteTxnMarkersResponse.java       |   49 +-
 .../authenticator/SaslServerAuthenticator.java  |    5 +-
 .../apache/kafka/clients/NetworkClientTest.java |    5 +-
 .../kafka/clients/NodeApiVersionsTest.java      |    2 +-
 .../kafka/common/protocol/ApiKeysTest.java      |   13 +-
 .../kafka/common/protocol/ProtoUtilsTest.java   |    9 +-
 .../types/ProtocolSerializationTest.java        |   20 +-
 .../requests/ApiVersionsResponseTest.java       |    2 +-
 .../common/requests/RequestHeaderTest.java      |    6 +-
 .../common/requests/RequestResponseTest.java    |    2 +-
 .../authenticator/SaslAuthenticatorTest.java    |    7 +-
 .../controller/ControllerChannelManager.scala   |    4 +-
 .../group/GroupMetadataManager.scala            |    2 +-
 .../transaction/TransactionLog.scala            |    2 +-
 .../scala/kafka/log/ProducerStateManager.scala  |    2 +-
 .../scala/kafka/network/RequestChannel.scala    |    4 +-
 .../src/main/scala/kafka/server/KafkaApis.scala |    6 +-
 .../kafka/api/AuthorizerIntegrationTest.scala   |   18 +-
 .../kafka/server/ApiVersionsRequestTest.scala   |    2 +-
 .../unit/kafka/server/ApiVersionsTest.scala     |   14 +-
 .../unit/kafka/server/MetadataCacheTest.scala   |   12 +-
 .../unit/kafka/server/RequestQuotaTest.scala    |    6 +-
 113 files changed, 3243 insertions(+), 2956 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/0cf77080/checkstyle/import-control.xml
----------------------------------------------------------------------
diff --git a/checkstyle/import-control.xml b/checkstyle/import-control.xml
index bea7e20..3329b2d 100644
--- a/checkstyle/import-control.xml
+++ b/checkstyle/import-control.xml
@@ -105,6 +105,7 @@
       <allow pkg="org.apache.kafka.common.errors" />
       <allow pkg="org.apache.kafka.common.protocol.types" />
       <allow pkg="org.apache.kafka.common.record" />
+      <allow pkg="org.apache.kafka.common.requests" />
     </subpackage>
 
     <subpackage name="record">

http://git-wip-us.apache.org/repos/asf/kafka/blob/0cf77080/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 f046696..c3c15df 100644
--- a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
@@ -26,6 +26,7 @@ import org.apache.kafka.common.network.Selectable;
 import org.apache.kafka.common.network.Send;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.protocol.CommonFields;
 import org.apache.kafka.common.protocol.types.Struct;
 import org.apache.kafka.common.requests.AbstractRequest;
 import org.apache.kafka.common.requests.AbstractResponse;
@@ -570,8 +571,8 @@ public class NetworkClient implements KafkaClient {
         // Always expect the response version id to be the same as the request version id
         Struct responseBody = requestHeader.apiKey().parseResponse(requestHeader.apiVersion(),
responseBuffer);
         correlate(requestHeader, responseHeader);
-        if (throttleTimeSensor != null && responseBody.hasField(AbstractResponse.THROTTLE_TIME_KEY_NAME))
-            throttleTimeSensor.record(responseBody.getInt(AbstractResponse.THROTTLE_TIME_KEY_NAME),
now);
+        if (throttleTimeSensor != null && responseBody.hasField(CommonFields.THROTTLE_TIME_MS))
+            throttleTimeSensor.record(responseBody.get(CommonFields.THROTTLE_TIME_MS), now);
         return responseBody;
     }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/0cf77080/clients/src/main/java/org/apache/kafka/clients/consumer/StickyAssignor.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/StickyAssignor.java b/clients/src/main/java/org/apache/kafka/clients/consumer/StickyAssignor.java
index d7bfaf1..247b619 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/StickyAssignor.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/StickyAssignor.java
@@ -16,6 +16,17 @@
  */
 package org.apache.kafka.clients.consumer;
 
+import org.apache.kafka.clients.consumer.internals.AbstractPartitionAssignor;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.protocol.types.ArrayOf;
+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.protocol.types.Type;
+import org.apache.kafka.common.utils.CollectionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.io.Serializable;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
@@ -31,17 +42,6 @@ import java.util.Map.Entry;
 import java.util.Set;
 import java.util.TreeSet;
 
-import org.apache.kafka.clients.consumer.internals.AbstractPartitionAssignor;
-import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.protocol.types.ArrayOf;
-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.protocol.types.Type;
-import org.apache.kafka.common.utils.CollectionUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 /**
  * <p>The sticky assignor serves two purposes. First, it guarantees an assignment that
is as balanced as possible, meaning either:
  * <ul>

http://git-wip-us.apache.org/repos/asf/kafka/blob/0cf77080/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java b/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java
index d37eddf..0e087eb 100644
--- a/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java
+++ b/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java
@@ -19,33 +19,117 @@ package org.apache.kafka.common.protocol;
 import org.apache.kafka.common.protocol.types.Schema;
 import org.apache.kafka.common.protocol.types.SchemaException;
 import org.apache.kafka.common.protocol.types.Struct;
+import org.apache.kafka.common.protocol.types.Type;
 import org.apache.kafka.common.record.RecordBatch;
+import org.apache.kafka.common.requests.AddOffsetsToTxnRequest;
+import org.apache.kafka.common.requests.AddOffsetsToTxnResponse;
+import org.apache.kafka.common.requests.AddPartitionsToTxnRequest;
+import org.apache.kafka.common.requests.AddPartitionsToTxnResponse;
+import org.apache.kafka.common.requests.AlterConfigsRequest;
+import org.apache.kafka.common.requests.AlterConfigsResponse;
+import org.apache.kafka.common.requests.AlterReplicaDirRequest;
+import org.apache.kafka.common.requests.AlterReplicaDirResponse;
+import org.apache.kafka.common.requests.ApiVersionsRequest;
+import org.apache.kafka.common.requests.ApiVersionsResponse;
+import org.apache.kafka.common.requests.ControlledShutdownRequest;
+import org.apache.kafka.common.requests.ControlledShutdownResponse;
+import org.apache.kafka.common.requests.CreateAclsRequest;
+import org.apache.kafka.common.requests.CreateAclsResponse;
+import org.apache.kafka.common.requests.CreateTopicsRequest;
+import org.apache.kafka.common.requests.CreateTopicsResponse;
+import org.apache.kafka.common.requests.DeleteAclsRequest;
+import org.apache.kafka.common.requests.DeleteAclsResponse;
+import org.apache.kafka.common.requests.DeleteRecordsRequest;
+import org.apache.kafka.common.requests.DeleteRecordsResponse;
+import org.apache.kafka.common.requests.DeleteTopicsRequest;
+import org.apache.kafka.common.requests.DeleteTopicsResponse;
+import org.apache.kafka.common.requests.DescribeAclsRequest;
+import org.apache.kafka.common.requests.DescribeAclsResponse;
+import org.apache.kafka.common.requests.DescribeConfigsRequest;
+import org.apache.kafka.common.requests.DescribeConfigsResponse;
+import org.apache.kafka.common.requests.DescribeGroupsRequest;
+import org.apache.kafka.common.requests.DescribeGroupsResponse;
+import org.apache.kafka.common.requests.DescribeLogDirsRequest;
+import org.apache.kafka.common.requests.DescribeLogDirsResponse;
+import org.apache.kafka.common.requests.EndTxnRequest;
+import org.apache.kafka.common.requests.EndTxnResponse;
+import org.apache.kafka.common.requests.FetchRequest;
+import org.apache.kafka.common.requests.FetchResponse;
+import org.apache.kafka.common.requests.FindCoordinatorRequest;
+import org.apache.kafka.common.requests.FindCoordinatorResponse;
+import org.apache.kafka.common.requests.HeartbeatRequest;
+import org.apache.kafka.common.requests.HeartbeatResponse;
+import org.apache.kafka.common.requests.InitProducerIdRequest;
+import org.apache.kafka.common.requests.InitProducerIdResponse;
+import org.apache.kafka.common.requests.JoinGroupRequest;
+import org.apache.kafka.common.requests.JoinGroupResponse;
+import org.apache.kafka.common.requests.LeaderAndIsrRequest;
+import org.apache.kafka.common.requests.LeaderAndIsrResponse;
+import org.apache.kafka.common.requests.LeaveGroupRequest;
+import org.apache.kafka.common.requests.LeaveGroupResponse;
+import org.apache.kafka.common.requests.ListGroupsRequest;
+import org.apache.kafka.common.requests.ListGroupsResponse;
+import org.apache.kafka.common.requests.ListOffsetRequest;
+import org.apache.kafka.common.requests.ListOffsetResponse;
+import org.apache.kafka.common.requests.MetadataRequest;
+import org.apache.kafka.common.requests.MetadataResponse;
+import org.apache.kafka.common.requests.OffsetCommitRequest;
+import org.apache.kafka.common.requests.OffsetCommitResponse;
+import org.apache.kafka.common.requests.OffsetFetchRequest;
+import org.apache.kafka.common.requests.OffsetFetchResponse;
+import org.apache.kafka.common.requests.OffsetsForLeaderEpochRequest;
+import org.apache.kafka.common.requests.OffsetsForLeaderEpochResponse;
+import org.apache.kafka.common.requests.ProduceRequest;
+import org.apache.kafka.common.requests.ProduceResponse;
+import org.apache.kafka.common.requests.SaslAuthenticateRequest;
+import org.apache.kafka.common.requests.SaslAuthenticateResponse;
+import org.apache.kafka.common.requests.SaslHandshakeRequest;
+import org.apache.kafka.common.requests.SaslHandshakeResponse;
+import org.apache.kafka.common.requests.StopReplicaRequest;
+import org.apache.kafka.common.requests.StopReplicaResponse;
+import org.apache.kafka.common.requests.SyncGroupRequest;
+import org.apache.kafka.common.requests.SyncGroupResponse;
+import org.apache.kafka.common.requests.TxnOffsetCommitRequest;
+import org.apache.kafka.common.requests.TxnOffsetCommitResponse;
+import org.apache.kafka.common.requests.UpdateMetadataRequest;
+import org.apache.kafka.common.requests.UpdateMetadataResponse;
+import org.apache.kafka.common.requests.WriteTxnMarkersRequest;
+import org.apache.kafka.common.requests.WriteTxnMarkersResponse;
 
 import java.nio.ByteBuffer;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.apache.kafka.common.protocol.types.Type.BYTES;
+import static org.apache.kafka.common.protocol.types.Type.NULLABLE_BYTES;
+import static org.apache.kafka.common.protocol.types.Type.RECORDS;
 
 /**
  * Identifiers for all the Kafka APIs
  */
 public enum ApiKeys {
-    PRODUCE(0, "Produce"),
-    FETCH(1, "Fetch"),
-    LIST_OFFSETS(2, "Offsets"),
-    METADATA(3, "Metadata"),
-    LEADER_AND_ISR(4, "LeaderAndIsr", true),
-    STOP_REPLICA(5, "StopReplica", true),
-    UPDATE_METADATA_KEY(6, "UpdateMetadata", true),
-    CONTROLLED_SHUTDOWN_KEY(7, "ControlledShutdown", true),
-    OFFSET_COMMIT(8, "OffsetCommit"),
-    OFFSET_FETCH(9, "OffsetFetch"),
-    FIND_COORDINATOR(10, "FindCoordinator"),
-    JOIN_GROUP(11, "JoinGroup"),
-    HEARTBEAT(12, "Heartbeat"),
-    LEAVE_GROUP(13, "LeaveGroup"),
-    SYNC_GROUP(14, "SyncGroup"),
-    DESCRIBE_GROUPS(15, "DescribeGroups"),
-    LIST_GROUPS(16, "ListGroups"),
-    SASL_HANDSHAKE(17, "SaslHandshake"),
-    API_VERSIONS(18, "ApiVersions") {
+    PRODUCE(0, "Produce", ProduceRequest.schemaVersions(), ProduceResponse.schemaVersions()),
+    FETCH(1, "Fetch", FetchRequest.schemaVersions(), FetchResponse.schemaVersions()),
+    LIST_OFFSETS(2, "ListOffsets", ListOffsetRequest.schemaVersions(), ListOffsetResponse.schemaVersions()),
+    METADATA(3, "Metadata", MetadataRequest.schemaVersions(), MetadataResponse.schemaVersions()),
+    LEADER_AND_ISR(4, "LeaderAndIsr", true, LeaderAndIsrRequest.schemaVersions(), LeaderAndIsrResponse.schemaVersions()),
+    STOP_REPLICA(5, "StopReplica", true, StopReplicaRequest.schemaVersions(), StopReplicaResponse.schemaVersions()),
+    UPDATE_METADATA(6, "UpdateMetadata", true, UpdateMetadataRequest.schemaVersions(),
+            UpdateMetadataResponse.schemaVersions()),
+    CONTROLLED_SHUTDOWN(7, "ControlledShutdown", true, ControlledShutdownRequest.schemaVersions(),
+            ControlledShutdownResponse.schemaVersions()),
+    OFFSET_COMMIT(8, "OffsetCommit", OffsetCommitRequest.schemaVersions(), OffsetCommitResponse.schemaVersions()),
+    OFFSET_FETCH(9, "OffsetFetch", OffsetFetchRequest.schemaVersions(), OffsetFetchResponse.schemaVersions()),
+    FIND_COORDINATOR(10, "FindCoordinator", FindCoordinatorRequest.schemaVersions(),
+            FindCoordinatorResponse.schemaVersions()),
+    JOIN_GROUP(11, "JoinGroup", JoinGroupRequest.schemaVersions(), JoinGroupResponse.schemaVersions()),
+    HEARTBEAT(12, "Heartbeat", HeartbeatRequest.schemaVersions(), HeartbeatResponse.schemaVersions()),
+    LEAVE_GROUP(13, "LeaveGroup", LeaveGroupRequest.schemaVersions(), LeaveGroupResponse.schemaVersions()),
+    SYNC_GROUP(14, "SyncGroup", SyncGroupRequest.schemaVersions(), SyncGroupResponse.schemaVersions()),
+    DESCRIBE_GROUPS(15, "DescribeGroups", DescribeGroupsRequest.schemaVersions(),
+            DescribeGroupsResponse.schemaVersions()),
+    LIST_GROUPS(16, "ListGroups", ListGroupsRequest.schemaVersions(), ListGroupsResponse.schemaVersions()),
+    SASL_HANDSHAKE(17, "SaslHandshake", SaslHandshakeRequest.schemaVersions(), SaslHandshakeResponse.schemaVersions()),
+    API_VERSIONS(18, "ApiVersions", ApiVersionsRequest.schemaVersions(), ApiVersionsResponse.schemaVersions())
{
         @Override
         public Struct parseResponse(short version, ByteBuffer buffer) {
             // Fallback to version 0 for ApiVersions response. If a client sends an ApiVersionsRequest
@@ -54,24 +138,37 @@ public enum ApiKeys {
             return parseResponse(version, buffer, (short) 0);
         }
     },
-    CREATE_TOPICS(19, "CreateTopics"),
-    DELETE_TOPICS(20, "DeleteTopics"),
-    DELETE_RECORDS(21, "DeleteRecords"),
-    INIT_PRODUCER_ID(22, "InitProducerId"),
-    OFFSET_FOR_LEADER_EPOCH(23, "OffsetForLeaderEpoch", true),
-    ADD_PARTITIONS_TO_TXN(24, "AddPartitionsToTxn", false, RecordBatch.MAGIC_VALUE_V2),
-    ADD_OFFSETS_TO_TXN(25, "AddOffsetsToTxn", false, RecordBatch.MAGIC_VALUE_V2),
-    END_TXN(26, "EndTxn", false, RecordBatch.MAGIC_VALUE_V2),
-    WRITE_TXN_MARKERS(27, "WriteTxnMarkers", true, RecordBatch.MAGIC_VALUE_V2),
-    TXN_OFFSET_COMMIT(28, "TxnOffsetCommit", false, RecordBatch.MAGIC_VALUE_V2),
-    DESCRIBE_ACLS(29, "DescribeAcls"),
-    CREATE_ACLS(30, "CreateAcls"),
-    DELETE_ACLS(31, "DeleteAcls"),
-    DESCRIBE_CONFIGS(32, "DescribeConfigs"),
-    ALTER_CONFIGS(33, "AlterConfigs"),
-    ALTER_REPLICA_DIR(34, "AlterReplicaDir"),
-    DESCRIBE_LOG_DIRS(35, "DescribeLogDirs"),
-    SASL_AUTHENTICATE(36, "SaslAuthenticate");
+    CREATE_TOPICS(19, "CreateTopics", CreateTopicsRequest.schemaVersions(), CreateTopicsResponse.schemaVersions()),
+    DELETE_TOPICS(20, "DeleteTopics", DeleteTopicsRequest.schemaVersions(), DeleteTopicsResponse.schemaVersions()),
+    DELETE_RECORDS(21, "DeleteRecords", DeleteRecordsRequest.schemaVersions(), DeleteRecordsResponse.schemaVersions()),
+    INIT_PRODUCER_ID(22, "InitProducerId", InitProducerIdRequest.schemaVersions(),
+            InitProducerIdResponse.schemaVersions()),
+    OFFSET_FOR_LEADER_EPOCH(23, "OffsetForLeaderEpoch", true, OffsetsForLeaderEpochRequest.schemaVersions(),
+            OffsetsForLeaderEpochResponse.schemaVersions()),
+    ADD_PARTITIONS_TO_TXN(24, "AddPartitionsToTxn", false, RecordBatch.MAGIC_VALUE_V2,
+            AddPartitionsToTxnRequest.schemaVersions(), AddPartitionsToTxnResponse.schemaVersions()),
+    ADD_OFFSETS_TO_TXN(25, "AddOffsetsToTxn", false, RecordBatch.MAGIC_VALUE_V2, AddOffsetsToTxnRequest.schemaVersions(),
+            AddOffsetsToTxnResponse.schemaVersions()),
+    END_TXN(26, "EndTxn", false, RecordBatch.MAGIC_VALUE_V2, EndTxnRequest.schemaVersions(),
+            EndTxnResponse.schemaVersions()),
+    WRITE_TXN_MARKERS(27, "WriteTxnMarkers", true, RecordBatch.MAGIC_VALUE_V2, WriteTxnMarkersRequest.schemaVersions(),
+            WriteTxnMarkersResponse.schemaVersions()),
+    TXN_OFFSET_COMMIT(28, "TxnOffsetCommit", false, RecordBatch.MAGIC_VALUE_V2, TxnOffsetCommitRequest.schemaVersions(),
+            TxnOffsetCommitResponse.schemaVersions()),
+    DESCRIBE_ACLS(29, "DescribeAcls", DescribeAclsRequest.schemaVersions(), DescribeAclsResponse.schemaVersions()),
+    CREATE_ACLS(30, "CreateAcls", CreateAclsRequest.schemaVersions(), CreateAclsResponse.schemaVersions()),
+    DELETE_ACLS(31, "DeleteAcls", DeleteAclsRequest.schemaVersions(), DeleteAclsResponse.schemaVersions()),
+    DESCRIBE_CONFIGS(32, "DescribeConfigs", DescribeConfigsRequest.schemaVersions(),
+            DescribeConfigsResponse.schemaVersions()),
+    ALTER_CONFIGS(33, "AlterConfigs", AlterConfigsRequest.schemaVersions(),
+            AlterConfigsResponse.schemaVersions()),
+    ALTER_REPLICA_DIR(34, "AlterReplicaDir", AlterReplicaDirRequest.schemaVersions(),
+            AlterReplicaDirResponse.schemaVersions()),
+    DESCRIBE_LOG_DIRS(35, "DescribeLogDirs", DescribeLogDirsRequest.schemaVersions(),
+            DescribeLogDirsResponse.schemaVersions()),
+    SASL_AUTHENTICATE(36, "SaslAuthenticate", SaslAuthenticateRequest.schemaVersions(),
+            SaslAuthenticateResponse.schemaVersions());
+
 
     private static final ApiKeys[] ID_TO_TYPE;
     private static final int MIN_API_KEY = 0;
@@ -100,21 +197,48 @@ public enum ApiKeys {
     /** indicates the minimum required inter broker magic required to support the API */
     public final byte minRequiredInterBrokerMagic;
 
-    ApiKeys(int id, String name) {
-        this(id, name, false);
+    public final Schema[] requestSchemas;
+    public final Schema[] responseSchemas;
+    public final boolean requiresDelayedAllocation;
+
+    ApiKeys(int id, String name, Schema[] requestSchemas, Schema[] responseSchemas) {
+        this(id, name, false, requestSchemas, responseSchemas);
     }
 
-    ApiKeys(int id, String name, boolean clusterAction) {
-        this(id, name, clusterAction, RecordBatch.MAGIC_VALUE_V0);
+    ApiKeys(int id, String name, boolean clusterAction, Schema[] requestSchemas, Schema[]
responseSchemas) {
+        this(id, name, clusterAction, RecordBatch.MAGIC_VALUE_V0, requestSchemas, responseSchemas);
     }
 
-    ApiKeys(int id, String name, boolean clusterAction, byte minRequiredInterBrokerMagic)
{
+    ApiKeys(int id, String name, boolean clusterAction, byte minRequiredInterBrokerMagic,
+            Schema[] requestSchemas, Schema[] responseSchemas) {
         if (id < 0)
             throw new IllegalArgumentException("id must not be negative, id: " + id);
         this.id = (short) id;
         this.name = name;
         this.clusterAction = clusterAction;
         this.minRequiredInterBrokerMagic = minRequiredInterBrokerMagic;
+
+        if (requestSchemas.length != responseSchemas.length)
+            throw new IllegalStateException(requestSchemas.length + " request versions for
api " + name
+                    + " but " + responseSchemas.length + " response versions.");
+
+        for (int i = 0; i < requestSchemas.length; ++i) {
+            if (requestSchemas[i] == null)
+                throw new IllegalStateException("Request schema for api " + name + " for
version " + i + " is null");
+            if (responseSchemas[i] == null)
+                throw new IllegalStateException("Response schema for api " + name + " for
version " + i + " is null");
+        }
+
+        boolean requestRetainsBufferReference = false;
+        for (Schema requestVersionSchema : requestSchemas) {
+            if (retainsBufferReference(requestVersionSchema)) {
+                requestRetainsBufferReference = true;
+                break;
+            }
+        }
+        this.requiresDelayedAllocation = requestRetainsBufferReference;
+        this.requestSchemas = requestSchemas;
+        this.responseSchemas = responseSchemas;
     }
 
     public static ApiKeys forId(int id) {
@@ -129,23 +253,19 @@ public enum ApiKeys {
     }
 
     public short latestVersion() {
-        if (id >= Protocol.CURR_VERSION.length)
-            throw new IllegalArgumentException("Latest version for API key " + this + " is
not defined");
-        return Protocol.CURR_VERSION[id];
+        return (short) (requestSchemas.length - 1);
     }
 
     public short oldestVersion() {
-        if (id >= Protocol.MIN_VERSIONS.length)
-            throw new IllegalArgumentException("Oldest version for API key " + this + " is
not defined");
-        return Protocol.MIN_VERSIONS[id];
+        return 0;
     }
 
     public Schema requestSchema(short version) {
-        return schemaFor(Protocol.REQUESTS, version);
+        return schemaFor(requestSchemas, version);
     }
 
     public Schema responseSchema(short version) {
-        return schemaFor(Protocol.RESPONSES, version);
+        return schemaFor(responseSchemas, version);
     }
 
     public Struct parseRequest(short version, ByteBuffer buffer) {
@@ -169,18 +289,16 @@ public enum ApiKeys {
         }
     }
 
-    private Schema schemaFor(Schema[][] schemas, short version) {
-        if (id > schemas.length)
-            throw new IllegalArgumentException("No schema available for API key " + this);
-        if (version < 0 || version > latestVersion())
+    private Schema schemaFor(Schema[] versions, short version) {
+        if (!isVersionSupported(version))
             throw new IllegalArgumentException("Invalid version for API key " + this + ":
" + version);
-
-        Schema[] versions = schemas[id];
-        if (versions[version] == null)
-            throw new IllegalArgumentException("Unsupported version for API key " + this
+ ": " + version);
         return versions[version];
     }
 
+    public boolean isVersionSupported(short apiVersion) {
+        return apiVersion >= oldestVersion() && apiVersion <= latestVersion();
+    }
+
     private static String toHtml() {
         final StringBuilder b = new StringBuilder();
         b.append("<table class=\"data-table\"><tbody>\n");
@@ -206,4 +324,19 @@ public enum ApiKeys {
         System.out.println(toHtml());
     }
 
+    private static boolean retainsBufferReference(Schema schema) {
+        final AtomicReference<Boolean> foundBufferReference = new AtomicReference<>(Boolean.FALSE);
+        SchemaVisitor detector = new SchemaVisitorAdapter() {
+            @Override
+            public void visit(Type field) {
+                if (field == BYTES || field == NULLABLE_BYTES || field == RECORDS) {
+                    foundBufferReference.set(Boolean.TRUE);
+                }
+            }
+        };
+        foundBufferReference.set(Boolean.FALSE);
+        ProtoUtils.walk(schema, detector);
+        return foundBufferReference.get();
+    }
+
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/0cf77080/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
new file mode 100644
index 0000000..e12cde4
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/protocol/CommonFields.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.protocol;
+
+import org.apache.kafka.common.protocol.types.Field;
+
+public class CommonFields {
+    public static final Field.Int32 THROTTLE_TIME_MS = new Field.Int32("throttle_time_ms",
+            "Duration in milliseconds for which the request was throttled due to quota violation
(Zero if the " +
+                    "request did not violate any quota)", 0);
+    public static final Field.Str TOPIC_NAME = new Field.Str("topic", "Name of topic");
+    public static final Field.Int32 PARTITION_ID = new Field.Int32("partition", "Topic partition
id");
+    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
+    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");
+    public static final Field.Str PRINCIPAL = new Field.Str("principal", "The ACL principal");
+    public static final Field.NullableStr PRINCIPAL_FILTER = new Field.NullableStr("principal",
"The ACL principal filter");
+    public static final Field.Str HOST = new Field.Str("host", "The ACL host");
+    public static final Field.NullableStr HOST_FILTER = new Field.NullableStr("host", "The
ACL host filter");
+    public static final Field.Int8 OPERATION = new Field.Int8("operation", "The ACL operation");
+    public static final Field.Int8 PERMISSION_TYPE = new Field.Int8("permission_type", "The
ACL permission type");
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/0cf77080/clients/src/main/java/org/apache/kafka/common/protocol/ProtoUtils.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/ProtoUtils.java b/clients/src/main/java/org/apache/kafka/common/protocol/ProtoUtils.java
index 5d39dff..f9be12c 100644
--- a/clients/src/main/java/org/apache/kafka/common/protocol/ProtoUtils.java
+++ b/clients/src/main/java/org/apache/kafka/common/protocol/ProtoUtils.java
@@ -17,7 +17,7 @@
 package org.apache.kafka.common.protocol;
 
 import org.apache.kafka.common.protocol.types.ArrayOf;
-import org.apache.kafka.common.protocol.types.Field;
+import org.apache.kafka.common.protocol.types.BoundField;
 import org.apache.kafka.common.protocol.types.Schema;
 import org.apache.kafka.common.protocol.types.Type;
 
@@ -33,8 +33,8 @@ public class ProtoUtils {
         if (node instanceof Schema) {
             Schema schema = (Schema) node;
             visitor.visit(schema);
-            for (Field f : schema.fields()) {
-                handleNode(f.type, visitor);
+            for (BoundField f : schema.fields()) {
+                handleNode(f.def.type, visitor);
             }
         } else if (node instanceof ArrayOf) {
             ArrayOf array = (ArrayOf) node;


Mime
View raw message