...

• 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-configs.sh, kafka-topics.sh, etc) to talk directly to Zookeeper. 
  • Allows ZNodes to be completely locked down via ACLs
  • Further hides the Zookeeper details of Kafka

A couple of few specific use cases are worth pointing out:

1. The Metadata request exposes topic metadata, but it does not expose topic configs. ListConfigs will make that information available to any client of the Kafka protocol and the AdminClient will expose it to normal users.
2. AlterConfigs would make it possible to update topic configs, but also client and replication quotas via the protocol and the AdminClient.
3. One should be able to tell how a topic or broker is configured including defaults and overrides.

Public Interfaces

ACLs

A new ACL resource type `Configs` will be introduced with Describe, Alter and All operations. Initially, this resource type will work at a global level like the Cluster resource type. That is, the only valid resource name will be "kafka-cluster". This leaves the door open for evolving this to be more fine-grained in the future.

Introduce 2 new ACL operations: ReadConfigs and WriteConfigs.

Protocol APIs

Wire Format types

Entity types will be encoded to INT8 in the following way:

KIP-140: Add administrative RPCs for adding, deleting, and listing ACLs introduced a wire format representation for ResourceType and AclOperation. We weill add new values to both types as follows:

AclOperation

• 0: 

0:

• Unknown

• 1: Any

(for filtering, but currently unused)
• 2: Broker
• 3: Topic
• 4: Client

• 2: All

• 3: Read

• 4: Write

• 5: Create

• 6: Delete

• 7: Alter

• 8: Describe

• 9: ClusterAction

• 10: ReadConfigs (new)

• 11: WriteConfigs (new)

ResourceType

• 0: Unknown

• 1: Any

• 2: Topic

• 3: Group

• 4: Cluster

• 5: Broker (new)

• 6: Client (new)

• 7: User (new)

List Configs

ListConfigs Request (Version: 0) => [entitiesresource]   
  entitiesresource => entityresource_type entityresource_name
    entityresource_type => INT8
    entityresource_name => STRING

Request semantics:

1. Can be sent to any broker
2. If there are multiple instructions for the same entity resource in one request the extra request will be ignored
   • This is because the list of entities resources is modeled server side as a set
   • Multiple entities resources results in the same end goal, so handling this error for the user should be okay
   • This is similar to how delete topics handles requests
3. Entity Valid resource types are "Topic", "Client", "User" and "Broker".
4. If entityresource_type is "Broker" and entityresource_name matches the broker that has received the request, read-only configs for this broker are also returned.
5. The principal must be authorized to "DescribeReadConfigs" on the "ConfigsCluster" resource type or ("DescribeReadConfigs" is also included in the "ClusterAll" resource typeoperation). Unauthorized requests will receive a ConfigsAuthorization ClusterAuthorizationFailed error code.
6. Errors are reported independently per resource.

ListConfigs Response (Version: 0) => error_code [entities]
  error_codeentities => INT16
  entities => entityerror_code resource_type entityresource_name [configs]
	error_code    entity=> INT16
	resource_type => INT8
    entityresource_name => STRING
    configs =>
      config_name => STRING
      config_value => STRING
      read_only => BOOLEAN
      is_default => BOOLEAN
      is_sensitive => BOOLEAN

...

AlterConfigs Request (Version: 0) => [entitiesresources] validate_only
  validate_only => BOOLEAN
  entitiesresources => entityresource_type entityresource_name [configs]
    entityresource_type => INT8
    entityresource_name => STRING
    configs =>
      config_name => STRING
      config_value => STRING

...

1. Can be sent to any broker
2. If there are multiple instructions for the same entity in resource in one request, an InvalidRequestException will be logged on the broker and a single error code for that topic will be returned to the client
   • This is because the list of entities resources is modeled server side as a map with entity as the resource as the key
3. Entity Valid resource types are "Topic", "Client", "User" and "Broker".
4. The principal must principal must be authorized to "AlterWriteConfigs" on the "ConfigsCluster" resource type or ("AlterWriteConfigs" is also included in the "ClusterAll" resource typeoperation). Unauthorized requests will receive a ConfigsAuthorization for all entitiesClusterAuthorizationFailed error code.
5. If an Alter operation is attempted on a read-only config, an UnsupportedOperation error will be returned for the relevant entityresource.
6. The request is not transactional. 
   1. If an error occurs for an entityresource, others could still be updated.
   2. Errors are reported independently per entityresource.
7. For tools that allow users to alter configs, a validation/dry-run mode where validation errors are reported but no creation is attempted is available via the validate_only parameter.

AlterConfigs Response (Version: 0) => [responses]   
  responses => entityresource_type entityresource_name error_code error_message
    entity	resource_type => INT8
    entity	resource_name => STRING
    error_code => INT16
    error_message => STRING

...

package org.apache.kafka.server.policy;

public interface AlterConfigsPolicy extends Configurable, AutoCloseable {

    /**
     * Class containing the create request parameters.
     */
    class RequestMetadata {
        /**
         * Create an instance of this class with the provided parameters.
         *
         * This constructor is public to make testing of <code>AlterConfigPolicy</code> implementations easier.
         */
        public RequestMetadata(ConfigEntityConfigResource entityresource, Map<String, String> configs) { ... }

        /**
         * Return configs in the request.
         */
        public Map<String, String> configs() { ... }
    }

    /**
     * Validate the request parameters and throw a <code>PolicyViolationException</code> with a suitable error
     * message if the alter configs request parameters for the provided entityresource do not satisfy this policy.
     *
     * Clients will receive the POLICY_VIOLATION error code along with the exception's message. Note that validation
     * failure only affects the relevant entityresource, other entitiesresources in the request will still be processed.
     *
     * @param requestMetadata the alter configs request parameters for the provided entityresource.
     * @throws PolicyViolationException if the request parameters do not satisfy this policy.
     */
    void validate(RequestMetadata requestMetadata) throws PolicyViolationException;
}

...

public class AdminClient {
    public ListConfigsResult listConfigs(Collection<ConfigEntity>Collection<ConfigResource> entitiesresources, ListConfigsOptions options);
    public AlterConfigsResult alterConfigs(Map<ConfigEntityMap<ConfigResource, List<Config>>Config> configs, AlterConfigsOptions options);
}

public class ListConfigsOptions { 
    public ListConfigsOptions timeoutMs(Integer timeout);
}

public class ListConfigsResult {
    public Map<ConfigResource, KafkaFuture<Config>> results()
    public KafkaFuture<Map<ConfigEntityKafkaFuture<Map<ConfigResource, Collection<Config>>>Config>> all();
}
 
public class AlterConfigsOptions { 
    public AlterConfigsOptions timeoutMs(Integer timeout);
	public AlterConfigsOptions validateOnly(boolean validateOnly);
}

public class AlterConfigsResult {
    public KafkaFuture<Void> all();
    public Map<ConfigEntityMap<ConfigResource, KafkaFuture<Void>> results();
}
 
public class ConfigEntityConfigResource {
    enum Type {
		CLIENT, BROKER, TOPIC, USER, UNKNOWN;
    }
 
    public ConfigEntityConfigResource(Type type, String name) { ... }
    
    public Type type() { ... }
    public String name() { ... }
}

 
public class Config {
    public Config(Collection<ConfigEntry> configs) { ... }
    public Collection<ConfigEntry> entries() { ... }
}
public class ConfigEntry {
    public ConfigEntry(String name, String value) {
        this(name, value, false, false, false);
    }
    public ConfigEntry(String name, String value, boolean isDefault, boolean isSensitive, boolean isReadOnly) { ... }
    public String name() { ... }
    public String value() { ... }
    public boolean isDefault { ... }
    public boolean isSensitive { ... }
    public boolean isReadOnly { ... }
}

...

Compatibility, Deprecation, and Migration Plan

We are adding a new ACL resource typeoperations, so third-party Authorizer implementations will potentially have to be updated in order to support itthem. Since this only affects newly introduced protocol and AdminClient APIs, it's not a compatibility issue. It simply means that the new functionality won't be available for users of such Authorizer implementations until they are updated.

With regards to forwards compatibility, ConfigEntityConfigResource.Type has an UNKNOWN element in case it receives data from a newer broker that cannot be mapped to one of the existing enum types.

...

1. Allowing sensitive data to be returned: it's good security practice to never expose sensitive data. If necessary, the user can reset the relevant sensitive data (e.g. a password).
2. Using a string for the wire representation of entity_type: it would be inconsistent with the approach used by KIP-140: Add administrative RPCs for adding, deleting, and listing ACLsIntroducing a new Configs resource instead of ReadConfigs and WriteConfigs operations: there is always a one to one mapping between a resource and its configs, so there isn't much value in creating a separate resource for Configs. By adding new operations to existing resources, it's easier to see all the ACLs that affect a given resource.

Future work

1. Forward requests to the relevant brokers in order to return `read-only` broker configs.
2. Finer grained ACLs for More fine-grained ACLs so that a user authorized to ReadConfigs/WriteConfigs on a given resource type (topic, broker, etc.) can read/write the relevant configs.