THIS IS A TEST INSTANCE. ALL YOUR CHANGES WILL BE LOST!!!!
...
Extending protocol with these messages:
- CreateTopic(Request | Response), AlterTopic, DeleteTopic, ListTopics, DescribeTopic
- PreferredLeaderReplicaElection
- ReassignPartitions Evolving TopicMetadataRequest to V1
- New client: AdminClient - a Wire Protocol client for administrative operations
- New CLI tool: Kafka Shell - a single unified command line interface for administrative operations
...
Overall the idea is to extend existing Wire Protocol to cover all existing topic commands - create-topic, alter-topic, delete-topic, describe-topic, list-topics, reassign-partitions, preferred-replica-leader-election. At the same time, since Wire Protocol is a public API to Kafka cluster, it was agreed that the new Admin schema needs to be "orthogonal", i.e. new messages shouldn't duplicate each other or existing messages, if those already cover particular use cases. It is proposed to map existing command to Wire Protocol in the following way:
| Command | Wire Protocol Message | Note |
|---|---|---|
| create-topic | CreateTopicRequest | |
| alter-topic | AlterTopicRequest | |
| delete-topic | DeleteTopicRequest | |
| describe-topic | TopicMetadataRequest_V1 | Using new version on TopicMetadataRequest which will include topic-level config |
| list-topics | TopicMetadataRequest | Using an empty list as input for request, which results in returning metadata for all existing topcis |
| reassign-partitions | AlterTopicReqeust | Using batch AlterTopicRequest with replica assignment specified |
| preferred-replica-leader-election | PreferredReplicaLeaderElectionRequest |
It is also important that all Admin requests are intended to be asynchronous. This means requests will only initiate particular command, and will not wait until the command is actually completed. There are different reasons to that, but we wan't to make sure that we give users all needed tools to verify whether issued command has been completed. E.g. one can consider topic is created once it is possible to consumer from / produce to this topic. In this case user can leverage TopicMetadataRequest to check all brokers received metadata about the newly created topic.
Finally, Topic Admin schema requests are likely to be used not only in CLI tool, where the common use case in is create/change/delete /get a single entity. Since Kafka is able to maintain a huge number of topics it is important that user can efficiently issue many requests at one time. That's why all Topic Admin messages essentially are batch requests, i.e. it is possible to group commands of one type for many topics in one batch reducing network calls.
...
New Protocol Errors
It is proposed to add these error codes to the protocol.
Error | Description | Requests |
|---|---|---|
TopicAlreadyExists | Topic with this name already exists. | CreateTopicRequest |
InvalidArgumentTopicName | Topic name contains invalid characters. | CreateTopicRequest |
InvalidArgumentPartitions | Either partition field is invalid (e.g. negative), or not defined when needed. | CreateTopicRequest, AlterTopicRequest |
InvalidArgumentReplicationFactor | Either replication-factor field is invalid (e.g. negative), or not defined when needed. | CreateTopicRequest, |
InvalidArgumentReplicaAssignment | Either replication -factor assignment field is invalid (e.g. contains duplicates), or not defined when needed. |
|
InvalidArgumentTopicConfig | Either topic-level config setting or value is incorrect. | CreateTopicRequest, AlterTopicRequest |
DecreasePartitionsNotAllowed | Invalid partitions argument: decreasing partitions is prohibited when altering topic. | AlterTopicRequest |
PreferredReplicaLeaderElectionInProgress | Preferred replica leader election procedure has been already started. | PreferredReplicaLeaderElectionRequest |
ReassignPartitionsInProgress | Reassign partitions procedure has been already started. | AlterTopicRequest |
| MultipleInstructionsForOneTopic | Only one mutation is allowed at once: e.g. change topic replication factor or change topic config. | CreateTopic, AlterTopicRequest |
MultipleTopicInstructionsInOneBatch | Multiple topic instructions for the same topic in one batch request | CreateTopicRequest, AlterTopicRequest, DeleteTopicRequest |
Generally, the Admin Client (see section 3) or another request dispatcher should have enough context to provide descriptive error message.
The same notation as in A Guide To The Kafka Protocol is used here.
...
CreateTopicRequest => [TopicName Partitions Replicas ReplicaAssignment]TopicName => stringPartitions => int32Replicas => int32ReplicaAssignment => [PartitionId [ReplicaId]] |
Request semantics:
User can define only one from (
Partitions + Replicas), ReplicaAssignment in one instruction. (Note: there is a special use case - automatic topic creation for TopicMetadataRequest, to trigger it user should set client_id=consumer and define only topic name). Otherwise, MultipleInstructionsForOneTopic is returned.In case
ReplicaAssignment is defined number of partitions and replicas will be extracted calculated from the supplied ReplicaAssignment. In case of defined (Partitions + Replicas) replica assignment will be automatically generated by the server.One
CreateTopicRequest may include only one topic creation command for the topic with the given name in one batch, otherwise MultipleTopicInstructionsInOneBatch is returned....
CreateTopicResponse => [TopicName ErrorCode]ErrorCode => int16TopicName => string |
CreateTopicResponse contains a map between topic and topic creation result error code (see New Protocol Errors).
Alter Topic Request
AlterTopicRequest => [TopicName Partitions Replicas ReplicaAssignment [AddedConfigEntry] [DeletedConfig]]TopicName => string Replicas => int32 Partitions => int32ReplicaAssignment => [PartitionId [ReplicaId]]AddedConfigEntry => ConfigKey ConfigValue ConfigKey => string ConfigValue => string DeletedConfig => string |
AlterTopicRequest will be used to change topic's replication characteristics, alter topic-level config or reassign partitions.
Request semantics:
User can define only one from Partitions, Replicas, ReplicaAssignment, AddedConfigEntry, DeletedConfig. Otherwise, MultipleInstructionsForOneTopic is returned.
One AlterTopicRequest may include only one topic alteration command for the topic with the given name in one batch, otherwise MultipleTopicInstructionsInOneBatch is returned.
Alter Topic Response
ErrorCode => int16TopicName => string |
AlterTopicResponse is similar to CreateTopicResponse.Delete Topic Request
DeleteTopicRequest => [TopicName]TopicName => string |
DeleteTopicRequest requires only topic names which should be deleted.One
DeleteTopicRequest may include only one topic deletion command for the topic with the given name in one batch, otherwise MultipleTopicInstructionsInOneBatch is returned.Delete Topic Response
ErrorCode => int16TopicName => string |
DeleteTopicResponse is similar to CreateTopicResponse.
Topic Metadata Request V1
TopicMetadataReqeust_V1 => [TopicName]TopicName => string |
TopicMetadataRequest_V1 is an evolved version of TopicMetadataRequest. This request is intended to support two admin operations - get topic metadata, and check whether some particular admin command (which are all asynchronous) has been completed.TopicMetadataRequest_V1 requires only topic names. As with the first version, an empty topic name set results in returning information for all existing topics.Also
TopicMetadataRequest_V1 comparing to the previous version won't trigger topic creation automatically if the topic with the given name doesn't exist.Topic Metadata Response V1
TopicMetadataResponse_V1 => [Broker][TopicMetadata]Broker => NodeId Host Port (any number of brokers may be returned)NodeId => int32Host => stringPort => int32TopicMetadata => TopicErrorCode TopicName [PartitionMetadata] [ConfigEntry]TopicErrorCode => int16 PartitionMetadata => PartitionErrorCode PartitionId Leader ReplicasLag IsrPartitionErrorCode => int16PartitionId => int32Leader => int32ReplicasLag => [int32 int32]Isr => [int32]ConfigEntry => string string |
The new version of TopicMetadataResponse in addition to TopicMetadataResponse_V0 will include topic level configuration for each topic and replica fetch lag per partition - how far partition replica is behind from the leader replica.
Replication Commands Schema
Preferred Replica Leader Election Request
PreferredReplicaLeaderElectionRequest => [Topic [PartitionId]]Topic => stringPartitionId => int32 |
PreferredReplicaLeaderEleactionRequest initiates preferred replica leader election procedure, similar . Similar to Topic Admin requests this request in intended to be non-blocking. The schema consist of one field - array of partitions for which preferred replica leader should be elected.To start preferred replica leader election procedure for all existing partition an empty partitions array should be sent.
Preferred Replica Leader Election Response
PreferredReplicaLeaderElectionResponse => [Topic ErrorCode]Topic => stringErrorCode => int16 |
PreferredReplicaLeaderElectionResponse is similar to ReassignPartitionsResponse.
Status of the procedure may be checked with TopicMetadataRequest - the head of replicas list field and leader broker should be the same.
...