...

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 hasn't been clean 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 useful information beyond what the configuration APIs could offerprovide, and provide enable for future extensibility as quotas types are added or evolved.

...

kafka-quotas.sh/QuotasCommand:

A QuotasCommand would be constructed with an associated bin/kafka-quotas.sh script for managing quotas via command line.

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. --names=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:
--match: Comma-separated prefix=name pairs, e.g. --match=prefix=user

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

Exclusive to --alter:
--add: Adds or updates a configuration entry to the given value. The unit must be specified.
--delete: Removes a configuration entry.
--validate-only: If set, validates the alteration but doesn't perform it.

TODO: Specify output.

, 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-user-".

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

List:

$./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:

$./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:

$./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

Proposed Changes

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

...

{
  "apiKey": 50,
  "type": "request",
  "name": "AlterQuotasRequest",
  "validVersions": "0",
  "flexibleVersions": "none",
  "fields": [
    { "name": "Entry", "type": "[]EntryData", "versions": "0+",
      "about": "The quota configuration entries to alter.", "fields": [
      { "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 name of the entity." }
      ]},
      { "name": "Op", "type": "[]OpData", "versions": "0+",
        "about": "An individual quota configuration entry to alter.", "fields": [
        { "name": "Type", "type": "string", "versions": "0+",
          "about": "The quota type." },
        { "name": "Units", "type": "string", "versions": "0+",
          "about": "The units for the quota type." },
        { "name": "Value", "type": "int64", "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": 50,
  "type": "response",
  "name": "AlterQuotasResponse",
  "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": "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": "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 name of the entity." }
      ]}
    ]}
  ]
}

kafka-configs.sh/ConfigCommand:

...

Compatibility, Deprecation, and Migration Plan

...