Versions Compared

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

Table of Contents

Status

Current state: Under discussionAccepted - 2.6.0 contains describe and alter functionality, resolve is pending for a future release.

Discussion thread: TODO here

JIRA: KAFKA-7740

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

Motivation

Quota management via Admin Client has gone through a couple drafts of proposals (KIP-248, KIP-422). While improvements have been made to the Admin interface for configuration handling, fitting quotas into the API is awkward as they don't fit the natural key-value pairing, nor is the configuration output expressive enough to return all useful information. Therefore, it'd be beneficial to have a quota-native API for managing quotas, which would offer an intuitive and less error-prone interface, convey additional information beyond what the configuration APIs provide, and enable for future extensibility as quotas types are added or evolved.

Background

By default, quotas are defined in terms of a user and client ID, where the user acts as an opaque principal name, and the client ID as a generic group identifier. When setting quotas, an administrator has flexibility in how it specifies the user and client ID for which the quota applies, where the user and client ID may be specifically named, indicated as a default, or omitted entirely. Since quotas have flexible configurations, there is a method for resolving the quotas that apply to a request: a hierarchy structure is used, where the most specific defined quota will be matched to a request's user and client ID.

As represented by the current ZK node structure, the order in which quotas are matched are as follows. Note , from highest priority to lowest (note <user> is a specified user principal, <client-id> is a specified client ID, and <default> is a special default user/client ID that matches to all users or clients IDs.):

        /config/users/<user>/clients/<client-id>
        /config/users/<user>/clients/<default>
        /config/users/<user>
        /config/users/<default>/clients/<client-id>
        /config/users/<default>/clients/<default>
        /config/users/<default>
        /config/clients/<client-id>
        /config/clients/<default>

As such, reasoning around quotas can be complex, as it's not immediately obvious which quotas may apply to a given user and/or client ID. Providing descriptive information about how quotas are matched is the first goal of this KIP. Likewise, retrieving and modifying quota values can be done in a more expressive and robust way, which is the second goal of the KIP.

APIs

In order to clearly specify the APIs, let's first disambiguate some terminology: Every client request that is processed is associated with a quota entity, which is a map of entity types to their corresponding entity names for the request. Using the current entity types, an entity is of the form {user=test-user, client-id=test-client}, where user and client-id are the types, and test-user and test-client are the names. However, when specifying a quota configuration entry, only a subset of the entity types need to be provided, which is referred to as an entity match, for example, {user=<default>}.

...

  1. The config-centric mode describes what is exactly specified for the configuration for the given entity match. However, it would also be useful to also be able to determine which matches have configuration values defined, so the presence of a filter is used for gathering information about the entity matches that the administrator is interested in. This is DescribeQuotas DescribeClientQuotas.
  2. The entity-centric mode describes what quotas apply to an entity. Note that an entity may match to various configuration entries depending on how the quotas are specified, e.g. the producer byte rate may be specified for the user, but the consumer byte rate for the client ID. Since it may not be clear how quotas were matched for an entity from the configuration, additional information should be returned to provide more context. This is DescribeEffectiveQuotas ResolveClientQuotas.

Altering quotas only works on a config-centric manner, and therefore doesn't need distinguishing. For a given entity match, the administrator should be able to specify which quotas apply, or alternatively remove existing quotas so they no longer match. This is AlterQuotas AlterClientQuotas.

Units

This KIP introduces the concept of a quota unit to be applied to a quota value. Currently, only a single unit is used for quotas: bytes-per-second, however this has limitations to effective quota management. For example, since it's a global throughput value, it doesn't scale well as brokers are added or removed, and so a broker-bytes-per-second unit could be added to better manage this behavior. As units are added, the possible quota configuration entries becomes the cross product of the quota types by the quota units, which means that it'd be possible to specify bytes-per-second both on a global and per-broker basis, and quota enforcement would occur at whichever limit was hit first. Additional considerations could be made for a fair-share system, where units of shares could be configured for quotas, and when bandwidth is contested, the share count of the active entities could be used to determine their restricted throughput.

It's beyond the scope of this KIP to add new units and implement their corresponding functionality in the broker, however it must be noted for future extensibility of the APIs.

Types Rationale

While there's two defined entity types in AK, a server-side plugin mechanism allows for further expansion. Likewise, as use cases evolve, finer-grained quota control may be necessary. Therefore, entity types should not be statically bound to publicly defined constants, and instead the API should support flexible entity types by interpreting them as a String identifier. Any entity types that the broker doesn't understand should throw an IllegalArgumentException back to the client.

The quota types (producer byte rate, consumer byte rate, etc.) and units should also be given the same consideration. The possible quota applications may expand in the future, and the API shouldn't lock in which quota types are accessible. Modification of quota types/units that are unknown should also fail with error.

Since a fixed set of entity types aren't defined, an entity should be represented by a Map<String, String>, which maps an entity type to the entity name.

Public Interfaces

Admin client calls will be added to correspond to describeQuotas, describeEffectiveQuotas, and alterQuotas, with supporting types defined in the common.quotas package.

Common types in package org.apache.kafka.common.quota:

The quota values are of type Double, which presents a complication in that the RPC protocol doesn't support floating point values. To accommodate this, RPC protocol message type 'double' will be added, which will serialize doubles according to the IEEE 754 floating-point "double format" bit layout.

Types Rationale

While there's two defined entity types in AK, a server-side plugin mechanism allows for further expansion. Likewise, as use cases evolve, finer-grained quota control may be necessary. Therefore, entity types should not be statically bound to publicly defined constants, and instead the API should support flexible entity types by interpreting them as a String identifier. Any entity types that the broker doesn't understand should throw an IllegalArgumentException back to the client.

The quota types (producer byte rate, consumer byte rate, etc.) should also be given the same consideration. The possible quota applications may expand in the future, and the API shouldn't lock in which quota types are accessible. Modification of quota types that are unknown should also fail with error.

Since a fixed set of entity types aren't defined, an entity should be represented by a Map<String, String>, which maps an entity type to the entity name.

Public Interfaces

Admin client calls will be added to correspond to DescribeClientQuotas, ResolveClientQuotas, and AlterClientQuotas, with supporting types defined in the common.quotas package.

Common types in package org.apache.kafka.common.quota (2.6.0)

Code Block
languagejava
/**
 * Describes a client quota entity, which is a mapping of entity types to their names.
 */
public class ClientQuotaEntity {

    /**
     * The type of an entity entry.
     */
    public static final String USER = "user";
    public static final String CLIENT_ID = "client-id";

    /**
     * Constructs a quota entity for the given types and names. If a name is null,
     * then it is mapped to the built-in default entity name.
     *
     * @param entries maps entity type to its name
     */
    public ClientQuotaEntity(Map<String, String> entries);
Code Block
languagejava
/**
 * Describes a fully-qualified entity.
 */
public class QuotaEntity {
    /**
     * Type@return map of an entity entry.entity type to its name
     */
    public enumMap<String, Type {String> entries();
}

/**
 * Describes a component for applying a USER,
client quota       CLIENT_ID;
    }filter.
 */
public class ClientQuotaFilterComponent {

    /**
     * RepresentsConstructs and thereturns defaulta namefilter forcomponent anthat entity, i.e.exactly matches the provided entity that's matched
     * whenname anfor exactthe match isn't foundentity type.
     */
    public final* static@param String QUOTA_ENTITY_NAME_DEFAULT = // implementation defined

    /**entityType the entity type the filter component applies to
     * `entries`@param describesentityName the fully-qualified entity. Thename keythat's is a {@code Type} string, howevermatched exactly
     */
 there may also existpublic keysstatic thatClientQuotaFilterComponent are not enumerated by {@code Type} that still apply, e.g.ofEntity(String entityType, String entityName);

    /**
     * theConstructs serverand mayreturns internallya associatefilter anothercomponent type.that When querying entities, it's necessarymatches the built-in default entity name
     * to return all quota types because quota values for thesethe types may influence the effectiveentity type.
     *
 quota value. However, when altering* a@param quota,entityType anythe typesentity thattype aren'tthe specifiedfilter mustcomponent beapplies ableto
     */
 to be inferred bypublic thestatic server, otherwise an error is returned.ClientQuotaFilterComponent ofDefaultEntity(String entityType);

     /**
     * ForConstructs example, {("CLIENT_ID" -> "test-client"),
     *   and returns a filter component that matches any specified name for the
     * entity type.
     ("USER" -> "test-user"),*
     * @param entityType the entity type the filter component       ("GROUP" -> "internal-group")}.applies to
     */
    public static ClientQuotaFilterComponent QuotaEntityofEntityType(Map<String, String> entriesString entityType);
}

    /**
 * Describes a quota key.
  * @return the component's entity type
     */
    public class QuotaKey {String entityType();

    /**
     * The@return quotathe types.
optional match    */string, where:
    public enum* Type {
       if CONSUMER_BYTE_RATEpresent,
 the name that's matched exactly
   PRODUCER_BYTE_RATE,
  *      REQUEST_PERCENTAGE;
   if }

empty, matches the default /**name
     * The units for a quota value. Note there may be multiple units for a given quota type if null, matches any specified name
     */
    public Optional<String> match();
}

/**
 * Describes thata influencesclient quota entity behaviorfilter.
     */
public class ClientQuotaFilter {

 public enum Units {/**
     * A  RATE_BPS;
    }

filter to be applied to matching client quotas.
     /**
     * @param typecomponents the components to quotafilter typeon
     * @param strict unitswhether the unitsfilter foronly theincludes quotaspecified typecomponents
     */
    publicprivate QuotaKeyClientQuotaFilter(TypeCollection<ClientQuotaFilterComponent> typecomponents, Unitsboolean unitsstrict);
}

    /**
     * Constructs and Describesreturns a quota filter that entitymatches filter.
 */
public class QuotaFilter {all provided components. Matching entities
    public enum* Rulewith {
entity types that are not specified by a EXACT,component will also be //included exactin namethe matchresult.
     *
   PREFIX;  * //@param matchescomponents allthe namescomponents withfor the givenfilter
 prefix
    }*/

    public static ClientQuotaFilter contains(Collection<ClientQuotaFilterComponent> components);

    /**
     * A filtering rule to be applied.
     *Constructs and returns a quota filter that matches all provided components. Matching entities
     * @paramwith entityType the entity typetypes thethat ruleare appliesnot to
specified by a component will *not* @parambe included rulein the rule to applyresult.
     *
     * @param matchcomponents the non-nullcomponents string that's applied by for the rulefilter
     */
    public QuotaFilter(QuotaEntity.Type entityType, Rule rule, String matchstatic ClientQuotaFilter containsOnly(Collection<ClientQuotaFilterComponent> components);
}

DescribeQuotas:

Code Block
languagejava
public class DescribeQuotasOptions extends AbstractOptions<DescribeQuotasOptions> { /**
    // Empty.
}

/**
 * The result of the {@link Admin#describeQuotas(Collection<QuotaFilter>, DescribeQuotasOptions)} call.
 */
public class DescribeQuotasResult {* Constructs and returns a quota filter that matches all configured entities.
     */
    public static ClientQuotaFilter all();

    /**
     * Maps an entity to its configured quota value(s). Note if no value is defined for a quota@return the filter's components
     */
    public Collection<ClientQuotaFilterComponent> components();

    /**
     * type@return forwhether thatthe entity's config, then it filter is not included in the resulting value map.strict, i.e. only includes specified components
     */
    public boolean strict();
}

/**
 @param* entitiesDescribes thea collectionconfiguration ofalteration entitiesto thatbe matchedmade theto filter
a client quota  entity.
 */
public class ClientQuotaAlteration {

    public DescribeQuotasResult(KafkaFuture<Map<QuotaEntity, Map<QuotaKey, Long>>> entities);static class Op {

        /**
     * Returns a map from* quota@param entitykey tothe aquota futuretype whichto canalter
 be used to check the status of the operation.
* @param value if set */
then the existing value publicis KafkaFuture<Map<QuotaEntityupdated,
 Map<QuotaKey, Long>>> entities();
}

public interface Admin extends AutoCloseable {
 *   ...

    /**
     * Describes allotherwise entities matching all provided filters (logical AND) that have at least one
     * quota value defined.
if null, the existing value is cleared
         */
        public Op(String key, Double value);

        /**
       * @param filters* filtering@return rulesthe toquota applytype to matching entitiesalter
     * @param options the options*/
 to use
     * @returnpublic result containing all matching entities
String key();

        /*/*
    DescribeQuotasResult  describeQuotas(Collection<QuotaFilter> filters, DescribeQuotasOptions options);
}

DescribeEffectiveQuotas:

Code Block
languagejava
public class DescribeEffectiveQuotasOptions extends AbstractOptions<DescribeEffectiveQuotasOptions> {

 * @return if set then the existing value is updated,
    /**
     *  Whether to exclude the list of overridden valuesotherwise forif everynull, quotathe typeexisting invalue theis result.cleared
         */
        public DescribeEffectiveQuotasOptionsDouble setOmitOverriddenValuesvalue(boolean omitOverriddenValues);
    }

/**
 * The result ofprivate thefinal {@link Admin#describeEffectiveQuotas(Collection<QuotaEntity>, DescribeEffectiveQuotasOptions)} call.
 */
public class DescribeEffectiveQuotasResult {ClientQuotaEntity entity;
    private final Collection<Op> ops;

    /**
     * Information about a specific quota configuration entry.@param entity the entity whose config will be modified
     */
 @param ops the publicalteration classto Entry {perform
        */**
    public     * @param source the entity source for the valueClientQuotaAlteration(ClientQuotaEntity entity, Collection<Op> ops);

    /**
     * @param@return value the non-null value
    entity whose config will be modified
     */
    public ClientQuotaEntity   public Entry(QuotaEntity source, Long valueentity();
    }

    /**
     * Information@return about the valuealteration for a quota type.to perform
     */
    public Collection<Op> ops();
}

DescribeClientQuotas (2.6.0)

Code Block
languagejava
public class DescribeClientQuotasOptions extends ValueAbstractOptions<DescribeClientQuotasOptions> {
        // Empty.
}

/**
 * The result of the {@link Admin#describeClientQuotas(Collection, DescribeClientQuotasOptions)} call.
 */
public @paramclass entryDescribeClientQuotasResult the{

 quota entry  /**
     * Maps an entity to *its @paramconfigured overriddenEntries all values that are overridden due to being lower inquota value(s). Note if no value is defined for a quota
     * type for that *entity's config, then it is not included in the resulting value map.
     *
     * @param entities future for specificity,the orcollection nullof ifentities notthat requested
matched the   filter
     */
        public ValueDescribeClientQuotasResult(EntryKafkaFuture<Map<ClientQuotaEntity, entryMap<String, List<Entry>Double>>> overriddenEntriesentities);

    }

    //**
     * MapsReturns a collectionmap of entities to their effective quota values.
     *
     * @param config the quota configuration for the requested entitiesfrom quota entity to a future which can be used to check the status of the operation.
     */
    public DescribeEffectiveQuotasResult(Map<QuotaEntityKafkaFuture<Map<ClientQuotaEntity, KafkaFuture<Map<QuotaKeyMap<String, Value>>>Double>>> config);entities();
}

public interface Admin extends AutoCloseable {
    ...

    /**
     * ReturnsDescribes aall mapentities frommatching quotathe entityprovided tofilter athat futurehave whichat can be used to check the status of the operationleast one client quota configuration
     * value defined.
     */ <p>
    public Map<QuotaEntity,* KafkaFuture<Map<QuotaKey,The Value>>> config();

    /**
     * Returns a future which succeeds only if all quota descriptions succeed.following exceptions can be anticipated when calling {@code get()} on the future from the
     * returned {@link DescribeClientQuotasResult}:
     */ <ul>
    public KafkaFuture<Void> all();
}

public interface Admin extends AutoCloseable {
    ...

*   <li>{@link org.apache.kafka.common.errors.ClusterAuthorizationException}
     /**
   If the authenticated *user Describesdidn't thehave effectivedescribe quotasaccess forto the provided entities.cluster.</li>
     *
   <li>{@link org.apache.kafka.common.errors.InvalidRequestException}
  * @param entities the* entities to describeIf the effectiverequest quotasdetails for
are invalid. e.g., an invalid *entity @paramtype options the options to usewas specified.</li>
     * @return the effective quotas for the entities
     */<li>{@link org.apache.kafka.common.errors.TimeoutException}
     *   If the request timed out before the describe could finish.</li>
    DescribeEffectiveQuotasResult describeEffectiveQuotas(Collection<QuotaEntity> entities, DescribeEffectiveQuotasOptions options);
}

AlterQuotas

Code Block
languagejava
titleAlterQuotas
public class AlterQuotasEntry { * </ul>
     * <p>
    public * classThis Opoperation {
is supported by brokers with version 2.6.0 or /**higher.
     *
     * @param keyfilter the quotafilter typeto andapply unitsto tomatch alterentities
         * @param value if set then options the existingoptions value is updated,
    to use
     * @return the            otherwise if null,DescribeClientQuotasResult containing the existing value is cleared
 result
        */
    DescribeClientQuotasResult describeClientQuotas(ClientQuotaFilter filter, DescribeClientQuotasOptions options);
}

ResolveClientQuotas (pending future release)

Code Block
languagejava
public class ResolveClientQuotasOptions extends AbstractOptions<ResolveClientQuotasOptions> {
    public Op(QuotaKey key, Long value);
    }

    // Empty.
}

/**
 * The result of *the @param entity the entity whose config will be modified{@link Admin#resolveClientQuotas(Collection, ResolveClientQuotasOptions)} call.
 */
public class ResolveClientQuotasResult {
     /**
 @param ops the alteration to* performInformation -about ifa valuespecific isquota set, then the existing value is updated,configuration entry.
     */
    public class Entry {
     otherwise if null, the existing value is cleared
 /**
         */
 @param source the publicentity AlterQuotasEntry(QuotaEntity entity, Collection<Op> ops);
}

public class AlterQuotasOptions extends AbstractOptions<AlterQuotasOptions> {
source for the value
         /**
 @param value the non-null *value
 Sets whether the request should be validated without altering the configs. */
     */
    public AlterQuotasOptions validateOnly(boolean validateOnlyEntry(QuotaEntity source, Double value);
    }

    /**
     * TheInformation resultabout ofthe thevalue {@link Admin#alterQuotas(Collection<AlterQuotasEntry>, AlterQuotasOptions)} call.
 for a quota type.
     *
 * The API of this    * NOTE: We maintain a `Value` class isbecause evolving,additional seeinformation {@linkmay Admin}be for details.
 */
public class AlterQuotasResult {
added, e.g.,
     *    public AlterQuotasResult(Map<QuotaEntity, KafkaFuture<Void>> futures);

a list of overridden /**entries.
     */
 Returns a map frompublic quotaclass entityValue to{
 a future which can be used to check/**
 the status of the operation.
     */
 @param entry the public Map<QuotaEntity, KafkaFuture<Void>> values();

quota entry
    /**
     * Returns a future which succeeds only if all quota alterations succeed./
     */
    public KafkaFuture<Void> all();
}

public interface Admin extends AutoCloseable {Value(Entry entry);
    ...}

    /**
     * Maps Altersa collection theof quotasentities asto specifiedtheir forresolved thequota entriesvalues.
     *
     * @param alterationsconfig the alterations to performquota configuration for the requested entities
     */
 @return the result  public ResolveClientQuotasResult(Map<QuotaEntity, KafkaFuture<Map<String, Value>>> config);

    /**
     * Returns a map from quota entity to a future which can be used to check the status of the alterationsoperation.
     */
    AlterQuotasResult alterQuotas(Collection<AlterQuotasEntry> entriespublic Map<QuotaEntity, KafkaFuture<Map<String, AlterQuotasOptionsValue>>> optionsconfig();
}

kafka-quotas.sh/QuotasCommand:

A QuotasCommand would be constructed with an associated bin/kafka-quotas.sh script for managing quotas via command line, and would have three modes of operation, roughly correlating to each of the API calls:

  1. List: Lists the quota entities for the given entity specification and their corresponding quota values, as explicitly specified in the configuration. The user may provide explicit entity types+names, or a pattern to apply to an entity type find matching entity names. If an entity type is omitted from the input, it is treated as a wildcard.
  2. Describe: Describes the effective quotas for an entity, including contextual information about how those quotas were derived. This includes what configuration entries matched to the entity and, if requested, the overridden, less-specific matches for the entity.
  3. Alter: Modifies a quota configuration entry in an incremental manner, i.e. specify which entries to add, update, and/or remove.
Flags

Various flags will be used to accomplish these operations.

Common flags:
--bootstrap-server: The standard bootstrap server.
--command-config: Property file for the Admin client.

Operations (mutually exclusive):
--list: Lists the entities that match the given specification, and prints out their configuration values.
--describe: Describes the effective quota values for an entity.
--alter: Alters the configuration for the given specification.

Entity specification flags (common to all):
--names: Comma-separated list of type=name pairs, e.g. "user=some-user,client-id=some-client-id"
--defaults: Comma-separated list of entity types with the default name, e.g. "defaults=user,client-id" (Note a separate flag is necessary since names are opaque.)

Exclusive to --list:
--prefix: Comma-separated prefix=name pairs, e.g. "user=test-".

Exclusive to --describe:
--include-overrides: Whether to include overridden config entries.

Exclusive to --alter:
--add: Comma-separated list of entries to add or update to the configuration, in format "name:unit=value".
--delete: Comma-separated list of entries to remove from the configuration, in format "name:unit".
--validate-only: If set, validates the alteration but doesn't perform it.

Output

In general, the output of the entities will be of the form: {entity-type=entity-name, ...}, where entity-name is sanitized for output since it is an opaque string. When displaying configuration values, the form: quota-name:quota-unit=quota-value.

List:

Code Block
$./bin/kafka-quotas.sh --bootstrap-server localhost:9092 --list \
                       --names=client-id=my-client

{user=user-two, client-id=my-client}
consumer_byte_rate:shares=200
producer_byte_rate:bps=10000000

{user=user-one, client-id=my-client}
producer_byte_rate:broker_bps=2000000

{user=<default>, client-id=my-client}
consumer_byte_rate:shares=100
producer_byte_rate:broker_bps=500000

$./bin/kafka-quotas.sh --bootstrap-server localhost:9092 --list \
                       --prefix=user=user-

{user=user-two, client-id=my-client}
consumer_byte_rate:shares=200
producer_byte_rate:bps=10000000

{user=user-one, client-id=my-client}
producer_byte_rate:broker_bps=2000000

Describe:

Code Block
$./bin/kafka-quotas.sh --bootstrap-server localhost:9092 --describe \
                       --names=user=user-one,client-id=my-client

consumer_byte_rate:shares=200 {user=user-one, client-id=my-client}
producer_byte_rate:bps=10000000 {user=user-one, client-id=my-client}
producer_byte_rate:broker_bps=500000 {user=<default>, client-id=my-client}

$./bin/kafka-quotas.sh --bootstrap-server localhost:9092 --describe \
                       --names=user=user-two,client-id=my-client    \
                       --include-overrides

consumer_byte_rate:shares=100 {user=<default>, client-id=my-client}
producer_byte_rate:broker_bps=2000000 {user=user-two, client-id=my-client}
*producer_byte_rate:broker_bps=500000 {user=<default>, client-id=my-client}

Alter:

Code Block
$./bin/kafka-quotas.sh --bootstrap-server localhost:9092 --describe \
                       --names=client-id=my-client --defaults=user  \
                       --add=producer_byte_rate:shares=100          \
                       --delete=producer_byte_rate:broker_bps

<no output on success>

$./bin/kafka-quotas.sh --bootstrap-server localhost:9092 --list     \
                       --names=client-id=my-client --defaults=user

{user=<default>, client-id=my-client}
consumer_byte_rate:shares=100
producer_byte_rate:shares=100

...



    /**
     * Returns a future which succeeds only if all quota descriptions succeed.
     */
    public KafkaFuture<Void> all();
}

public interface Admin extends AutoCloseable {
    ...

    /**
     * Resolves the effective quota values for the provided entities.
     *
     * @param entities the entities to describe the resolved quotas for
     * @param options the options to use
     * @return the resolved quotas for the entities
     */
    ResolveClientQuotasResult resolveClientQuotas(Collection<QuotaEntity> entities, ResolveClientQuotasOptions options);
}

AlterClientQuotas (2.6.0)

Code Block
languagejava
titleAlterQuotas
/**
 * Options for {@link Admin#alterClientQuotas(Collection, AlterClientQuotasOptions)}.
 *
 * The API of this class is evolving, see {@link Admin} for details.
 */
public class AlterClientQuotasOptions extends AbstractOptions<AlterClientQuotasOptions> {

    /**
     * Returns whether the request should be validated without altering the configs.
     */
    public boolean validateOnly();

    /**
     * Sets whether the request should be validated without altering the configs.
     */
    public AlterClientQuotasOptions validateOnly(boolean validateOnly);
}

/**
 * The result of the {@link Admin#alterClientQuotas(Collection, AlterClientQuotasOptions)} call.
 *
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class AlterClientQuotasResult {

    /**
     * Maps an entity to its alteration result.
     *
     * @param futures maps entity to its alteration result
     */
    public AlterClientQuotasResult(Map<ClientQuotaEntity, KafkaFuture<Void>> futures);

    /**
     * Returns a map from quota entity to a future which can be used to check the status of the operation.
     */
    public Map<ClientQuotaEntity, KafkaFuture<Void>> values();

    /**
     * Returns a future which succeeds only if all quota alterations succeed.
     */
    public KafkaFuture<Void> all();
}

public interface Admin extends AutoCloseable {
    ...

    /**
     * Alters client quota configurations with the specified alterations.
     * <p>
     * Alterations for a single entity are atomic, but across entities is not guaranteed. The resulting
     * per-entity error code should be evaluated to resolve the success or failure of all updates.
     * <p>
     * The following exceptions can be anticipated when calling {@code get()} on the futures obtained from
     * the returned {@link AlterClientQuotasResult}:
     * <ul>
     *   <li>{@link org.apache.kafka.common.errors.ClusterAuthorizationException}
     *   If the authenticated user didn't have alter access to the cluster.</li>
     *   <li>{@link org.apache.kafka.common.errors.InvalidRequestException}
     *   If the request details are invalid. e.g., a configuration key was specified more than once for an entity.</li>
     *   <li>{@link org.apache.kafka.common.errors.TimeoutException}
     *   If the request timed out before the alterations could finish. It cannot be guaranteed whether the update
     *   succeed or not.</li>
     * </ul>
     * <p>
     * This operation is supported by brokers with version 2.6.0 or higher.
     *
     * @param entries the alterations to perform
     * @return the AlterClientQuotasResult containing the result
     */
    AlterClientQuotasResult alterClientQuotas(Collection<ClientQuotaAlteration> entries, AlterClientQuotasOptions options);
}

kafka-configs.sh/ConfigCommand (2.6.0)

As a result of introducing the APIs, the ConfigCommand will be updated to support the users and clients entity types when using the --bootstrap-server option.  The modification to ConfigCommand was adopted in KIP-543, and usage will remain unchanged from the original --zookeeper functionality.

kafka-client-quotas.sh/ClientQuotasCommand (pending future release)

A ClientQuotasCommand would be constructed with an associated bin/kafka-client-quotas.sh script for managing quotas via command line, and would have three modes of operation, roughly correlating to each of the API calls:

  1. Describe: Describes the quota entities for the given entity specification and their corresponding quota values, as explicitly specified in the configuration. The user may provide explicit entity types+names, or a pattern to apply to an entity type find matching entity names. If an entity type is omitted from the input, it is treated as a wildcard.
  2. Resolve: Resolves the effective quotas for an entity, including contextual information about how those quotas were derived. This includes what configuration entries matched to the entity.
  3. Alter: Modifies a quota configuration entry in an incremental manner, i.e. specify which entries to add, update, and/or remove.
Flags

Various flags will be used to accomplish these operations.

Common flags:
--bootstrap-server: The standard bootstrap server.
--command-config: Property file for the Admin client.

Operations (mutually exclusive):
--describe: Describes the entities that match the given specification, and prints out their configuration values.
--resolve: Resolves the effective quota values for an entity.
--alter: Alters the configuration for the given specification.

Entity specification flags (common to all):
--names: Comma-separated list of type=name pairs, e.g. "user=some-user,client-id=some-client-id"
--defaults: Comma-separated list of entity types with the default name, e.g. "defaults=user,client-id" (Note a separate flag is necessary since names are opaque.)

Exclusive to --describe: None.

Exclusive to --resolve: None.

Exclusive to --alter:
--add: Comma-separated list of entries to add or update to the configuration, in format "name=value".
--delete: Comma-separated list of entries to remove from the configuration, in format "name".
--validate-only: If set, validates the alteration but doesn't perform it.

Input

When specifying configuration entries, the form: quota-name[=quota-value] is used.

Output

In general, the output of the entities will be of the form: {entity-type=entity-name, ...}, where entity-name is sanitized for output since it is an opaque string. When displaying configuration values, the form: quota-name=quota-value.

Describe:

Code Block
$./bin/kafka-client-quotas.sh --bootstrap-server localhost:9092 --describe \
                              --names=client-id=my-client

{user=user-one, client-id=my-client}
consumer_byte_rate=4000000
producer_byte_rate=1000000

{user=user-two, client-id=my-client}
producer_byte_rate=2000000

{user=<default>, client-id=my-client}
consumer_byte_rate=1000000
producer_byte_rate=500000

Resolve:

Code Block
$./bin/kafka-client-quotas.sh --bootstrap-server localhost:9092 --resolve \
                              --names=user=user-two,client-id=my-client

consumer_byte_rate=2000000 {user=user-two, client-id=my-client}
producer_byte_rate=500000 {user=<default>, client-id=my-client}

Alter:

Code Block
$./bin/kafka-client-quotas.sh --bootstrap-server localhost:9092 --alter   \
                              --names=client-id=my-client --defaults=user \

In addition to the API changes above, the following write protocol would be implemented:

DescribeQuotas:

Code Block
{
  "apiKey": 48,
  "type": "request",
  "name": "DescribeQuotasRequest",
  "validVersions": "0",
  "flexibleVersions": "none",
  "fields": [
    { "name": "Filter", "type": "[]QuotaFilterData", "versions": "0+",
      "about": "Filters to apply to quota entities.", "fields": [
      { "name": "EntityType", "type": "string", "versions": "0+",
        "about": "The entity type that the filter applies to." },
      { "name": "Rule", "type": "string", "versions": "0+",
        "about": "The rule the filter performs." },
      { "name": "Match", "type": "string", "versions": "0+",
        "about": "The string to apply the rule against." }
    ]}
  ]
}

{
  "apiKey": 48,
  "type": "response",
  "name": "DescribeQuotasResponse",
  "validVersions": "0",
  "flexibleVersions": "none",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "Entry", "type": "[]EntryData", "versions": "0+",
      "about": "A result entry.", "fields": [
      { "name": "ErrorCode", "type": "int16", "versions": "0+",
        "about": "The error code, or `0` if the quota description succeeded." },
      { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+",
        "about": "The error message, or `null` if the quota description succeeded." },
      { "name": "Entity", "type": "[]QuotaEntityData", "versions": "0+",
        "about": "The quota entity description.", "fields": [
        { "name": "EntityType", "type": "string", "versions": "0+",
          "about": "The entity type." },
        { "name": "EntityName", "type": "string", "versions": "0+",
  --add=consumer_byte_rate=2000000        "about": "The entity name." }\
      ]},
      { "name": "Type", "type": "string", "versions": "0+",
        "about": "The quota type." },
      { "name": "Units", "type": "string", "versions": "0+",
 --delete=producer_byte_rate

<no output on success>

$./bin/kafka-client-quotas.sh --bootstrap-server localhost:9092 --describe \
              "about": "The units for the value." },
      { "name": "Value",     --names=client-id=my-client --defaults=user

{user=<default>, client-id=my-client}
consumer_byte_rate=2000000

Proposed Changes

In addition to the API changes above, the following write protocol would be implemented:

DescribeClientQuotas (2.6.0)

Code Block
{
  "apiKey": 48,
  "type": "int64request",
  "versionsname": "0+DescribeClientQuotasRequest",
        "aboutvalidVersions": "The quota value." }
    ]}
  ]
}

DescribeEffectiveQuotas:

Code Block
{
  "apiKey": 49,
  "type": "request",
  "name": "DescribeEffectiveQuotasRequest""0",
  "validVersions": "0",
  "flexibleVersions": "none",
  "fields": [
    { "name": "EntityComponents", "type": "[]QuotaEntityDataComponentData", "versions": "0+",
      "about": "TheFilter components to apply to quota entity descriptionentities.", "fields": [
      { "name": "EntityType", "type": "string", "versions": "0+",
        "about": "The entity type that the filter component applies to." },
      { "name": "EntityNameMatchType", "type": "stringint8", "versions": "0+",
        "about": "TheHow to match the entity {0 name." }
    ]= exact name, 1 = default name, 2 = any specified name}." },
      { "name": "OmitOverriddenValuesMatch", "type": "boolstring", "versions": "0+", "nullableVersions": "0+",
        "about": "WhetherThe string to excludematch theagainst, listor ofnull overriddenif valuesunused for everythe quotamatch type." }
    ]
}

{,
  "apiKey": 49,
 { "typename": "responseStrict",
  "nametype": "DescribeEffectiveQuotasResponsebool",
  "validVersionsversions": "0+",
      "about": "Whether the match is strict, i.e. should exclude entities with unspecified entity types." }
  ]
}

{
  "apiKey": 48,
  "type": "response",
  "name": "DescribeClientQuotasResponse",
  "validVersions": "0",
  "flexibleVersions": "none",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "EntryErrorCode", "type": "[]QuotaEntryDataint16", "versions": "0+",
      "about": "EffectiveThe error code, or `0` if the quota entriesdescription succeeded.", "fields": [
   },
    { "name": "ErrorCodeErrorMessage", "type": "int16string", "versions": "0+",
 "nullableVersions": "0+",
      "about": "The error codemessage, or `0``null` if the effective quota description succeeded." },
      { "name": "ErrorMessageEntries", "type": "string[]EntryData", "versions": "0+", "nullableVersions": "0+",
        "about": "TheA errorresult message, or `null` if the effective quota description succeeded." },entry.", "fields": [
      { "name": "QuotaEntityEntity", "type": "[]QuotaEntityEntityData", "versions": "0+",
        "about": "EffectiveThe quota entity entriesdescription.", "fields": [
        { "name": "EntityType", "type": "string", "versions": "0+",
          "about": "The entity type." },
        { "name": "EntityName", "type": "string", "versions": "0+",
   "nullableVersions": "0+",
          "about": "The entity name, or null if the default." }
      ]},
      { "name": "QuotaValuesValues", "type": "[]QuotaValueDataValueData", "versions": "0+",
        	"about": "QuotaThe configurationquota values for the entity.", "fields": [
        { "name": "TypeKey", "type": "string", "versions": "0+",
          "about": "The quota configuration typekey." },
        { "name": "UnitsValue", "type": "stringfloat64", "versions": "0+",
          "about": "The unitsquota for the quota typeconfiguration value." },
      ]}
    ]}
  ]
}

ResolveClientQuotas (pending future release)

Code Block
{
  "nameapiKey": "Entry"50,
  "type": "[]ValueEntryDatarequest",
  "versionsname": "0+ResolveClientQuotasRequest",
          "about"validVersions": "0",
  "flexibleVersions": "Quota value entries.none",
  "fields": [
          { "name": "QuotaEntityEntity", "type": "[]ValueQuotaEntityQuotaEntityData", "versions": "0+",
            "about": "EffectiveThe quota entity entriesdescription.", "fields": [
            { "name": "EntityType", "type": "string", "versions": "0+",
              "about": "The entity type." },
            { "name": "EntityName", "type": "string", "versions": "0+",
              "about": "The entity name." }
    ]}
      ]
},
          { "name

{
  "apiKey": "Value"50,
  "type": "int64response",
  "versionsname": "0+"ResolveClientQuotasResponse",
            "aboutvalidVersions": "The quota configuration value." }0",
        ]}"flexibleVersions": "none",
      ]}
    ]}
  ]
}

AlterQuotas:

Code Block
{
  "apiKey": 50,
  "fields": [
    { "name": "ThrottleTimeMs", "type": "requestint32",
  "nameversions": "AlterQuotasRequest0+",
  "validVersions": "0",
    "flexibleVersionsabout": "none",
  "fields": [The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "Entry", "type": "[]EntryDataQuotaEntryData", "versions": "0+",
      "about": "TheResolved quota configuration entries to alter.", "fields": [
      { "name": "ErrorCode", "type": "int16", "versions": "0+",
      { "name": "QuotaEntity", "type": "[]QuotaEntity", "versions": "0+",
        "about": "The quota entity to alter.", "fields": [
        { "name": "EntityType", "type": "string", "versions": "0+",
          "about": "The entity type." },
        { "name": "EntityName", "type": "string", "versions": "0+",   "about": "The error code, or `0` if the resolved quota description succeeded." },
      { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+",
        "about": "The error message, or `null` if the resolved quota description succeeded." },
      { "name": "QuotaEntity", "type": "[]QuotaEntity", "versions": "0+",
        "about": "Resolved quota entries.", "fields": [
        { "name": "EntityType", "type": "string", "versions": "0+",
          "about": "The entity type." },
        { "name": "EntityName", "type": "string", "versions": "0+",
          "about": "The entity name." }
      ]},
      { "name": "QuotaValues", "type": "[]QuotaValueData", "versions": "0+",
        "about": "Quota configuration values.", "fields": [
        { "name": "Type", "type": "string", "versions": "0+",
          "about": "The quota type." },
        { "name": "Entry", "type": "[]ValueEntryData", "versions": "0+",
          "about": "Quota value entries.", "fields": [
          { "name": "QuotaEntity", "type": "[]ValueQuotaEntity", "versions": "0+",
            "about": "Resolved quota entries.", "fields": [
            { "name": "EntityType", "type": "string", "versions": "0+",
              "about": "The entity type." },
            { "name": "EntityName", "type": "string", "versions": "0+",
              "about": "The entity name." }
          ]},
          { "name": "Value", "type": "double", "versions": "0+",
            "about": "The quota configuration value." }
        ]}
      ]}
    ]}
  ]
}

AlterClientQuotas (2.6.0)

Code Block
{
  "apiKey": 49,
  "type": "request",
  "name": "AlterClientQuotasRequest",
  "validVersions": "0",
  "flexibleVersions": "none",
  "fields": [
    { "name": "Entries", "type": "[]EntryData", "versions": "0+",
      "about": "The quota configuration entries to alter.", "fields": [
      { "name": "Entity", "type": "[]EntityData", "versions": "0+",
        "about": "The quota entity to alter.", "fields": [
        { "name": "EntityType", "type": "string", "versions": "0+",
          "about": "The entity type." },
        { "name": "EntityName", "type": "string", "versions": "0+", "nullableVersions": "0+",
          "about": "The name of the entity, or null if the default." }
      ]},
      { "name": "Ops", "type": "[]OpData", "versions": "0+",
        "about": "An individual quota configuration entry to alter.", "fields": [
        { "name": "Key", "type": "string", "versions": "0+",
          "about": "The quota configuration key." },
        { "name": "Value", "type": "float64", "versions": "0+",
          "about": "The value to set, otherwise ignored if the value is to be removed." },
        { "name": "Remove", "type": "bool", "versions": "0+",
          "about": "Whether the quota configuration value should be removed, otherwise set." }
      ]}
    ]},
    { "name": "ValidateOnly", "type": "bool", "versions": "0+",
      "about": "Whether the alteration should be validated, but not performed." }
  ]
}

{
  "apiKey": 49,
  "type": "response",
  "name": "AlterClientQuotasResponse",
  "validVersions": "0",
  "flexibleVersions": "none",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "Entries", "type": "[]EntryData", "versions": "0+",
      "about": "The quota configuration entries to alter.", "fields": [
      { "name": "ErrorCode", "type": "int16", "versions": "0+",
        "about": "The error code, or `0` if the quota alteration succeeded." },
      { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+",
        "about": "The error message, or `null` if the quota alteration succeeded." },
      { "name": "Entity", "type": "[]EntityData", "versions": "0+",
        "about": "The quota entity to alter.", "fields": [
        { "name": "EntityType", "type": "string", "versions": "0+",
          "about": "The entity type." },
        { "name": "EntityName", "type": "string", "versions": "0+", "nullableVersions": "0+",
          "about": "The name of the entity, or null if the default." }
      ]}
    ]}
  ]
}

Kafka RPC 'double' support (2.6.0)

Note that, while the ByteBuffer natively supports serializing a Double, the format in which the value is serialized is not strongly specified, so the preference is to explicitly ensure a standard representation of double-precision 64-bit IEEE 754 format. This is achieved in Java using Double.doubleToRawLongBits() and Double.longBitsToDouble() and should be easily portable to other languages.

Code Block
titleclients/src/main/java/org/apache/kafka/common/utils/ByteUtils.java
    /**
     * Read a double-precision 64-bit format IEEE 754 value.
     *
     * @param buffer The buffer to read from
     * @return The long value read
     */
    public static double readDouble(ByteBuffer buffer) {
        return  "about": "The name of the entity." }Double.longBitsToDouble(buffer.getLong());
    }

    /**
      ]},
      { "name": "Op", "type": "[]OpData", "versions": "0+",
        "about": "An individual quota configuration entry to alter.", "fields": [* Write the given double following the double-precision 64-bit format IEEE 754 value into the buffer.
     *
     * @param value The value to write
     * @param buffer { "name": "Type", "type": "string", "versions": "0+",
The buffer to write to
     */
    public static  "about": "The quota type." },void writeDouble(double value, ByteBuffer buffer) {
        buffer.putLong(Double.doubleToRawLongBits(value));
    }

The protocol type definition:

Code Block
titleclients/src/main/java/org/apache/kafka/common/protocol/types/Type.java
    public static final DocumentedType DOUBLE = new DocumentedType() {
        @Override{ "name": "Units", "type": "string", "versions": "0+",
          "about": "The units for the quota type." },
        public { "name": "Value", "type": "int64", "versions": "0+",void write(ByteBuffer buffer, Object o) {
          "about": "The value to set, otherwise ignored if the value is to be removed." }, ByteUtils.writeDouble((Double) o, buffer);
        }

        { "name": "Remove", "type": "bool", "versions": "0+",
@Override
        public Object read(ByteBuffer buffer) {
   "about": "Whether the quota configuration value should be removed, otherwisereturn set." }
ByteUtils.readDouble(buffer);
        ]}

        ]},@Override
    { "name": "ValidateOnly", "type": "bool", "versions": "0+",
   public int sizeOf(Object o) {
    "about": "Whether the alteration should be validated, but not performed." }return 8;
  ]
}

{
  "apiKey": 50,
  "type": "response", }

  "name": "AlterQuotasResponse",
  "validVersions": "0",
  "flexibleVersions": "none",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      @Override
        public String typeName() {
            return "aboutDOUBLE":;
 "The duration in milliseconds for which the request}

 was throttled due to a quota violation, or zero@Override
 if the request did not violate any quota." },
    { "name": "Entry", "type": "[]EntryData", "versions": "0+",
public Double validate(Object item) {
             "about": "The quota configuration entries to alter.", "fields": [
if (item instanceof Double)
            { "name": "ErrorCode", "type": "int16", "versions": "0+", return (Double) item;
        "about": "The error code, orelse
 `0` if the quota alteration succeeded." },
      { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", throw new SchemaException(item + " is not a Double.");
        }

        "about": "The error message, or `null` if the quota alteration succeeded." },
@Override
        public String documentation() {
       { "name": "QuotaEntity", "type": "[]QuotaEntity", "versions": "0+",
        "about": "The quota entity to alter.", "fields": [
 return "Represents a double-precision 64-bit format IEEE 754 value. " +
               { "name": "EntityType", "type": "string", "versions": "0+",
          "about": "The entity type." },
        { "name": "EntityName", "type": "string", "versions": "0+",
          "about": "The name of the entity." }
      ]}
    ]}
  ]
}The values are encoded using eight bytes in network byte order (big-endian).";
        }
    };


In generator/src/main/java/org/apache/kafka/message/MessageGenerator.java, the following operations will be used (code omitted for brevity):

Code Block
titlegenerator/src/main/java/org/apache/kafka/message/MessageGenerator.java
Hash code: Double.hashCode(value)

Empty value: (double) 0

Parsing a default value string: Double.parseDouble(defaultValue)

Compatibility, Deprecation, and Migration Plan

All changes would be are forward-compatible, and no migration plan is necessary. It's outside the scope of this KIP to deprecate any functionality.

Rejected Alternatives

  • Use existing describeConfigs/incrementalAlterConfigs for quota functionality. This falls short for a couple reasons. First, quotas entity names are more dynamic than brokers and tasks which makes them awkward to fit into generic tools which expect a single unique, distinct key, e.g. ConfigCommand. Second, there's no tool that expresses a way to get the effective quota resolved quota for an entity without some heavy engineering on the client side, which lacks extensibility and is more expensive to perform, especially over large collection of entities. Therefore, it makes sense to approach quotas as a standalone set of APIs that provide more targeted information and can properly support future extensibility.

...