...

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

• 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

• General 

  • 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

• Topic Admin Schema

  • 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?

• 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

...

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 Topic Request

CreateTopic 

Request (Version: 0) => [create_topic_requests] timeout 
  

create_topic_requests => 

topic partitions replication_factor [replica_assignment] [configs] 
    topic => STRING
    

partitions => 

INT32
    

replication_

factor => 

INT32
   

 replica_assignment => 

List ACLs Response

ListAclsResponse => [responses] error_code 
  responses => resource [acls] 
    resource => resource_type resource_name 
      resource_type => INT8
      resource_name => STRING
    acls => acl_principle acl_permission_type acl_host acl_operation 
      acl_principle => STRING
      acl_permission_type => INT8
      acl_host => STRING
      acl_operation => INT8
  error_code => INT16

AlterAclsRequest => [requests]   requests => resource [actions] 
    resource => resource_type resource_name 
      resource_type => INT8
      resource_name => STRING
    actions => action acl 
      acl => acl_principle acl_permission_type acl_host acl_operation 
        acl_principle => STRING
        acl_permission_type => INT8
        acl_host => STRING
        acl_operation => INT8
      action => INT8

Alter ACLs Response

AlterAclsResponse => [responses]       
  responses => resource [results]     
    resource => resource_type resource_name 
      resource_type => INT8
      resource_name => STRING
    results => action acl error_code 
      acl => acl_principle acl_permission_type acl_host acl_operation 
        acl_principle => STRING
        acl_permission_type => INT8
        acl_host => STRING
        acl_operation => INT8
      action => INT8
      error_code => INT16

Topic Admin Schema

Create Topic Request

partition_id [replicas] 
      partition_id => INT32
      replicas => INT32
    configs => config_key config_value 
      config_key => STRING
      config_value => STRING
  timeout => INT32

Create Topic Response

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

Create Topic Response

...

CreateTopicResponse contains a map between topic and topic creation result error code (see New Protocol Errors). 

Alter Topic Request

Alter Topic Response

...

Delete Topic Request

Delete Topic Response

...

DeleteTopicResponse is similar to CreateTopicResponse.

2. Server-side Admin Request handlers

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. 

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 (e.g. reassign partitions).

Since Topic commands will support batching (and so will AdminClient) user besides Admin API will be provided with request builders which will help to create requests correctly.

Proposed API:

...

...

public Map<String, DescribeTopicOutput> describeTopics(List<String> topicNames) throws ApiException;

/**
* Initiates config alteration. This is an asynchronous call, it returns immediately once the server has accepted request and stored/changed respective data in zookeeper.
* To simulate a simple blocking call Future.get can be called. This will ensure that updated configs were persisted
*

* @param entityType Type of entity being described (topic, client etc..)
* @param alterConfigRequest holder (built by means of respective Builder) of all required arguments to alter topics
* @return java.util.concurrent.Future which holds topics alteration result - a map topic-name - complete after image of the entity configs
*
* @throws ApiException in case of the configs could be altered for all topics
*/
public Future<Map<String, EntityConfig>> alterConfig(String entityType, AlterConfigRequest) throws ApiException;

/**

* Describes config for any entity

* @param entityType Type of entity being described (topic, client etc..)

* @param entityNames Array of entity names to describe (topic names, client id's etc)

* @return a mapping between entity name and it's config. If config cannot be fetched for a particular entity, an error value of EntityConfig is returned
* @throws ApiException in case config cannot be fetched for all topics
*/
public Map<String, EntityConfig> describeConfig(String entityType, List<String> entityNames) throws ApiException;

...

ListAclsRequest => principal resource 
  resource => resource_type resource_name 
    resource_type => INT8
    resource_name => STRING
  principal => NULLABLE_STRING

ListAclsResponse => [responses] error_code 
  responses => resource [acls] 
    resource => resource_type resource_name 
      resource_type => INT8
      resource_name => STRING
    acls => acl_principle acl_permission_type acl_host acl_operation 
      acl_principle => STRING
      acl_permission_type => INT8
      acl_host => STRING
      acl_operation => INT8
  error_code => INT16

AlterAclsRequest => [requests]   requests => resource [actions] 
    resource => resource_type resource_name 
      resource_type => INT8
      resource_name => STRING
    actions => action acl 
      acl => acl_principle acl_permission_type acl_host acl_operation 
        acl_principle => STRING
        acl_permission_type => INT8
        acl_host => STRING
        acl_operation => INT8
      action => INT8

AlterAclsResponse => [responses]       
  responses => resource [results]     
    resource => resource_type resource_name 
      resource_type => INT8
      resource_name => STRING
    results => action acl error_code 
      acl => acl_principle acl_permission_type acl_host acl_operation 
        acl_principle => STRING
        acl_permission_type => INT8
        acl_host => STRING
        acl_operation => INT8
      action => INT8
      error_code => INT16

2. Server-side Admin Request handlers

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. 

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 (e.g. reassign partitions).

Since Topic commands will support batching (and so will AdminClient) user besides Admin API will be provided with request builders which will help to create requests correctly.

Proposed API: TBD

...

...