From 0cce68a50897c3f797bb6ac3cecaa3d56f574390 Mon Sep 17 00:00:00 2001 From: Tom Bentley Date: Wed, 27 Sep 2017 11:16:25 +0100 Subject: [PATCH] MINOR: Add documentation about each of the messages in the protocol. --- .../apache/kafka/common/protocol/ApiKeys.java | 238 +++++++++++++----- .../kafka/common/protocol/Protocol.java | 3 + 2 files changed, 183 insertions(+), 58 deletions(-) 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 cf1bff557331b..d0b65801173f6 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 @@ -109,29 +109,95 @@ * Identifiers for all the Kafka APIs */ public enum ApiKeys { - 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()) { + PRODUCE(0, "Produce", + "Sent by a producer client to a leader broker to " + + "request that the data be appended to the given partition's log.", + ProduceRequest.schemaVersions(), ProduceResponse.schemaVersions()), + FETCH(1, "Fetch", + "Sent by a consumer (which could be a following broker) to a leader broker to fetch records for the " + + "partitions present in the request.", + FetchRequest.schemaVersions(), FetchResponse.schemaVersions()), + LIST_OFFSETS(2, "ListOffsets", + "Sent by a consumer (which could be a following broker) to a leader broker to obtain the consumer offsets " + + "for the requested partitions.", + ListOffsetRequest.schemaVersions(), ListOffsetResponse.schemaVersions()), + METADATA(3, "Metadata", + "Sent by a client to a broker to obtain metadata about the cluster and the topics in it. " + + "A metadata request is required to bootstrap a client, so it knows about the cluster controller " + + "and partitions leaders.", + MetadataRequest.schemaVersions(), MetadataResponse.schemaVersions()), + LEADER_AND_ISR(4, "LeaderAndIsr", true, + "Sent by the controller to other brokers to change the whether they are leaders or followers for " + + "the partitions given in the request.", + LeaderAndIsrRequest.schemaVersions(), LeaderAndIsrResponse.schemaVersions()), + STOP_REPLICA(5, "StopReplica", true, + "Sent by the controller to brokers hosting replicas of partitions when the partitions are going " + + "offline or should be deleted.", + StopReplicaRequest.schemaVersions(), StopReplicaResponse.schemaVersions()), + UPDATE_METADATA(6, "UpdateMetadata", true, + "Sent by the controller to other brokers in the cluster to get them to update their metadata about " + + "the cluster.", + UpdateMetadataRequest.schemaVersions(), UpdateMetadataResponse.schemaVersions()), + CONTROLLED_SHUTDOWN(7, "ControlledShutdown", true, + "Sent by a broker that is about to shutdown to the controller, so that the controller " + + "can arrange for the partitions currently lead by that broker to be led by other brokers.", + ControlledShutdownRequest.schemaVersions(), ControlledShutdownResponse.schemaVersions()), + OFFSET_COMMIT(8, "OffsetCommit", + "Sent by a consumer to the consumer group coordinator to save the consumer's offset.", + OffsetCommitRequest.schemaVersions(), OffsetCommitResponse.schemaVersions()), + OFFSET_FETCH(9, "OffsetFetch", + "Sent by a consumer to the consumer group coordinator to recover the consumer's offset.", + OffsetFetchRequest.schemaVersions(), OffsetFetchResponse.schemaVersions()), + FIND_COORDINATOR(10, "FindCoordinator", + "Sent by client to any broker to find the broker acting as the coordinator for the given group or transaction. " + + "Renamed and generalised by KIP-98.", + FindCoordinatorRequest.schemaVersions(), FindCoordinatorResponse.schemaVersions()), + JOIN_GROUP(11, "JoinGroup", + "Sent by a consumer to the broker acting as consumer group coordinator " + + "to join the given consumer group. " + + "The coordinator will elect one client the leader. " + + "Only if the client is elected leader will the response include group membership. " + + "Specified in the Kafka 0.9 Consumer Rewrite Design", + JoinGroupRequest.schemaVersions(), JoinGroupResponse.schemaVersions()), + HEARTBEAT(12, "Heartbeat", + "Sent by a consumer to the broker acting as consumer group coordinator " + + "to verify that the assigned partitions from a " + + "previous SYNC_GROUP request are still valid. " + + "Specified in the Kafka 0.9 Consumer Rewrite Design", + HeartbeatRequest.schemaVersions(), HeartbeatResponse.schemaVersions()), + LEAVE_GROUP(13, "LeaveGroup", + "Sent by a consumer to the consumer group coordinator when the consumer is leaving the group, " + + "triggering a rebalance via the HEARTBEAT/JOIN_GROUP/SYNC_GROUP protocol.", + LeaveGroupRequest.schemaVersions(), LeaveGroupResponse.schemaVersions()), + SYNC_GROUP(14, "SyncGroup", + "Sent by a consumer the the broker acting as consumer group coordinator " + + "following a JOIN_GROUP exchange. The request from the " + + "consumer that was elected as group leader provides the group state " + + "(in particular the assignment of consumers to partitions) and " + + "this is forwarded as the response to the remaining group members." + + "Specified in the Kafka 0.9 Consumer Rewrite Design", + SyncGroupRequest.schemaVersions(), SyncGroupResponse.schemaVersions()), + DESCRIBE_GROUPS(15, "DescribeGroups", + "Sent by an (admin) client to a broker to get detailed information (such as group membersip) about a specific consumer group. " + + "Specified by KIP-40.", + DescribeGroupsRequest.schemaVersions(), DescribeGroupsResponse.schemaVersions()), + LIST_GROUPS(16, "ListGroups", + "Sent by an (admin) client to a particular broker to get information about the consumers groups managed by that broker. " + + "To get a list of all consumers groups in the cluster, it needs to be sent to all brokers." + + "Specified by KIP-40.", + ListGroupsRequest.schemaVersions(), ListGroupsResponse.schemaVersions()), + SASL_HANDSHAKE(17, "SaslHandshake", + "Sent by a client to a broker to supply the SASL mechanism. Following the receipt of a successful " + + "handshake response the client performs SASL authentication using the accepted mechanism outside " + + "the Kafka protcol. " + + "Specified in KIP-43.", + SaslHandshakeRequest.schemaVersions(), SaslHandshakeResponse.schemaVersions()), + API_VERSIONS(18, "ApiVersions", + "Sent by a client to a broker to discover the protocol versions the broker supports. " + + "The response enumerates each supported ApiKey together with the minimum and maximum supported " + + "versions of that key. " + + "Specified in KIP-35.", + 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 @@ -140,38 +206,92 @@ public Struct parseResponse(short version, ByteBuffer buffer) { return parseResponse(version, buffer, (short) 0); } }, - 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()), + CREATE_TOPICS(19, "CreateTopics", + "Sent by an (admin) client to the controller to create the topics given in the request.", + CreateTopicsRequest.schemaVersions(), CreateTopicsResponse.schemaVersions()), + DELETE_TOPICS(20, "DeleteTopics", + "Sent by an (admin) client to the controller to delete the topics given in the request.", + DeleteTopicsRequest.schemaVersions(), DeleteTopicsResponse.schemaVersions()), + DELETE_RECORDS(21, "DeleteRecords", + "Sent by an client to the partition leader to request the deletion of records from topic partitions, " + + "as identified in the request.", + DeleteRecordsRequest.schemaVersions(), DeleteRecordsResponse.schemaVersions()), + INIT_PRODUCER_ID(22, "InitProducerId", + "Sent by a producer to its transaction coordinator to to get the assigned PID, increment its epoch, and " + + "fence any previous producers sharing the same TransactionalId. " + + "Specified by KIP-98.", + InitProducerIdRequest.schemaVersions(), InitProducerIdResponse.schemaVersions()), + OFFSET_FOR_LEADER_EPOCH(23, "OffsetForLeaderEpoch", true, + "Sent by a follower to the leader to retrieve the start offset corresponding to the given leader epochs. " + + "The follower uses this offset to truncate its log. " + + "Specified in KIP-101.", + OffsetsForLeaderEpochRequest.schemaVersions(), OffsetsForLeaderEpochResponse.schemaVersions()), ADD_PARTITIONS_TO_TXN(24, "AddPartitionsToTxn", false, RecordBatch.MAGIC_VALUE_V2, + "Sent by a producer to its transaction coordinator to add a partition to the current ongoing transaction. " + + "Specified by KIP-98.", 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_LOG_DIRS(34, "AlterReplicaLogDirs", AlterReplicaLogDirsRequest.schemaVersions(), - AlterReplicaLogDirsResponse.schemaVersions()), - DESCRIBE_LOG_DIRS(35, "DescribeLogDirs", DescribeLogDirsRequest.schemaVersions(), - DescribeLogDirsResponse.schemaVersions()), - SASL_AUTHENTICATE(36, "SaslAuthenticate", SaslAuthenticateRequest.schemaVersions(), - SaslAuthenticateResponse.schemaVersions()), - CREATE_PARTITIONS(37, "CreatePartitions", CreatePartitionsRequest.schemaVersions(), - CreatePartitionsResponse.schemaVersions()); + ADD_OFFSETS_TO_TXN(25, "AddOffsetsToTxn", false, RecordBatch.MAGIC_VALUE_V2, + "Sent by the producer to its transaction coordinator to indicate a consumer offset commit operation is " + + "called as part of the current ongoing transaction. " + + "Specified by KIP-98.", + AddOffsetsToTxnRequest.schemaVersions(), AddOffsetsToTxnResponse.schemaVersions()), + END_TXN(26, "EndTxn", false, RecordBatch.MAGIC_VALUE_V2, + "Sent by producer to its transaction coordinator to prepare committing or aborting the current " + + "ongoing transaction. " + + "Specified by KIP-98.", + EndTxnRequest.schemaVersions(), EndTxnResponse.schemaVersions()), + WRITE_TXN_MARKERS(27, "WriteTxnMarkers", true, RecordBatch.MAGIC_VALUE_V2, + "Sent by transaction coordinator to broker to commit the transaction. " + + "Specified by KIP-98.", + WriteTxnMarkersRequest.schemaVersions(), WriteTxnMarkersResponse.schemaVersions()), + TXN_OFFSET_COMMIT(28, "TxnOffsetCommit", false, RecordBatch.MAGIC_VALUE_V2, + "Sent by transactional producers to consumer group coordinator to commit offsets within a single " + + "transaction. " + + "Specified by KIP-98.", + TxnOffsetCommitRequest.schemaVersions(), TxnOffsetCommitResponse.schemaVersions()), + DESCRIBE_ACLS(29, "DescribeAcls", + "Sent by an (admin) client to any broker to request the details of access control lists. " + + "Specified by KIP-140.", + DescribeAclsRequest.schemaVersions(), DescribeAclsResponse.schemaVersions()), + CREATE_ACLS(30, "CreateAcls", + "Sent by an (admin) client to any broker to create access control lists for " + + "managing access to the given entities. " + + "Specified by KIP-140.", + CreateAclsRequest.schemaVersions(), CreateAclsResponse.schemaVersions()), + DELETE_ACLS(31, "DeleteAcls", + "Sent by an (admin) client to any broker to request the deletion of the given access control lists. " + + "Specified by KIP-140.", + DeleteAclsRequest.schemaVersions(), DeleteAclsResponse.schemaVersions()), + DESCRIBE_CONFIGS(32, "DescribeConfigs", + "Sent by an (admin) client to a broker to obtain the configs of the given entites. " + + "For describing broker configs the request must be sent to the required broker. " + + "For describing topic configs the request can be send to any broker. " + + "Specified by KIP-133.", + DescribeConfigsRequest.schemaVersions(), DescribeConfigsResponse.schemaVersions()), + ALTER_CONFIGS(33, "AlterConfigs", + "Sent by an (admin) client to any broker to change the configs of the entities given in the request." + + "Currently only topic config alteration is supported. " + + "Specified by KIP-133.", + AlterConfigsRequest.schemaVersions(), AlterConfigsResponse.schemaVersions()), + ALTER_REPLICA_LOG_DIRS(34, "AlterReplicaLogDirs", + "Sent by an (admin) client to a particular broker to have it change the log directory it uses to " + + "store the logs of the given partitions." + + "Specified by KIP-113.", + AlterReplicaLogDirsRequest.schemaVersions(), AlterReplicaLogDirsResponse.schemaVersions()), + DESCRIBE_LOG_DIRS(35, "DescribeLogDirs", + "Sent by an (admin) client to a particular broker to discover the log dirs available on that broker." + + "Specified by KIP-113.", + DescribeLogDirsRequest.schemaVersions(), DescribeLogDirsResponse.schemaVersions()), + SASL_AUTHENTICATE(36, "SaslAuthenticate", + "Sent by a client to a broker wrapping a SASL authentication message so that the Kafka response can " + + "include a mechanism-independent indication of authentication fail that is distinguishable from " + + "a broker failure by the client." + + "Specified by KIP-152.", + SaslAuthenticateRequest.schemaVersions(), SaslAuthenticateResponse.schemaVersions()), + CREATE_PARTITIONS(37, "CreatePartitions", + "Sent by an (admin) client to the controller to request partitions be added to a topic." + + "Specified by KIP-195", + CreatePartitionsRequest.schemaVersions(), CreatePartitionsResponse.schemaVersions()); private static final ApiKeys[] ID_TO_TYPE; private static final int MIN_API_KEY = 0; @@ -200,20 +320,21 @@ public Struct parseResponse(short version, ByteBuffer buffer) { /** indicates the minimum required inter broker magic required to support the API */ public final byte minRequiredInterBrokerMagic; + public final String doc; 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, String doc, Schema[] requestSchemas, Schema[] responseSchemas) { + this(id, name, false, doc, requestSchemas, responseSchemas); } - 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, String doc, Schema[] requestSchemas, Schema[] responseSchemas) { + this(id, name, clusterAction, RecordBatch.MAGIC_VALUE_V0, doc, requestSchemas, responseSchemas); } ApiKeys(int id, String name, boolean clusterAction, byte minRequiredInterBrokerMagic, - Schema[] requestSchemas, Schema[] responseSchemas) { + String doc, Schema[] requestSchemas, Schema[] responseSchemas) { if (id < 0) throw new IllegalArgumentException("id must not be negative, id: " + id); this.id = (short) id; @@ -240,6 +361,7 @@ public Struct parseResponse(short version, ByteBuffer buffer) { } } this.requiresDelayedAllocation = requestRetainsBufferReference; + this.doc = doc; this.requestSchemas = requestSchemas; this.responseSchemas = responseSchemas; } diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java b/clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java index b5042c3f92a8d..406d11daf929a 100644 --- a/clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java +++ b/clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java @@ -136,6 +136,9 @@ public static String toHtml() { b.append(" API (Key: "); b.append(key.id); b.append("):\n\n"); + if (key.doc != null) { + b.append("

").append(key.doc).append("

\n\n"); + } // Requests b.append("Requests:
\n"); Schema[] requests = key.requestSchemas;