Versions Compared

Old Version 53

changes.mady.by.user Brian Byrne

Saved on

compared with

New Version Current

changes.mady.by.user Brian Byrne

Saved on

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: 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 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>}.

...

The quota values are of type Double, which presents a complication in that the RPC protocol doesn't support floating point values. To account for accommodate this, Double values will be serialized via Double.doubleToLongBits() and deserialized via Double.longBitsToDouble(). Java strongly defines the byte format, so there's no issue about compatibility. 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.

...

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 fully-qualifiedquota entity.
 */
public class QuotaEntity, which is a mapping of entity types to their names.
 */
public class ClientQuotaEntity {

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

    /**
     * RepresentsConstructs thea defaultquota nameentity for the given antypes entity,and inames.e. If thea entityname that's matchedis null,
     * when an exact match isn't found then it is mapped to the built-in default entity name.
     */
     * public@param staticentries finalmaps String QUOTA_ENTITY_NAME_DEFAULT = // implementation definedentity type to its name
     */
    public ClientQuotaEntity(Map<String, String> entries);

    /**
     * `entries`@return describesmap theof fully-qualified entity. The key is a {@code Type} string, howeverentity type to its name
     */
    public *Map<String, there may also exist keys that are not enumerated by {@code Type} that still apply, e.g.String> entries();
}

/**
 * Describes a component for applying a client quota filter.
 */
public class ClientQuotaFilterComponent {

     /**
 the server may internally associate* anotherConstructs type.and Whenreturns queryinga entities, it's necessary
     * to return all quota types because quota values for these types may influence the resolvedfilter component that exactly matches the provided entity
     * name for the entity type.
     *
     * @param entityType quotathe value.entity However,type whenthe alteringfilter acomponent quota,applies anyto
 types that aren't specified must be able   * @param entityName the entity name that's matched exactly
     */
 to   bepublic inferredstatic byClientQuotaFilterComponent theofEntity(String serverentityType, otherwise an error is returned. String entityName);

    /**
     *
 Constructs and returns a filter *component Forthat example, {("CLIENT_ID" -> "test-client"),matches the built-in default entity name
     * for the entity type.
     *
     * ("USER" -> "test-user"),
     *  @param entityType the entity type the filter component applies to
     */
    public static ClientQuotaFilterComponent  ofDefaultEntity("GROUP" -> "internal-group")}.String entityType);

     /**/
    public QuotaEntity(Map<String, String> entries);
}

/**
 * Describes a quota entity filter.
 */
public class QuotaFilter {
    /*** Constructs and returns a filter component that matches any specified name for the
     * A filter to be appliedentity type.
     *
     * @param entityType the entity type the filter component applies to
     */
 @param match the non-nullpublic stringstatic that's matched exactlyClientQuotaFilterComponent ofEntityType(String entityType);

     /**/
     * public@return QuotaFilter(String entityType, String match);
}

DescribeClientQuotas:

Code Block
languagejava
public class DescribeClientQuotasOptions extends AbstractOptions<DescribeClientQuotasOptions> {the component's entity type
     */
    public // Empty.
}

String entityType();

    /**
  *  The result* of@return the {@link Admin#DescribeClientQuotas(Collection<QuotaFilter>, DescribeClientQuotasOptions)} call.
 */
public class DescribeClientQuotasResult {

    /**optional match string, where:
     *         if present, the name that's matched exactly
     * Maps an entity to its configured quota value(s). Note if noempty, valuematches isthe defined for a quotadefault name
     * type for that entity's config, then it is not includedif innull, thematches resultingany valuespecified map.name
     */
    public Optional<String> match();
}

/**
 @param* entitiesDescribes thea collectionclient ofquota entities that matched the filter
     entity filter.
 */
    public DescribeClientQuotasResult(KafkaFuture<Map<QuotaEntity, Map<String, Double>>> entities);class ClientQuotaFilter {

    /**
     * A Returnsfilter ato mapbe fromapplied quotato entitymatching toclient aquotas.
 future which can be used*
 to check the status of* the@param operation.
components the components to filter */on
    public KafkaFuture<Map<QuotaEntity, Map<String, Double>>> entities();
}

public interface Admin extends AutoCloseable {
    ...* @param strict whether the filter only includes specified components
     */
    private ClientQuotaFilter(Collection<ClientQuotaFilterComponent> components, boolean strict);

    /**
     * DescribesConstructs alland entitiesreturns matchinga allquota providedfilter filtersthat (logicalmatches AND)all thatprovided havecomponents. atMatching least oneentities
     * quotawith valueentity defined.
types that are not specified *
by a component will also *be @paramincluded filters filters to apply to matching entitiesin the result.
     *
     * @param optionscomponents the optionscomponents for tothe usefilter
     */
  @return  resultpublic containingstatic allClientQuotaFilter matching entitiescontains(Collection<ClientQuotaFilterComponent> components);

     /**/
    DescribeClientQuotasResult describeClientQuotas(Collection<QuotaFilter> filters, DescribeClientQuotasOptions options);
}

ResolveClientQuotas:

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

/**
 * The result of the {@link Admin#ResolveClientQuotas(Collection<QuotaEntity>, ResolveClientQuotasOptions)} call.
 */
public class ResolveClientQuotasResult {
    /*** Constructs and returns a quota filter that matches all provided components. Matching entities
     * with entity types that are not specified by a component will *not* be included in the result.
     *
     * Information@param aboutcomponents athe specificcomponents quotafor configurationthe entry.filter
     */
    public static classClientQuotaFilter Entry {
  containsOnly(Collection<ClientQuotaFilterComponent> components);

      /**
     * Constructs and returns *a @paramquota sourcefilter thethat entitymatches sourceall for the value
configured entities.
     */
    public static * @param value the non-null valueClientQuotaFilter all();

    /**
     * @return the filter's */components
     */
    public Collection<ClientQuotaFilterComponent> Entry(QuotaEntity source, Double valuecomponents();
    }

    /**
     * Information@return aboutwhether the value for a quota type. filter is strict, i.e. only includes specified components
     */
    public boolean strict();
}

/**
 NOTE:* WeDescribes maintain a `Value`configuration class because additional information may be added, e.g.,
     *       a list of overridden entries.
     */alteration to be made to a client quota entity.
 */
public class ClientQuotaAlteration {

    public static class ValueOp {

        /**
         * @param entrykey the quota entrytype to alter
         */
 @param value if set then the existing publicvalue Value(Entry entry);is updated,
    }

     /**
     * Maps a collection of entities to their resolved quota values.
     *otherwise if null, the existing value is cleared
     * @param config the quota configuration for the requested entities */
     */
    public ResolveClientQuotasResultOp(Map<QuotaEntity,String KafkaFuture<Map<Stringkey, Value>>>Double configvalue);

    /**
     /**
 Returns a map from quota entity to a future* which@return canthe bequota usedtype to checkalter
 the status of the operation.
     */
     public  Map<QuotaEntity, KafkaFuture<Map<String,public Value>>>String configkey();

        /**
     * Returns a future which* succeeds@return only if allset quotathen descriptionsthe succeed.
existing value is updated,
  */
    public KafkaFuture<Void> all();
}

public interface Admin extends AutoCloseable {
 *        ...

 otherwise if  /**
     * Describes the resolved quotas for the provided entities.null, the existing value is cleared
         */
     *
   public Double value();
 * @param entities the}

 entities to describe theprivate resolvedfinal quotasClientQuotaEntity forentity;
    private *final @param options the options to useCollection<Op> ops;

    /**
     * @return@param entity the resolvedentity whose quotasconfig forwill thebe entitiesmodified
     */
 @param ops the ResolveClientQuotasResultalteration resolveClientQuotas(Collection<QuotaEntity> entities, ResolveClientQuotasOptions options);
}

AlterClientQuotas

Code Block
languagejava
titleAlterQuotas
public class AlterClientQuotasEntry {to perform
     */
    public class Op {
   ClientQuotaAlteration(ClientQuotaEntity entity, Collection<Op> ops);

      /**
         * @param key@return the quotaentity typewhose toconfig alter
will be modified
     */
  * @param valuepublic if set then the existing value is updated,ClientQuotaEntity entity();

    /**
         * @return the alteration to perform
     */
    otherwisepublic Collection<Op> ops();
}

DescribeClientQuotas (2.6.0)

Code Block
languagejava
public class DescribeClientQuotasOptions extends AbstractOptions<DescribeClientQuotasOptions> {if null, the existing value is cleared
         */
        public Op(String key, Double value);
    }// Empty.
}

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

    /**
     * @paramMaps an entity theto entityits whoseconfigured config will be modified
     * @param ops the alteration to perform - if value is set, then the existing value is updated,quota 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 the collection otherwiseof if null,entities that matched the existing value is clearedfilter
     */
    public AlterClientQuotasEntryDescribeClientQuotasResult(QuotaEntityKafkaFuture<Map<ClientQuotaEntity, entityMap<String, Collection<Op>Double>>> opsentities);
}

public class AlterClientQuotasOptions extends AbstractOptions<AlterClientQuotasOptions> {
/**
     /**
 Returns a map from *quota entity Setsto whethera thefuture requestwhich shouldcan be validated without altering used to check the status of the configsoperation.
     */
    public AlterClientQuotasOptions validateOnly(boolean validateOnly KafkaFuture<Map<ClientQuotaEntity, Map<String, Double>>> entities();
}

/**
 * The result of the {@link Admin#AlterClientQuotas(Collection<AlterClientQuotasEntry>, AlterClientQuotasOptions)} call.
 *
 * The API of this class is evolving, see {@link Admin} for details.
 */
public class AlterClientQuotasResult {
    public AlterClientQuotasResult(Map<QuotaEntity, KafkaFuture<Void>> futures);

    /**public interface Admin extends AutoCloseable {
    ...

    /**
     * Describes all entities matching the provided filter that have at least one client quota configuration
     * value defined.
     * Returns<p>
 a map from quota entity* toThe afollowing future whichexceptions can be usedanticipated towhen checkcalling the status of the operation.{@code get()} on the future from the
     */
 returned {@link DescribeClientQuotasResult}:
 public Map<QuotaEntity, KafkaFuture<Void>> values();

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

public interface Admin extends AutoCloseable {
    ...

    /**
     * Alters the quotas as specified for the entries.<li>{@link org.apache.kafka.common.errors.ClusterAuthorizationException}
     *   If the authenticated user didn't have describe access to the cluster.</li>
     *   <li>{@link org.apache.kafka.common.errors.InvalidRequestException}
     *   If the request details are invalid. e.g., an invalid entity type was specified.</li>
     *   <li>{@link org.apache.kafka.common.errors.TimeoutException}
     *  @param alterationsIf the alterationsrequest totimed perform
out before the describe  * @return the result of the alterationscould finish.</li>
     * </ul>
     */ <p>
    AlterClientQuotasResult alterClientQuotas(Collection<AlterClientQuotasEntry> entries, AlterClientQuotasOptions options);
}

kafka-client-quotas.sh/ClientQuotasCommand:

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:

 * This operation is supported by brokers with version 2.6.0 or higher.
     *
     * @param filter the filter to apply to match entities
     * @param options the options to use
     * @return the DescribeClientQuotasResult containing the result
     */
    DescribeClientQuotasResult describeClientQuotas(ClientQuotaFilter filter, DescribeClientQuotasOptions options);
}

ResolveClientQuotas (pending future release)

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

/**
 * The result of the {@link Admin#resolveClientQuotas(Collection, ResolveClientQuotasOptions)} call.
 */
public class ResolveClientQuotasResult {
    /**
     * Information about a specific quota configuration entry.
     */
    public class Entry {
        /**
         * @param source the entity source for the value
         * @param value the non-null value
         */
        public Entry(QuotaEntity source, Double value);
    }

    /**
     * Information about the value for a quota type.
     *
     * NOTE: We maintain a `Value` class because additional information may be added, e.g.,
     *       a list of overridden entries.
     */
    public class Value {
        /**
         * @param entry the quota entry
         */
        public Value(Entry entry);
    }

    /**
     * Maps a collection of entities to their resolved quota values.
     *
     * @param config the quota configuration for the requested entities
     */
    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 operation.
     */
    public Map<QuotaEntity, KafkaFuture<Map<String, Value>>> config();

    /**
     * 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 \
                              --add=consumer_byte_rate=2000000            \
                              --delete=producer_byte_rate

<no output on success>

$./bin/kafka-client-quotas.sh --bootstrap-server localhost:9092 --describe \
                              --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": "request",
  "name": "DescribeClientQuotasRequest",
  "validVersions": "0",
  "flexibleVersions": "none",
  "fields": [
    { "name": "Components", "type": "[]ComponentData", "versions": "0+",
      "about": "Filter components to apply to quota entities.", "fields": [
      { "name": "EntityType", "type": "string", "versions": "0+",
        "about": "The entity type that the filter component applies to." },
      { "name": "MatchType", "type": "int8", "versions": "0+",
        "about": "How to match the entity {0 = exact name, 1 = default name, 2 = any specified name}." },
      { "name": "Match", "type": "string", "versions": "0+", "nullableVersions": "0+",
        "about": "The string to match against, or null if unused for the match type." }
    ]},
    { "name": "Strict", "type": "bool", "versions": "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": "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": "Entries", "type": "[]EntryData", "versions": "0+", "nullableVersions": "0+",
      "about": "A result entry.", "fields": [
      { "name": "Entity", "type": "[]EntityData", "versions": "0+",
        "about": "The quota entity description.", "fields": [
        { "name": "EntityType", "type": "string", "versions": "0+",
Code Block
$./bin/kafka-client-quotas.sh --bootstrap-server localhost:9092 --describe \
                           "about": "The  --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 \
  entity type." },
        { "name": "EntityName", "type": "string", "versions": "0+", "nullableVersions": "0+",
          "about": "The entity name, or null if the default." }
      ]},
   --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   \
         "name": "Values", "type": "[]ValueData", "versions": "0+",
	"about": "The quota values for the entity.", "fields": [
        { "name": "Key", "type": "string", "versions": "0+",
        --names=client-id=my-client --defaults=user \
 "about": "The quota configuration key." },
        { "name": "Value", "type": "float64", "versions": "0+",
          --add=consumer_byte_rate=2000000     "about": "The quota configuration value." }
       \]}
    ]}
  ]
}

ResolveClientQuotas (pending future release)

Code Block
{
  "apiKey": 50,
  "type": "request",
  "name": "ResolveClientQuotasRequest",
    "validVersions": "0",
  "flexibleVersions": "none",
  "fields": [
     --delete=producer_byte_rate

<no output on success>

$./bin/kafka-client-quotas.sh --bootstrap-server localhost:9092 --describe \{ "name": "Entity", "type": "[]QuotaEntityData", "versions": "0+",
      "about": "The quota entity description.", "fields": [
      { "name": "EntityType", "type": "string", "versions": "0+",
      --names=client-id=my-client --defaults=user

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

...

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

DescribeClientQuotas:

Code Block
{
  "apiKey": 48,
  "type": "request",
  "name": "DescribeClientQuotasRequest",
  "validVersions "about": "The entity type." },
      { "name": "EntityName", "type": "string", "versions": "0+",
  "flexibleVersions      "about": "none",
The entity "fieldsname.": [}
    ]}
  ]
}

{
  "nameapiKey": "Filter"50,
  "type": "[]QuotaFilterDataresponse",
  "versionsname": "0+ResolveClientQuotasResponse",
  "validVersions": "0",
   "aboutflexibleVersions": "Filters to apply to quota entities.","none",
  "fields": [
      { "name": "EntityTypeThrottleTimeMs", "type": "stringint32", "versions": "0+",
        "about": "The duration entityin milliseconds typefor thatwhich the filter applies torequest was throttled due to a quota violation, or zero if the request did not violate any quota." },
      { "name": "MatchEntry", "type": "string[]QuotaEntryData", "versions": "0+",
        "about": "TheResolved string to match against." }quota entries.", "fields": [
    ]}
  ]
}

{
  "apiKeyname": 48,
  "type": "response"ErrorCode",
  "nametype": "DescribeClientQuotasResponseint16",
  "validVersionsversions": "0+",
  "flexibleVersions      "about": "none",
  "fields": [
The error code, or `0` if the resolved quota description succeeded." },
      { "name": "ThrottleTimeMsErrorMessage", "type": "int32string", "versions": "0+",
      "aboutnullableVersions": "0+"The,
 duration in milliseconds for which the request was throttled due to a quota violation"about": "The error message, or zero`null` if the requestresolved did not violate any quotaquota description succeeded." },
      { "name": "EntryQuotaEntity", "type": "[]EntryDataQuotaEntity", "versions": "0+",
        "about": "AResolved resultquota entryentries.", "fields": [
        { "name": "ErrorCodeEntityType", "type": "int16string", "versions": "0+",
          "about": "The error code, or `0` if the quota description succeeded." }entity type." },
        { "name": "EntityName", "type": "string", "versions": "0+",
      {    "nameabout": "ErrorMessage", "typeThe entity name." }
      ]},
      { "name": "stringQuotaValues", "versionstype": "0+[]QuotaValueData", "nullableVersionsversions": "0+",
        "about": "TheQuota errorconfiguration messagevalues.", or `null` if the quota description succeeded." },
"fields": [
        { "name": "EntityType", "type": "[]QuotaEntityDatastring", "versions": "0+",
          "about": "The quota entity descriptiontype.", "fields": [ },
        { "name": "EntityTypeEntry", "type": "string[]ValueEntryData", "versions": "0+",
          "about": "TheQuota entityvalue typeentries.", "fields": },[
          { "name": "EntityNameQuotaEntity", "type": "string[]ValueQuotaEntity", "versions": "0+",
            "about": "The entity name." }Resolved quota entries.", "fields": [
      ]},
      { "name": "TypeEntityType", "type": "string", "versions": "0+",
              "about": "The quotaentity type." },
            { "name": "ValueEntityName", "type": "int64string", "versions": "0+",
              "about": "The quotaentity valuename." }
    ]}
  ]
}

ResolveClientQuotas:

Code Block
{
  "apiKey": 49,
  "type": "request",
  "name": "ResolveClientQuotasRequest",
  "validVersions": "0" ]},
  "flexibleVersions": "none",
  "fields": [
    { "name": "EntityValue", "type": "[]QuotaEntityDatadouble", "versions": "0+",
            "about": "The quota entityconfiguration descriptionvalue.", "fields": [ }
        ]}
      ]}
    ]}
  ]
}

AlterClientQuotas (2.6.0)

Code Block
{
  "nameapiKey": "EntityType"49,
  "type": "stringrequest",
  "versionsname": "0+AlterClientQuotasRequest",
        "aboutvalidVersions": "The entity type." }0",
  "flexibleVersions": "none",
  "fields": [
    { "name": "EntityNameEntries", "type": "string[]EntryData", "versions": "0+",
        "about": "The entity name quota configuration entries to alter." }, "fields": [
    ]},
  ]
}

{
  "apiKeyname": 49"Entity",
  "type": "response[]EntityData",
  "nameversions": "ResolveClientQuotasResponse0+",
  "validVersions      "about": "0",
  "flexibleVersions": "none",
  The quota entity to alter.", "fields": [
        { "name": "ThrottleTimeMsEntityType", "type": "int32string", "versions": "0+",
          "about": "The durationentity 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": "[]QuotaEntryData", "versionstype." },
        { "name": "EntityName", "type": "string", "versions": "0+", "nullableVersions": "0+",
          "about": "The name of the entity, or null if the default." }
   "about": "Resolved quota entries.", "fields": []},
      { "name": "ErrorCodeOps", "type": "int16[]OpData", "versions": "0+",
        "about": "TheAn errorindividual code,quota orconfiguration `0`entry if the resolved quota description succeeded." },
to alter.", "fields": [
        { "name": "ErrorMessageKey", "type": "string", "versions": "0+",
 "nullableVersions": "0+",
        "about": "The error message, or `null` if the resolved quota descriptionconfiguration succeededkey." },
        { "name": "QuotaEntityValue", "type": "[]QuotaEntityfloat64", "versions": "0+",
          "about": "ResolvedThe value quotato entries.", "fields": [set, otherwise ignored if the value is to be removed." },
        { "name": "EntityTypeRemove", "type": "stringbool", "versions": "0+",
          "about": "The entity type." },Whether the quota configuration value should be removed, otherwise set." }
      ]}
    ]},
    { "name": "EntityNameValidateOnly", "type": "stringbool", "versions": "0+",
          "about": "The entity name." }
    Whether the alteration should be validated, but not performed." }
  ]
},

{
  "apiKey": 49,
   { "nametype": "QuotaValuesresponse",
  "typename": "[]QuotaValueDataAlterClientQuotasResponse",
  "versionsvalidVersions": "0+",
        "aboutflexibleVersions": "Quota configuration values.none",
  "fields": [
        { "name": "TypeThrottleTimeMs", "type": "stringint32", "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 "about": "Theviolate any quota type." },
        { "name": "EntryEntries", "type": "[]ValueEntryDataEntryData", "versions": "0+",
          "about": "QuotaThe quota valueconfiguration entries to alter.", "fields": [
          { "name": "QuotaEntityErrorCode", "type": "[]ValueQuotaEntityint16", "versions": "0+",
            "about": "ResolvedThe quotaerror entries."code, "fields": [
      or `0` if the quota alteration succeeded." },
      { "name": "EntityTypeErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+",
        "about": "The error message, or `null` if "about": "The entity typethe quota alteration succeeded." },
            { "name": "EntityNameEntity", "type": "string[]EntityData", "versions": "0+",
              "about": "The quota entity nameto alter." }
          ]},
 , "fields": [
         { "name": "ValueEntityType", "type": "int64string", "versions": "0+",
            "about": "The quotaentity configuration valuetype." }
        ]}
   ,
   ]}
    ]}
  ]
}

AlterClientQuotas:

Code Block
{
  "apiKeyname": 50"EntityName",
  "type": "requeststring",
  "nameversions": "AlterClientQuotasRequest0+",
  "validVersionsnullableVersions": "0+",
    "flexibleVersions      "about": "none",The name of the entity, or null if the default." }
      ]}
  "fields": [
    { "name": "Entry", "type": "[]EntryData", "versions": "0+",
      "about": "The quota configuration entries to alter.", "fields": [  ]}
  ]
}

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.
     *
 { "name": "QuotaEntity", "type": "[]QuotaEntity", "versions": "0+",
        "about": "The quota entity to alter.", "fields": [ * @param buffer The buffer to read from
     * @return The long value read
     */
   { "name": "EntityType", "type": "string", "versions": "0+",
public static double readDouble(ByteBuffer buffer) {
        return Double.longBitsToDouble(buffer.getLong());
   "about": "The entity type." }, }

    /**
     * Write the { "name": "EntityName", "type": "string", "versions": "0+",
          "about": "The name of the entity." }
      ]},given double following the double-precision 64-bit format IEEE 754 value into the buffer.
     *
     * @param value The value to write
     * { "name": "Op", "type": "[]OpData", "versions": "0+",@param buffer The buffer to write to
     */
   "about": "An individual quota configuration entry to alter.", "fields": [ public static 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": "Type", "type": "string", "versions": "0+",
          "about": "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);
        }

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

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

{
    "apiKey": 50,
  "type": "response",
  "name": "AlterClientQuotasResponse",
  "validVersions": "0",
  "flexibleVersions": "none",
  "fields": [
 @Override
        public String typeName() {
       { "name": "ThrottleTimeMs", "type": "int32", "versions": return "0+DOUBLE",;
        }

    "about": "The duration in milliseconds@Override
 for which the request was throttled due topublic aDouble quota violation, or zero if the request did not violate any quota." },
    { "name": "Entry", "type": "[]EntryData", "versions": "0+",
validate(Object item) {
            if (item instanceof Double)
          "about": "The quota configuration entries to alter.", "fields": [return (Double) item;
      { "name": "ErrorCode", "type": "int16", "versions": "0+", else
        "about": "The error code, or `0` if the quotathrow alteration succeeded." },
      { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+",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 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.

...