# Kafka client quotas

> Throttle Kafka producers and consumers per principal and per tenant, never cutting them, and read the listener's signals.

Guide - Since version 1.6.0 - Flexway v1.7.1 - https://flexway.io/en/docs/operations-kafka-quotas/

> Operations guide: configure, deploy and observe the gateway. What the feature does, product side: [product guide](https://flexway.io/en/docs/product-overview/); calling it: [developer guide](https://flexway.io/en/docs/developer-overview/).

The Kafka listener's quotas are declared under `listeners.kafka.quotas`; its signals then say who is throttled, refused or closed, and why.

## Steps

### Quotas: KIP-219 throttling

Bytes produced, bytes fetched and requests per second, for each **verified principal** and for each
**tenant** (`listeners.kafka.quotas`). Never a cut: past a rate, the response
leaves **at once** with `throttle_time_ms`, then the gateway **stops reading** the client's connection for
that long ([KIP-219](https://cwiki.apache.org/confluence/display/KAFKA/KIP-219+-+Improve+quota+communication)). A client that ignores
the field - librdkafka never applied it ([librdkafka#5592](https://github.com/confluentinc/librdkafka/issues/5592)) -
is slowed all the same: what it sends waits in the socket. Without `quotas`, nothing is throttled.

| Key | Why | Default |
| --- | --- | --- |
| `client.producer_byte_rate` | Bytes produced per second (`Produce` requests, headers included) by one principal: the token's subject, the API key or the certificate - **never** the `client.id`, which a client picks freely | no bound |
| `client.consumer_byte_rate` | Bytes fetched per second (`Fetch` responses) by one principal | no bound |
| `client.request_rate` | Requests per second of one principal, every API, once authenticated | no bound |
| `tenant.producer_byte_rate`, `tenant.consumer_byte_rate`, `tenant.request_rate` | The same for every principal of a tenant together: a tenant buys no more throughput by adding keys | no bound |
| `window` | The burst admitted at once: a bucket holds `rate x window` (Kafka's quota windows are about a second) | `1s` (from `100ms` to `10s`) |
| `max_throttle` | The cap of `throttle_time_ms` - and of the silence after it: below the clients' `request.timeout.ms` (30 s by default in Kafka's clients), which a throttle must never exhaust | `10s` (from `100ms` to `30s`; provisional default, to be set by performance measurements) |
| `quota` | The name of a `quota` policy ([Setting quotas](https://flexway.io/en/docs/operations-quotas/)): the tenant's **daily or monthly** budget of `Produce` requests, counted under that name as for HTTP; spent, a `Produce` is refused `POLICY_VIOLATION` (44, not retriable) and nothing is written | none |

Every key is reloadable: the rate in force applies from the next request, and what a principal spent is
kept. With `server.counters`, the buckets are shared across replicas like the HTTP rate limits: each
replica decides alone and reconciles every `sync` (a rate held to within a `sync` window; the gap is not
measured yet). Without it, each replica counts for itself.

| Case | What happens |
| --- | --- |
| A client that honours KIP-219 (Java, franz-go) | An **immediate** response with `throttle_time_ms`; it waits that long before its next request |
| A client that ignores the field (librdkafka: confluent-kafka-go, Python, .NET) | The same response; it sends again at once, but its next request is served only once the delay ends: the rate holds without its cooperation |
| A `Fetch` of a principal or tenant already in debt | An **empty** response at once with `throttle_time_ms`; nothing is asked of the broker |
| The broker also throttles the gateway (its own quotas on the gateway's principal) | `throttle_time_ms` = the **larger** of the two delays; the broker's silence is respected: nothing is sent to it on that connection before it ends |
| The tenant's daily or monthly quota is spent | A `POLICY_VIOLATION` refusal on `Produce` (cause `quota`), not throttling; reads (`Fetch`) are **not yet** counted in the budget: their refusal code must first be checked against the main client libraries (a Java consumer throws on an unexpected code) |

### Per tenant

A tenant does not take all the brokers' bandwidth, whatever its number of keys:

```yaml
listeners:
  kafka:
    quotas:
      tenant: { producer_byte_rate: 10MiB, consumer_byte_rate: 40MiB, request_rate: 2000 }
```

### Per client

Each principal (token, key, certificate) has its bucket, under its tenant's:

```yaml
listeners:
  kafka:
    quotas:
      client: { producer_byte_rate: 1MiB, consumer_byte_rate: 4MiB, request_rate: 200 }
      tenant: { producer_byte_rate: 10MiB, consumer_byte_rate: 40MiB }
      window: 2s           # bursts of two seconds of rate
      max_throttle: 5s     # clients with a short request.timeout.ms
server:
  counters: { store: "${COUNTERS_URL}" }   # the rate held across every replica together
```

### With the broker's quotas

Nothing to declare: the gateway relays the larger of the two delays. The
broker sees the gateway's principal (`upstreams.<name>.sasl.username`) and the tenant's client id
(`acme.app-1`, [Names of groups and transactions](https://flexway.io/en/docs/operations-kafka-reference/#names-of-groups-and-transactions)); a broker quota on that principal slows
**every** tenant together, the gateway's quotas each its own:

```bash
kafka-configs --bootstrap-server kafka-0.kafka:9093 --command-config admin.properties \
  --alter --entity-type users --entity-name flexway --add-config producer_byte_rate=104857600
```

### Daily or monthly budget

The same plan as for HTTP, counted on `Produce`:

```yaml
listeners:
  kafka:
    quotas: { quota: plans }
server:
  counters: { store: "${COUNTERS_URL}" }   # required by every quota policy
policies:
  definitions:
    plans: { type: quota, config: { limits: { day: 100000, month: 2000000 } } }
```

## Verify

### A librdkafka client

Nothing to change on the client; what it sees past the quota:

```properties
# confluent-kafka-go, Python, .NET: ordinary properties
bootstrap.servers=bootstrap.shop.kafka.example.com:9093
security.protocol=SASL_SSL
sasl.mechanisms=PLAIN
sasl.username=acme-orders-producer
sasl.password=${FLEXWAY_KAFKA_KEY}
# past the quota: the producer moves at the declared rate; its "throttle" statistic (statistics.interval.ms)
# shows throttle_time_ms; request.timeout.ms (30 s) stays above max_throttle
```

### Reading the signals

Prometheus names:

```promql
# share of throttled requests, per virtual cluster
sum by (cluster) (rate(flexway_kafka_requests_total{decision="throttled"}[5m]))
  / sum by (cluster) (rate(flexway_kafka_requests_total[5m]))
# refusals by cause (the daily or monthly quota aside)
sum by (cause) (rate(flexway_kafka_requests_total{decision="refused", cause!="quota"}[5m]))
# connections refused by bound; normal closures apart
sum by (bound) (rate(flexway_kafka_connections_refused_total[5m]))
sum by (cause) (rate(flexway_kafka_connections_closed_total[5m]))
# p95 wait for a tenant's share
histogram_quantile(0.95, sum by (le, tenant) (rate(flexway_kafka_tenant_budget_wait_seconds_bucket[5m])))
```

## Signals: metrics, traces, audit

The `flexway.kafka.*` metrics leave over OTLP with the others (one path, `service.instance.id` per
replica); every dimension is **closed**: the contract's clusters, the API table's names, the sets below,
and `tenant` under the rule of every tenant dimension - 200 tenants keep their series, the next share
`_other`. **Never** a subject, a key, a certificate, a `client.id`, a physical Kafka topic or an address.

| Metric | Dimensions | What it says |
| --- | --- | --- |
| `flexway.kafka.connections` (gauge) | `cluster`, `tenant`, `mechanism` (`oauthbearer`, `api_key`, `mtls`, `none` before authentication; `_none` and `_anonymous` before the SNI) | Open connections: on a replica, the occupation of `max_connections` - which refuses nothing, one more connection waits in the kernel's queue |
| `flexway.kafka.requests` | `cluster`, `tenant`, `api`, `decision` (`allowed`, `throttled`, `refused`), `cause` (`acl`, `admin_api`, `quota`, `unknown_topic`, `authentication`, `kms_unavailable`, `unsupported`, `key_destroyed`; `none`) | Requests judged; `refused` as soon as the gateway refused any part |
| `flexway.kafka.bytes` | `cluster`, `tenant`, `direction` (`produce`, `fetch`) | The bytes the quotas count |
| `flexway.kafka.throttle` (histogram, s) | `cluster`, `tenant`, `quota` (`producer_byte_rate`, `consumer_byte_rate`, `request_rate`, `broker`), `scope` (`client`, `tenant`, `broker`) | `throttle_time_ms` of each throttled response |
| `flexway.kafka.connections.refused` | `bound` (`max_unauthenticated_connections`, `max_connections_per_ip`, `max_connections_per_tenant`, `max_connection_rate_per_tenant`), `tenant` for the two tenant bounds only | Connections a bound refused; never a tenant before authentication, never an address; a refusal by `max_connections_per_tenant` is **not** counted as `authentication` too |
| `flexway.kafka.connections.closed` | `cause` (`auth_timeout` without tenant, `idle_timeout`, `max_session_lifetime`), `tenant` | **Normal** closures, never mixed with refusals; `max_session_lifetime`: a session that reached its end (the bound, `exp` or `NotAfter`) without re-authenticating |
| `flexway.kafka.tenant_budget` | `cluster`, `tenant`, `outcome` (`waited`, `closed`) | Responses that waited for the tenant's share (`max_tenant_in_flight_bytes`, `wait` mode); connections closed at it (`close` mode) |
| `flexway.kafka.tenant_budget.wait` (histogram, s) | `cluster`, `tenant` | How long a response waited for that share |
| `flexway.kafka.encryption` | `cluster`, `tenant`, `op` (`encrypt`, `decrypt`), `outcome` (`ok`, `kms_unavailable`, `key_destroyed`) | [Field encryption](https://flexway.io/en/docs/operations-kafka-encryption/) |
| `flexway.kafka.upstream.connections` | `cluster`, `broker` (node id, or `bootstrap`), `outcome` (`ok`, `failed`) | The gateway's connections to each broker: health per broker |

| Key | Why | Default |
| --- | --- | --- |
| `observability.kafka.budget_wait_buckets` | Boundaries of `flexway.kafka.tenant_budget.wait`, increasing, 20 at most: to tighten when a fleet's normal waits stay under a millisecond. **Not reloadable** (a histogram's boundaries are set when it is created) | `1ms`, `5ms`, `10ms`, `50ms`, `100ms`, `500ms`, `1s`, `5s`, `10s`, `30s` (the write timeout) |
| `observability.kafka.closures` | Count the normal closures: `false` on a fleet of short-lived clients where they say nothing. Reloadable | `true` |

```yaml
observability:
  kafka:
    budget_wait_buckets: [100us, 1ms, 10ms, 100ms, 1s, 10s, 30s]
    closures: false
```

### Traces

One span per sampled **authenticated** request (`kafka <API>`, attributes `kafka.cluster`,
`tenant`, `auth.mechanism`, `kafka.decision`, `kafka.cause`, `kafka.throttle_time_ms`) - never a subject nor
a Kafka topic name.

### Audit

Every refused request and every **administration**
request (allowed or refused), in the same chain as HTTP - `host` = the virtual cluster, `method` =
`KAFKA`, `operation` = the API, `code` = the cause, the subject hashed; allowed data requests are counted,
not logged one by one.