...

The goals behind the command line shell are fundamentally to provide a centralized management for Kafka operations.

Status

Current state:  Under Discussion Accepted

Discussion thread: here

JIRA: here 

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

Folks Users of Kafka have created dozens of different systems to work with Kafka. If we can provide Providing a wire protocol that allows the brokers to execute administrative code then clients can more simply execute commands. We have a lot of ties with zookeeper making it near impossible for some code to-do nothing better/more than shell script wrapping kafka-topics.sh etc. With the wire protocol we should be able to have client in any language work with administrating Kafka. If someone wants a REST interface (for example) then can write that in whatever language they like. We should have a client from the project that is not only an example but a fully functionality replacement for the kafka-topics, reassign-partitions, consumer offset tools.

Public Interfaces

Briefly list any new interfaces that will be introduced as part of this proposal or any existing interfaces that will be removed or changed. The purpose of this section is to concisely call out the public contract that will come along with this feature.

A public interface is any change to the following:

• Command line tools and arguments

Proposed Changes

Proposed changes include 4 parts:

1. Wire Protocol extensions - to add new Admin messages
2. Server-side Admin commands handlers (TopicCommand-like)
3. Admin Client - an out-of-box client for performing administrative commands

4. Interactive Shell / CLI tool supporting administrative commands

Some open questions and items under discussion are marked with [x]. Please find Open Questions section for more details.

1. Wire Protocol Extensions

Overview

For each type of Admin Request a separate type of Wire protocol message is created.

Currently there are 3 types of requests:

• Topic commands which include CreateTopic(Request | Response), AlterTopic, DeleteTopic, DescribeTopic, ListTopics.
• Replication tools - ReassingPartition, VerifyReassingPartitions; PreferredReplicaLeaderElection, VerifyPreferredReplicaLeaderElection
• A special type of request to support Admin commands - ClusterMetadata

Please find details under specific RQ/RP schema proposal.

Schema

The same notation as in  A Guide To The Kafka Protocol is used here. The only difference - new Kafka Protocol metatype - MaybeOf ("?" in notation), when used means value is optional in message. To define value existence special control byte is prepended before each value (0 - field is absent, otherwise - read value normally) [1].

Create Topic Request

Create Topic Response

...

CreateTopicRequest requires topic name and either (partitions+replicas) or replicas assignment to create topic (validation is done on server side). You can also specify topic-level configs to create topic with (to use default set an empty array).

CreateTopicResponse is fairly simple - you receive error code (0 as always identifies NO_ERROR) and optionally error description. Usually it will hold the higher level exception that happened during command execution.

Alter Topic Request

Alter Topic Response

...

AlterTopicResponse is similar to CreateTopicResponse.

Delete Topic Request

Delete Topic Response

...

DeleteTopicResponse is similar to CreateTopicResponse.

Describe Topic Request

Describe Topic Response

List Topics Request

List Topics Response

...

ListTopicsResponse besides errorCode and optional errorDescription which are used in the same way as in previous messages, holds optional (non empty if execution was successful) two list of topic names - one for deleted topics (marked for deletion) and the second one for ordinary, alive topics.

Replication Commands Schema

Reassign Partitions

Reassign Partitions Request

Reassign Partitions Response

...

ReassignPartitionsRequest requires manual partition assignment string. Parsing / validation is done on server. This request will only initiate partition reassignment and return immediately. It is client's responsibility to block the user continually sending VerifyReassignPartitionsRequest to check reassignment status. The format is the following:

{

"partitions": [

{"topic": "foo",
 "partition": 1,
 "replicas": [1,2,3] }

],
 "version":1
}

ReassignPartitionResponse is similar to CreateTopicResponse.

 code and public api/client has many benefits including:

• Allows clients in any language to administrate Kafka
  • Wire protocol is supported by any language
• Provides public client for performing admin operations
  • Ensures integration test code in other projects and clients maintains compatibility
  • Prevents users from needing to use the Command classes and work around standard output and system exits
• Removing the need for admin scripts (kafka-topics.sh, kafka-acls.sh, etc) to talk directly to Zookeeper. 
  • Allows ZNodes to be completely locked down via ACLs
  • Further hides the Zookeeper details of Kafka

Public Interfaces

• Changes to Wire Protocol:
  • Adds the following new Request/Response messages:
    • CreateTopics
    • AlterTopics
    • DeleteTopics
    • ListAcls
    • AlterAcls
    • DescribeConfig (moved to KIP-133: Describe and Alter Configs Admin APIs)
    • AlterConfig (moved to KIP-133: Describe and Alter Configs Admin APIs)
  • Modifies Metadata Request/Response to allowing polling for in-progress or complete admin operations. Added fields include:
    • Add the ability to request no topics with a null topics list
    • Boolean indicating if a topic is marked for deletion
    • Boolean indicating if a topic is an internal topic
    • Rack information (if not added by KIP-36 Rack aware replica assignment)
    • Boolean indicating if a broker is the controller

Proposed Changes

Proposed changes include 2 parts:

1. Wire protocol additions and changes
2. Server-side message handlers and authorization

Follow Up Changes

Changes that should be considered shortly after or are enabled by this KIP included:

• General

  • New Java AdminClient implementation (KIP-117)
  • Refactor admin scripts and code to use new client where appropriate
  • Support forwarding requests to the required broker (coordinator, group leader, partition leader) (KAFKA-1912)
    • See Request Forwarding below

• Metadata Schema

  • Consider supporting regex topic filters in the request 
  • Filter internal topics using the returned metadata (

    Jira
    server ASF JIRA serverId 5aa69414-a9e9-3523-82ec-879b028fb15b key KAFKA-3954

    )

• Topic Admin Schema

  • Improve the broker side delete topic implementation
    • Delete is likely to get used more once creation/deletion is made easier with the client. The broker side implementation has had many jiras. 
    • Currently can't delete unhealthy topics. 
  • Support cluster consistent blocking to wait until all relevant brokers have the required metadata
    • This may require significant re-work of the controller to do correctly
    • See Cluster Consistent Blocking below
  • Implement auto-topic creation client side (KAFKA-2410)
  • Add topic creation to the MirrorMaker client?
  • Support renaming topics (KAFKA-2333)
    • This might required unique ids for topics instead of using the name (this would improve delete too)
  • Improve reliability and speed of topic deletion
    • Support force deleting unhealthy topics
    • Support marking for deletion and async data cleanup
      • This would required unique ids for topics instead of using the name (this is needed for rename too)
      • The topic can then be marked as deleted instead of requiring all data to be removed immediately and in the mean time a new topic with the same name can be created. 

• ACL Admin Schema

  • Review privileges for listing and altering ACLs to be more fine grained.
  • Provide an Authorizer interface using the new Java classes used by the ACL requests/responses (KAFKA-3509)
    • Deprecate the old one to encourage transition
  • Define standard Exceptions that can be thrown by the Authorizer in the interface (KAFKA-3266)
    • Otherwise all exceptions are unknown server exception to the client
  • Consider building a sync call into the Authorizer to ensure changes are propagated

Details

1. Wire Protocol Extensions

Schema

TopicExistsException

InvalidPartitionsException

InvalidReplicationFactorException

InvalidReplicaAssignmentException

InvalidConfigurationException

NotControllerException

Metadata Schema (Voted and Adopted in 0.10.0.0)

Metadata Request (version 1)

MetadataRequest => [topics] 

Metadata Response (version 1)

MetadataResponse => [brokers] controllerId [topic_metadata]   
  brokers => node_id host port rack 
    node_id => INT32
    host => STRING
    port => INT32
    rack => NULLABLE_STRING
  controllerId => INT32
  topic_metadata => topic_error_code topic is_internal [partition_metadata] 
    topic_error_code => INT16
    topic => STRING
    is_internal => BOOLEAN
    partition_metadata => partition_error_code partition_id leader [replicas] [isr] 
      partition_error_code => INT16
      partition_id => INT32
      leader => INT32
      replicas => INT32
      isr => INT32

Topic Admin Schema

Create Topics Request (KAFKA-2945): (Voted and Committed for 0.10.1.0)

CreateTopics Request (Version: 0) => [create_topic_requests] timeout 
  create_topic_requests => topic num_partitions replication_factor [replica_assignment] [configs] 
    topic => STRING
    num_partitions => INT32
    replication_factor => INT16
    replica_assignment => partition_id [replicas] 
      partition_id => INT32
      replicas => INT32
    configs => config_key config_value 
      config_key => STRING
      config_value => STRING
  timeout => INT32

Create Topics Response

CreateTopics Response (Version: 0) => [topic_error_codes] 
  topic_error_codes => topic error_code 
    topic => STRING
    error_code => INT16

DeleteTopics Request (Version: 0) => [topics] timeout 
  topics => STRING
  timeout => INT32

DeleteTopics Response (Version: 0) => [topic_error_codes] 
  topic_error_codes => topic error_code 
    topic => STRING
    error_code => INT16

ListAcls Request (Version: 0) => principal resource 
  principal => NULLABLE_STRING  
  resource => resource_type resource_name 
    resource_type => INT8
    resource_name => STRING

ListAcls Response (Version: 0) => [responses] error_code 
  responses => resource [acls] 
    resource => resource_type resource_name 
      resource_type => INT8
      resource_name => STRING
    acls => acl_principal acl_permission_type acl_host acl_operation 
      acl_principal => STRING
      acl_permission_type => INT8
      acl_host => STRING
      acl_operation => INT8
  error_code => INT16

AlterAcls Request (Version: 0) => [requests] 
  requests => resource [actions]     
    resource => resource_type resource_name 
      resource_type => INT8
      resource_name => STRING
    actions => action acl 
      action => INT8
      acl => acl_principal acl_permission_type acl_host acl_operation 
        acl_principal => STRING
        acl_permission_type => INT8
        acl_host => STRING
        acl_operation => INT8

AlterAcls Response (Version: 0) => [responses] 
  responses => resource [results] 
    resource => resource_type resource_name 
      resource_type => INT8
      resource_name => STRING
    results => action acl error_code 
      action => INT8
      acl => acl_principal acl_permission_type acl_host acl_operation 
        acl_principal => STRING
        acl_permission_type => INT8
        acl_host => STRING
        acl_operation => INT8
      error_code => INT16

At the highest level, admin requests will be handled on the brokers the same way that all message types are. However, because admin messages modify cluster metadata they should be handled by the controller. This allows the controller to propagate the changes to the rest of the cluster.  However, because the messages need to be handled by the controller does not necessarily mean they need to be sent directly to the controller. A message forwarding mechanism can be used to forward the message from any broker to the correct broker for handling. 

Because supporting all of this is quite the undertaking I will describe the "ideal functionality" and then the "intermediate functionality" that gets us some basic administrative support quickly while working towards the optimal state. 

Ideal Functionality:

1. A client sends an admin request to any broker
2. The admin request is forwarded to the required broker (likely the controller)
3. The request is handled and the server blocks until a timeout is reached or the requested operation is completed (failure or success)
   1. An operation is considered complete/successful when all required nodes have the correct/current state. 
   2. Immediate follow up requests to any broker will succeed.
   3. Requests that timeout may still be completed after the timeout. The users would need to poll to check the state. 
4. The response is generated and forwarded back to the broker that received the request.
5. A response is sent back to the client. 

Intermediate Functionality:

1. A client sends an admin write requests to the controller broker. Read requests can still go to any broker. 
   1. As a follow up request forwarding can be added transparently. (see below)
2. The request is handled and the server blocks until a timeout is reached or the requested operation is completed (failure or success)
   1. An operation is considered complete/successful when the controller node has the correct/current state.
   2. Immediate follow up requests to the controller will succeed. Others (not to the controller) are likely to succeed or cause a retriable exception that would eventually succeed. 
   3. Requests that timeout may still be completed after the timeout. The users would need to poll to check the state. 
3. A response is sent back to the client. 

The ideal functionality has 2 features that are more challenging initially. For that reason those features will be removed from the initial changes, but will be tracked as follow up improvements. However, this intermediate solution should allow for a relatively transparent  transition to the ideal functionality. 

Request Forwarding: KAFKA-1912

Request forwarding is relevant to any message the needs to be sent to the "correct" broker (ex: partition leader, group coordinator, etc). Though at first it may seam simple it has many technicall challenges that need to be decided in regards to connections, failure, retries, etc. Today, we depend on the client to choose the correct broker and clients that want to utilize the cluster "optimally" would likely continue to do so. For those reasons it can be handled it can be handled generically as an independent feature. 

Cluster Consistent Blocking:

Blocking an admin request until the entire cluster is aware of the correct/current state is difficult based on Kafka's current approach for propagating metadata. This approach varies based on the the metadata changing.

• Topic metadata changes are propagated via UpdateMetadata and LeaderAndIsr requests
• Config changes are propagated via zookeeper and listeners
• ACL changes depend on the implementation of the Authorizer interface 
  • The default SimpleACLAuthorizer uses zookeeper and listeners

Though all of these mechanisms are different, they are all commonly "eventually consistent". None of the mechanisms, as currently implemented, will block until the metadata has been propagated successfully. Changing this behavior would require a large amount of change to the KafkaController, additional inter-broker messages, and potentially a change to the Authorizer interface. These are are all changes that should not block the implementation of KIP-4.

The intermediate changes in KIP-4 should allow an easy transition to "complete blocking" when the work can be done. This is supported by providing optional local blocking in the mean time. This local blocking only blocks until the local state on the controller is correct. We will still provide a polling mechanism for users that do not want to block at all. A polling mechanism is required in the optimal implementation too because users still need a way to check state after a timeout occurs because operations like "create topic" are not transactional. Local blocking has the added benefit of avoiding wasted poll requests to other brokers when its impossible for the request to be completed. If the controllers state is not correct, then the other brokers cant be either. Clients who don't want to validate the entire cluster state is correct can block on the controller and avoid polling all together with reasonable confidence that though they may get a retriable error on follow up requests, the requested change was successful and the cluster will be accurate eventually.

Because we already add a timeout field to the requests wire protocols, changing the behavior to block until the cluster is consistent in the future would not require a protocol change. Though the version could be bumped to indicate a behavior change. 

Verify Reassign Partitions Request

Verify Reassign Partitions Response

Preferred Replica Leader Election

Preferred Replica Leader Election Request

Preferred Replica Leader Election Response

...

{"partitions":[

{"topic": "foo", "partition": 1},
{"topic": "foobar", "partition": 2}

]
}

PreferredReplicaLeaderElectionResponse is similar to CreateTopicResponse.

Status of the procedure may be checked with DescribeTopicRequest  - the head of replicas list field and leader broker should be the same.

2. Server-side Admin Request handlers

All incoming request will be handled by specific helper classes called from KafkaApis - TopicCommandHelper for topic admin commands, ReassignPartitionsCommandHelper and PreferredReplicaLeaderElectionCommandHelper.

All these commands are already implemented as standalone CLI tools, so there is no need to re-implement them. Unfortunately most of command classes are strongly coupled with CLI logic and can hardly be refactored, so for now (before we remove standalone CLI commands)  most likely the logic from those classes will be extracted and copied  to separate classes (as proposed - TopicCommandHelper etc).

3. Admin Client

This component is intended to be a Kafka out-of-box client implementation for Admin commands.

Admin client will use Kafka NetworkClient facility from /clients for cluster communication. Besides Admin commands, client will handle cluster metadata cache and will provide user with a convenient way of handling long running commands (as e.g. reassign partitions).

Proposed API:

 public class AdminClient {
    
	/**
     * A producer is instantiated by providing a set of key-value pairs as configuration. Most
	 * of the settings will be related to NetworkClient
     *
     * @param properties settings related to Network client and at least one broker from KafkaCluster to connect to
     */
    public AdminClient(Properties properties) 
    
	/**
     * Create topic with given number of partitions and replication factor, replica assignment will be handled by Kafka cluster
     *
     * @throws ApiException
     */
    public void createTopic(String topicName, int partitions, int replicationFactor, List<ConfigEntry> configs) throws ApiException;
    
	/**
     * Create topic with specified replica assignment (number of partitions and replication factor will be taken
     * from replica assignment string)
     *
     * @throws ApiException
     */
    public void createTopic(String topicName, String replicaAssignment, List<ConfigEntry> configs) throws ApiException;

    /**
     * Alter existing topic partitions and/or replica assignment among Kafka brokers
     *
     * @throws ApiException
     */
    public void alterTopic(String topicName, Integer partitions, String replicaAssignment,
                                    List<ConfigEntry> addedConfigs, List<String> deletedConfigs) throws ApiException;
    /**
     * Delete Kafka topic by name
     *
     * @throws ApiException
     */
    public void deleteTopic(String topicName) throws ApiException;
    
	/**
     * List all existing topics in Kafka cluster
     *
     * @throws ApiException
     */
    public List<String> listTopics() throws ApiException;
    
	/**
     * Request replication information about Kafka topic
     *
     * @throws ApiException
     */
    public DescribeTopicOutput describeTopic(String topicName) throws ApiException;
    
	/**
     * Initiate long-running reassign partitions procedure
     *
     * @param partitionsReassignment manual partitions assignment string (according to ReassignPartitionsCommand)
     * @return future of the reassignment result which is completed once server-side partitions reassignment has succeeded or
     * an error occurred so that partitions reassignment cannot be started
     * @throws ApiException
     */
    public Future<ReassignPartitionsResponse> reassignPartitions(String partitionsReassignment) throws ApiException;

    /**
     * Check the interim status of the partitions reassignment
     *
     * @param partitionsReassignment manual partitions assignment string (according to ReassignPartitionsCommand)
     * @return partition to reassignment result code (completed, in-progress, failed)
     * @throws ApiException
     */
    public Map<TopicPartition, Short> verifyReassignPartitions(String partitionsReassignment) throws ApiException;
    
	/**
     * Initiate long-running preferred replica leader election procedure
     *
     * @param partitions serialized partitions for which preferred replica leader election will be started
     *                   (according to PreferredReplicaLeaderElectionCommand)
     * @return future of the election result which is completed once server-side preferred replica is elected for provided partitions or
     * an error has occurred
     * @throws ApiException
     */
    public Future<PreferredReplicaLeaderElectionResponse> preferredReplicaLeaderElection(String partitions) throws ApiException;

    /**
     * Check the interim status of the preferred replica leader election
     *
     * @param partitions for which preferred replica leader election was started (according to PreferredReplicaLeaderElectionCommand)
     * @return partition to reassignment result code (completed, in-progress, failed)
     * @throws ApiException
     */
    public VerifyPreferredReplicaLeaderElectionResponse verifyPreferredReplicaLeaderElection(String partitions)
            throws ApiException;
	/**
     * A generic facility to send Admin request and return response counterpart
     *
     * @param adminRequest AdminRequest message
     * @param <T>          concrete AdminRequest type
     * @return response counterpart
     * @throws ApiException
     */
    private <T extends AbstractAdminResponse> T sendAdminRequest(AbstractAdminRequest<T> adminRequest) throws ApiException;

 
	/**
     * Refreshes cluster metadata cache - list of brokers and controller
     * 
     * @throws ApiException
     */
    private void updateClusterMetadata() throws Exception;

}

4. Interactive Shell / CLI tool

This component will wrap AdminClient and provide an interactive shell-like environment for executing administrative commands. The goal of these changes is let people use existing standalone tools but from a single script, optionally running commands in shell, so commands arguments/names are not changed comparing to existing tools, where possible.

The tool will be run in two modes:

...

Installation

This is an instruction how to build and start Kafka Command Line Tool (hereinafter - Shell). The implementation is in progress under KAFKA-1694.

To start Shell you need to have a running Kafka Cluster built from the given patch (attached under KAFKA-1694) and build the Shell itself.

Get the code.
Get the KAFKA-1772_1802_1775_1774_v2.patch attached to KAFKA-1694.
The patch was built against trunk, on top of revision 7e9368b . So reset to this commit and then run to apply the patch:

git am KAFKA-1772_1802_1775_1774_v2.patch

Build the code. Run:

./gradlew releaseTarGz_2_10_4

...

Sample usage

You can use Kafka Command Line Tool in two ways: 1) as a interactive shell 2) as a simple CLI.

E.g. to get list of topics you can:

1) Start Shell and run:

sudo bin/kafka.sh --shell --broker <host : port>

kafka> list-topics

Or

2) Run right from kafka.sh:

sudo bin/kafka.sh --list-topics --broker <host : port>

Compatibility, Deprecation, and Migration Plan

• When will we remove the existing behavior?

I don't know if that has to be decided now. Folks have already built wrapper tools, they can still keep using them if they want. We should code freeze them though.

Rejected Alternatives

If there are alternative ways of accomplishing the same thing, what were they? The purpose of this section is to motivate why the design is the way it is and not some other way.

TopicMetadataRequest/Response: After some debate we decided not to evolve the TopicMetadataResponse to remove the ISR field (which currently can return incorrect information). There is a use-case for this in KAFKA-2225, so we will treat this a bug and fix it going forward. See KAFKA-1367 for more details