Kafka client quotas

Operations guide: configure, deploy and observe the gateway. What the feature does, product side: product guide; calling it: developer guide.

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). A client that ignores the field - librdkafka never applied it (librdkafka#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): 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:

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:

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); a broker quota on that principal slows every tenant together, the gateway's quotas each its own:

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:

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:

# 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:

# 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
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

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.