# Native Kafka clients: a virtual cluster per tenant

> Kafka producers, consumers and admin tools - any library: franz-go, librdkafka (confluent-kafka-go, Python, .NET), the Java client - connect to the gateway…

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

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

Kafka producers, consumers and admin tools - **any library**: franz-go, librdkafka (confluent-kafka-go,
Python, .NET), the Java client - connect to the gateway instead of the broker, **with no code change**.
Each tenant sees **its own** virtual cluster: only the topics the contract declares and its operations
open, under their short names. The client writes to `orders`, the broker receives `acme.orders`;
consumer groups and transactional ids are written in sub-namespaces of the tenant (`acme.~g.billing`,
`acme.~t.tx-orders`, [Names of groups and transactions](#names-of-groups-and-transactions)), and
so is the client id (`acme.app-1`) - never another tenant's name, neither in an answer nor toward the
broker. The gateway presents **itself** to the broker (the upstream's TLS and SASL): the client never
knows its credentials. Available since `1.6.0`.

> **Not complete yet** - identities, ACLs and byte and request quotas are served and tested end to end.
> Two points remain open, with no announced version:
>
> - a tenant's daily or monthly budget counts writes (`Produce`) only: reads (`Fetch`) are not counted
>   until their refusal code is checked against the main client libraries
>   ([Quotas](#quotas-kip-219-throttling));
> - the defaults of `max_throttle` and `max_connection_rate_per_tenant` are provisional, pending
>   performance measurements: they may change in a minor version - if you depend on them, write them in
>   your configuration.

<!-- example: examples/kafka-proxy/flexway.yaml -->

```yaml
# Kafka clients behind the gateway (1.6.0): producers, consumers and admin tools of any
# library reach the virtual cluster "shop" the AsyncAPI contract declares - each tenant inside its own
# names, under the same identities as HTTP. Here the fallback inside a mesh, a port per broker; the TLS
# listener, brokers named by SNI, is in the guide.
version: 1
contracts:
  asyncapi: ./asyncapi.yaml
upstreams:
  events:
    protocol: kafka
    brokers: [kafka-0.kafka:9093, kafka-1.kafka:9093, kafka-2.kafka:9093]
    tls: { server_name: kafka.internal }
    sasl: { mechanism: SCRAM-SHA-512, username: flexway, password: "file:/etc/flexway-secrets/kafka/password" }
    timeout: 5s                                  # a connection to a broker, TLS and SASL included
auth:
  issuers:
    - name: customers
      discovery: https://login.example.net/.well-known/openid-configuration
      audience: kafka
      tenant_claim: org                          # every topic, group and transaction is the tenant's
  api_keys:
    store: "file:./api-keys.yaml"                # PLAIN: the key's id as the user, the key as the password
listeners:
  kafka:
    plaintext:
      mesh_only: true                            # clear text only where the mesh encrypts between pods
      advertised_host: flexway-kafka.flexway.svc.cluster.local
      clusters:
        shop: { ports: { from: 19092, to: 19095 }, first_node_id: 0 }   # bootstrap 19092, node n at 19093+n
```

```yaml
# examples/kafka-proxy/asyncapi.yaml
asyncapi: 3.0.0
info: { title: Shop events, version: "1.0.0" }
channels:
  orders:
    address: "{tenant}.orders"
    parameters:
      tenant: { x-flexway-bind: tenant }
    messages:
      order: { payload: { type: object, required: [id], properties: { id: { type: string } } } }
operations:
  produceOrders:
    action: send
    channel: { $ref: "#/channels/orders" }
    x-flexway-upstream: events
    x-flexway-kafka-proxy: { cluster: shop, transactions: true }
    security:
      - $ref: "#/components/securitySchemes/apiKey"
      - $ref: "#/components/securitySchemes/ordersWrite"
  consumeOrders:
    action: receive
    channel: { $ref: "#/channels/orders" }
    x-flexway-upstream: events
    x-flexway-kafka-proxy: { cluster: shop, admin_scope: "orders:admin" }
    security:
      - $ref: "#/components/securitySchemes/apiKey"
      - $ref: "#/components/securitySchemes/ordersRead"
components:
  securitySchemes:
    apiKey: { type: plain }                      # SASL PLAIN: an API key bound to the operation
    ordersWrite:                                 # SASL OAUTHBEARER: a token carrying the scope
      type: oauth2
      flows: { clientCredentials: { tokenUrl: "https://login.example.net/token", availableScopes: { "orders:write": Produce orders } } }
      scopes: ["orders:write"]
    ordersRead:
      type: oauth2
      flows: { clientCredentials: { tokenUrl: "https://login.example.net/token", availableScopes: { "orders:read": Read orders } } }
      scopes: ["orders:read"]
```

**The contract's convention** (common to the listeners: the one `x-flexway-device` follows at `1.7.0`):

| Where | Shape | Meaning |
| --- | --- | --- |
| AsyncAPI operation | `x-flexway-kafka-proxy: { cluster, transactions, admin_scope }` | The operation is served to the Kafka clients of the virtual cluster `cluster` (a DNS label); `transactions: true` (a `send` operation) opens the transactional producer; `admin_scope` opens the topic's administration to a token carrying that scope |
| `action` | `send`: the client **produces** (`Produce`); `receive`: it **reads** (`Fetch`, offsets, groups) | The same direction as in the asynchronous plane: seen from the broker, `send` writes to the channel |
| Channel | Its **key** is the topic name the client sees; its `address` is `<prefix>{tenant}.<name>`, its only parameter bound to the tenant | The prefix (empty, or ending with a dot) is the same for every served channel of one upstream; groups and transactional ids are written `<prefix><tenant>.<name>` |
| The operation's `security` | AsyncAPI 3 schemes: `oauth2` or `openIdConnect` with `scopes`, `plain` (or `apiKey` with `in: password`), `X509` | One alternative is enough; read for the operations served to Kafka clients only - an HTTP door keeps its OpenAPI operation's security |

**Identities** - the same as over HTTP:

| Client mechanism | Contract alternative | What the gateway checks |
| --- | --- | --- |
| SASL `OAUTHBEARER` | `oauth2` / `openIdConnect` + `scopes` | The token like a `Bearer` one: declared issuer, audience, signature, delegation chain; `exp` **required**: the session lasts until it, announced to the client (`session_lifetime_ms`, KIP-368) - it re-authenticates before, or the connection is closed; a re-authentication proves the same principal |
| SASL `PLAIN` | `plain` (or `apiKey`, `in: password`) | The API key by its digest ([API keys](https://flexway.io/en/docs/operations-api-keys/)): the **user name is the key's id**, the password the key; bound to its tenant and operations, bounded by its `expires_at` |
| Client certificate (mTLS) | `X509` | A certificate verified by the TLS listener (`listeners.kafka.tls.client_ca_file`) and declared in `auth.mtls.clients` ([Client certificates (mTLS)](https://flexway.io/en/docs/operations-client-certificates/)), bound to its tenant and operations; no SASL exchange |

An identity **without a tenant**, or whose tenant is not written in letters, digits and `-` (a dot, an
underscore), is refused at authentication - [Names of groups and transactions](#names-of-groups-and-transactions).

**The TLS listener, brokers named by SNI**: one port, the SNI name says which broker is meant -
`bootstrap.<cluster>.<sni_domain>` to bootstrap, `broker-<id>.<cluster>.<sni_domain>` for each;
`Metadata` and `FindCoordinator` answer these names. The certificate covers `*.<cluster>.<sni_domain>`
for each virtual cluster (a wildcard covers one label); every such name resolves to the gateway (a
wildcard DNS record, or the chart's `LoadBalancer` `Service`).

```yaml
# flexway.yaml - the TLS listener; the rest as in the example above
listeners:
  kafka:
    address: ":9093"
    tls:
      cert_file: /etc/flexway-kafka-tls/tls.crt        # *.shop.kafka.example.com
      key_file: /etc/flexway-kafka-tls/tls.key
      client_ca_file: /etc/flexway-kafka-tls/ca.crt    # optional: client certificates (X509)
    sni_domain: kafka.example.com
    advertised_port: 9093                # the port the client sees (the LoadBalancer's)
    max_request_bytes: 16MiB             # an authenticated request; 64 KiB before authentication
    max_response_bytes: 64MiB            # a broker's answer; a Fetch is held below half of it
    max_connections: 4096
    auth_timeout: 10s
    idle_timeout: 10m
auth:
  mtls:
    clients:
      - { id: billing, subject: "CN=billing-app", tenant: acme, operations: [consumeOrders] }
```

**The plaintext fallback, a port per broker** (the example): without TLS there is no SNI; each virtual
cluster gets a range - `from` to bootstrap, `from + 1 + n` for node `first_node_id + n` -, announced
under `advertised_host`. Refused at start without `mesh_only: true`: clear text only crosses a mesh
that encrypts (Linkerd, Istio in strict mTLS). The range covers **every** node id of the cluster;
otherwise the client gets an unreachable address and the log says so.

**Upstreams** - Apache Kafka (`apache/kafka`, Confluent Platform, MSK, Strimzi) and Redpanda are
declared alike; the gateway reads each broker's versions and announces to the client only their
intersection with what it judges:

```yaml
upstreams:
  events:                                # Apache Kafka: SCRAM over TLS
    protocol: kafka
    brokers: [kafka-0.kafka:9093, kafka-1.kafka:9093, kafka-2.kafka:9093]
    tls: { ca_file: /etc/flexway-secrets/kafka/ca.crt, server_name: kafka.internal }
    sasl: { mechanism: SCRAM-SHA-512, username: flexway, password: "file:/etc/flexway-secrets/kafka/password" }
  events-redpanda:                       # Redpanda: SASL SCRAM-SHA-256, or the gateway's mTLS
    protocol: kafka
    brokers: [redpanda-0.redpanda:9093]
    tls: { ca_file: /etc/flexway-secrets/redpanda/ca.crt, cert_file: /etc/flexway-secrets/redpanda/tls.crt, key_file: /etc/flexway-secrets/redpanda/tls.key }
    timeout: 5s                          # a connection to a broker, TLS and SASL included
```

**Clients** - standard Kafka properties, pointed at the gateway:

```go
// franz-go: an API key over SASL PLAIN on TLS
cl, err := kgo.NewClient(
	kgo.SeedBrokers("bootstrap.shop.kafka.example.com:9093"),
	kgo.DialTLSConfig(&tls.Config{MinVersion: tls.VersionTLS12}),
	kgo.SASL(plain.Auth{User: "acme-orders-producer", Pass: os.Getenv("FLEXWAY_KAFKA_KEY")}.AsMechanism()),
	kgo.TransactionalID("tx-orders"), // becomes acme.~t.tx-orders at the broker
)
// a token: kgo.SASL(oauth.Oauth(func(ctx context.Context) (oauth.Auth, error) { return oauth.Auth{Token: tok}, nil }))
```

```properties
# librdkafka (confluent-kafka-go, Python, .NET): an OAUTHBEARER token of the declared issuer
bootstrap.servers=bootstrap.shop.kafka.example.com:9093
security.protocol=SASL_SSL
sasl.mechanisms=OAUTHBEARER
sasl.oauthbearer.method=oidc
sasl.oauthbearer.client.id=billing
sasl.oauthbearer.client.secret=${BILLING_SECRET}
sasl.oauthbearer.token.endpoint.url=https://login.example.net/token
sasl.oauthbearer.scope=orders:read
group.id=billing                         # becomes acme.~g.billing
```

```properties
# Java client (producer, consumer, AdminClient, Kafka Connect): an API key over PLAIN
bootstrap.servers=bootstrap.shop.kafka.example.com:9093
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="acme-orders-reader" password="${FLEXWAY_KAFKA_KEY}";
# a client certificate instead: security.protocol=SSL, ssl.keystore.location=..., ssl.truststore.location=...
# group.protocol=classic (the default): KIP-848's group protocol is not served
```

**The decision, by Kafka API** - a closed table: an API missing from it is refused (`domain/kafkaproxy`,
held equal to franz-go's list by a test):

| API | Decision |
| --- | --- |
| `ApiVersions`, `SaslHandshake`, `SaslAuthenticate` | Answered by the gateway; `ApiVersions` announces only what it judges, at the intersection with the brokers' versions - never a version that names topics by id (`Produce` <= 12, `Fetch` <= 12, `Metadata` 1 to 12, `OffsetCommit` and `OffsetFetch` <= 9) |
| `Metadata`, `DescribeCluster` | The session's topics only, under their short names - a topic unknown or not opened is `UNKNOWN_TOPIC_OR_PARTITION`, never asked of the broker; brokers renamed (SNI or fallback port); cluster id = the virtual cluster's; automatic creation off; authorized operations never asked |
| `Produce` | A `send` operation; without one `TOPIC_AUTHORIZATION_FAILED`; record batches relayed **without being decoded** |
| `Fetch`, `ListOffsets`, `OffsetForLeaderEpoch` | A `receive` operation; `Fetch` bounded to `max_response_bytes / 2` and `max_fetch_wait`, always full (no incremental session); a replica's request is relayed as a consumer's (`ReplicaID` `-1`) |
| `FindCoordinator`, `JoinGroup`, `SyncGroup`, `Heartbeat`, `LeaveGroup`, `OffsetCommit`, `OffsetFetch`, `DescribeGroups`, `ListGroups` | A `receive` operation opens groups; `GROUP_AUTHORIZATION_FAILED` otherwise; `ListGroups` lists the tenant's only |
| `InitProducerId`, `AddPartitionsToTxn`, `AddOffsetsToTxn`, `EndTxn`, `TxnOffsetCommit` | Idempotent producer: a `send` operation; transactional: a `send` operation declaring `transactions: true` (`TRANSACTIONAL_ID_AUTHORIZATION_FAILED` otherwise) |
| `CreateTopics`, `DeleteTopics`, `DeleteRecords`, `CreatePartitions`, `DescribeConfigs`, `IncrementalAlterConfigs` (topics), `DeleteGroups`, `OffsetDelete` | **Refused** (`TOPIC_AUTHORIZATION_FAILED`, `GROUP_AUTHORIZATION_FAILED`) unless a token carries the operation's `admin_scope`; never a topic the contract does not declare; never a broker resource nor a group outside the tenant |
| ACLs, quotas, SCRAM credentials, reassignments, delegation tokens, `AlterConfigs`, broker-to-broker APIs, KIP-848 and KIP-932 groups, telemetry | **Never announced**; a client that sends one anyway is disconnected |

**Refused at load**, all at once: `x-flexway-kafka-proxy` without `listeners.kafka` (or the reverse), a
cluster name that is not a DNS label, a served operation without `security`, with a scheme the listener
does not verify (`scramSha256`, `gssapi`, `apiKey` outside the password), public, or with
`x-flexway-validation: full` (batches cross undecoded), `transactions` on a `receive` operation, a
channel whose address is not `<prefix>{tenant}.<name>` or whose key is not a topic name (nor `__`), two
prefixes on one upstream or on two upstreams reaching one broker, two channels whose names Kafka confuses (`.` read as `_`), a cluster on two upstreams, a token admitted without `auth.issuers`, a key
without `auth.api_keys`, a certificate without `client_ca_file`, fallback ports of a cluster the
contract does not declare, a served AsyncAPI operation's id equal to an OpenAPI operation's (a key bound
to one would open the other).

**Reload**: the address, TLS, `sni_domain`, `advertised_port`, the fallback range and
`max_connections`, `max_total_in_flight_bytes` and `max_tenant_in_flight_bytes` are read at start (`listeners.kafka` is refused on
reload if they change); the contract, identities, bounds and timeouts apply to the next request - a
connection keeps its identity as long as its credential stays the configuration's in force
([Sessions: revocation, lifetime, certificates](#sessions-revocation-lifetime-certificates)), and each request is judged on it.

**One upstream connection per client connection and broker**: a Kafka session (idempotent producer,
transactions, the broker's throttling) is never shared between clients; stopping drains (in-flight
requests answered, then the connection closed: the client reconnects to another replica). **Minimal
broker ACLs** for the gateway's principal: read, write and describe the served `<prefix><tenant>.*`
topics, the `<prefix><tenant>.~g.*` groups and `<prefix><tenant>.~t.*` transactional ids, `IdempotentWrite`; administration only
when an `admin_scope` is declared. The brokers' `NetworkPolicy` admits the gateway only: no client
bypasses it with the real addresses, which `Metadata` never gives.

## Names of groups and transactions

A tenant's topics are written `<prefix><tenant>.<name>`, its consumer groups and transactional ids in
**sub-namespaces of their own** (decision D-K7): `<prefix><tenant>.~g.<name>` and `<prefix><tenant>.~t.<name>`.
A Kafka topic name never holds `~`: no group of an application that names its groups after its topics, no
`flexway.ws.*` or `flexway.deliver.*` group of the gateway falls inside - the namespace is the listener's
alone. A served tenant is written in letters, digits and `-`: no dot (it ends the tenant in a name), no
underscore (Kafka reads `.` and `_` as one character in its metrics and refuses a topic that collides that
way).

| Client `acme` writes | The broker holds |
| --- | --- |
| topic `orders` | `acme.orders` (empty prefix) |
| `group.id=billing` | `acme.~g.billing` |
| `transactional.id=tx-orders` | `acme.~t.tx-orders` |
| `client.id=app-1` | `acme.app-1` (the broker's logs and quotas) |
| `ListGroups` | the `acme.~g.*` groups only, answered `billing`... |

**Broker ACLs** for the gateway's principal, empty prefix: `Topic` prefixed `acme.` (per tenant, or `*` when
the gateway is alone), `Group` prefixed `acme.~g.`, `TransactionalId` prefixed `acme.~t.`, `IdempotentWrite`.
Two upstreams reaching one broker under two prefixes are refused at load; addresses do not tell everything
(an alias, a load balancer, another listener of the same cluster), so at its first connection the listener
also reads the **cluster id** of each upstream's `Metadata`: a second upstream naming the same physical
cluster under another prefix is refused (log `kafka connection refused: two upstreams reach the same
physical cluster`, level Error). A broker that gives no cluster id (`Metadata` v0, v1) is told apart by its
addresses only.

## Sessions: revocation, lifetime, certificates

| Behaviour | Why | Default |
| --- | --- | --- |
| **Revocation on reload** | A key removed from the store, bound to other operations or to another tenant, a certificate removed from `auth.mtls.clients` or rebound, a token the issuers in force no longer accept: at the next request the open connection is **closed** - the client reconnects, and its new authentication is judged on the configuration in force | Always |
| `max_session_lifetime` | Every session is bounded, whatever its credential: a SASL client re-authenticates before it (`session_lifetime_ms` announced, KIP-368 - franz-go, librdkafka and Java do it on their own), a certificate's connection is closed at it; a revocation is seen at the latest then | `1h` (`1m` to `24h`) |
| A certificate's session | Bounded by the certificate's end of validity (`NotAfter`) too | Always |
| A token's session | **Verified again on every reload** by the issuers in force (the token is kept in memory for it, never written): an issuer removed from `auth.issuers`, an audience or a delegation bound changed, another principal behind it close the connection; bounded by its `exp` too | Always |

```yaml
listeners:
  kafka:
    max_session_lifetime: 30m      # a removed key, a revoked certificate: seen within 30 min at the latest
```

Remove the key `acme-orders-producer` from the store (the mounted `Secret`, read again on reload): the
producers opened with it see their connection closed at their next request, reconnect and fail
authentication (`SASL_AUTHENTICATION_FAILED`) - what to look for in their logs.

## Listener bounds

Each bound protects the gateway from a client - authenticated or not - that would take too much:

| Key | Why | Default |
| --- | --- | --- |
| `auth_timeout` | An **absolute** time to authenticate, TLS handshake included: a client that sends frames without ever authenticating is closed at it | `10s` |
| `max_unauthenticated_connections` | Places for connections not authenticated yet: past it a new one is closed at once - an anonymous flood never takes the authenticated clients' places | `256` |
| `max_connections_per_ip` | Connections of one address - a client opens one per broker, so a few dozen at most; below `max_unauthenticated_connections` so that one address cannot take every unauthenticated place. `0` behind a proxy that hides the clients' addresses (a `LoadBalancer` without `externalTrafficPolicy: Local`) | `16` |
| `max_connections_per_tenant` | Connections of one tenant, at authentication (`SASL_AUTHENTICATION_FAILED` past it): one tenant does not exhaust the listener for the others | `1024` |
| `max_connection_rate_per_tenant` | New connections of one tenant per second, at authentication (`SASL_AUTHENTICATION_FAILED` past it; a re-authentication takes none): a client reconnecting in a loop spends its tenant's handshakes, never the replica's. `0`: not bounded | `100` (provisional: to be set by performance measurements) |
| `max_fetch_wait` | How long a `Fetch` waits at the broker (the client's `max_wait_ms`, brought down to it): an upstream connection is never held without end | `30s` |
| `max_in_flight_bytes` | Broker answers held in memory per connection, read and not yet written: past it the broker is not read until the client reads (one answer always crosses alone) | `64MiB` |
| `max_tenant_in_flight_bytes` | The same bound for all the connections of **one tenant** - its share of the replica's: a tenant whose clients stop reading waits on its own share (or is closed, `on_tenant_budget_exceeded: close`), never on the replica's budget, so it cannot stall the other tenants (16 connections at `64MiB` would otherwise fill `1GiB`). From `max_in_flight_bytes` to `max_total_in_flight_bytes` (not reloadable) | an eighth of `max_total_in_flight_bytes` (`128MiB`) |
| `max_total_in_flight_bytes` | The same bound for the whole replica (not reloadable) | `1GiB` |
| Writing to the client | A client that does not read its answers for 30 s is closed, never waited for | `30s`, fixed |
| `max_topic_partitions` | Partitions an administrator may give a topic (`CreateTopics`, `CreatePartitions`) | `64` |

```yaml
listeners:
  kafka:
    address: ":9093"
    tls: { cert_file: /etc/flexway-kafka-tls/tls.crt, key_file: /etc/flexway-kafka-tls/tls.key }
    sni_domain: kafka.example.com
    auth_timeout: 10s
    max_connections: 4096
    max_unauthenticated_connections: 256
    max_connections_per_ip: 0            # behind a LoadBalancer that hides the addresses
    max_connections_per_tenant: 512
    max_connection_rate_per_tenant: 50   # new connections per second and tenant
    max_fetch_wait: 30s
    max_in_flight_bytes: 64MiB
    max_tenant_in_flight_bytes: 128MiB   # filling the replica takes eight tenants' clients, never one
    max_total_in_flight_bytes: 1GiB
    max_session_lifetime: 1h
    max_topic_partitions: 32
```

**Past a tenant's share** (`on_tenant_budget_exceeded`, reloadable): what happens to a connection whose
next broker answer would take its tenant past `max_tenant_in_flight_bytes`. Either way the other tenants
are not stalled.

| Value | Behaviour | When |
| --- | --- | --- |
| `wait` (default) | The broker is not read for that connection until the tenant's clients read their answers - Kafka's own backpressure; bounded by the 30 s write timeout | Most tenants: a healthy consumer's burst (a catch-up, a large `Fetch`) is slowed, never cut |
| `close` | The connection is closed at once (log `kafka connection closed: the tenant's in-flight share is reached`, level Warn, with the tenant and the bound); what it held is given back; the client reconnects | Untrusted or hostile tenants, strict isolation: no tenant keeps the listener's memory waiting on clients that do not read |

```yaml
listeners:
  kafka:
    max_tenant_in_flight_bytes: 128MiB
    on_tenant_budget_exceeded: wait      # the default: bursts slowed, never cut
```

```yaml
listeners:
  kafka:
    max_tenant_in_flight_bytes: 64MiB
    on_tenant_budget_exceeded: close     # tenants you do not trust to read their answers
```

## Bounded administration

An operation's `admin_scope` opens the administration of **its** topic within fixed limits: `CreateTopics`
and `CreatePartitions` without replica assignment (placement is the cluster's) and with at most
`max_topic_partitions` partitions; `CreateTopics` and `IncrementalAlterConfigs` set only configurations of
the allowed list - `cleanup.policy`, `compression.type`, `delete.retention.ms`, `max.compaction.lag.ms`,
`min.compaction.lag.ms`, `min.cleanable.dirty.ratio`, `max.message.bytes`, `message.timestamp.type`,
`retention.bytes`, `retention.ms`, `segment.bytes`, `segment.ms`. Anything else
(`unclean.leader.election.enable`, `min.insync.replicas`, throttles, placement) answers `POLICY_VIOLATION`.
`DescribeConfigs` returns neither synonyms nor documentation (the broker's own settings).

Each value is bounded too, and the bounds are fixed - a topic's owner tunes within them, the cluster's
operator alone goes beyond (on the broker, not through the gateway):

| Configuration | Allowed values | Why |
| --- | --- | --- |
| `retention.ms` | `60000` (1 min) to `31536000000` (365 d); `-1` refused | Unlimited retention fills the brokers' disks: the operator's decision |
| `retention.bytes` | `1MiB` to `1TiB` (in bytes); `-1` refused | Same |
| `segment.ms` | `600000` (10 min) to 30 d | A segment that rolls every millisecond opens files by the thousand |
| `segment.bytes` | `1MiB` to `1GiB` | Same, by size |
| `max.compaction.lag.ms` | 1 h to 365 d | Compaction that never rests takes the broker's CPU |
| `min.compaction.lag.ms`, `delete.retention.ms` | `0` to 30 d | |
| `min.cleanable.dirty.ratio` | `0.1` to `1` | Same as `max.compaction.lag.ms` |
| `max.message.bytes` | `1KiB` to `max_request_bytes` | A larger message would not cross the listener anyway |
| `cleanup.policy` | `delete`, `compact` or both | |
| `compression.type` | `producer`, `uncompressed`, `gzip`, `snappy`, `lz4`, `zstd` | |
| `message.timestamp.type` | `CreateTime`, `LogAppendTime` | |

A `DELETE` (`IncrementalAlterConfigs`) back to the broker's default is always allowed. A value outside its
bounds answers `POLICY_VIOLATION`:

```bash
kafka-configs --bootstrap-server bootstrap.shop.kafka.example.com:9093 --command-config admin.properties \
  --alter --entity-type topics --entity-name orders --add-config retention.ms=-1
# -> refused, POLICY_VIOLATION; --add-config retention.ms=604800000 (7 days) is accepted
```

## What never crosses

| Toward the client | Toward the broker |
| --- | --- |
| The broker's error messages (`ErrorMessage`, per-record errors): they may name a physical topic, a group, an address - **the code crosses, the words do not** | Unknown tagged fields of the client's request |
| The physical cluster's id: `Metadata` and `DescribeCluster` answer the virtual cluster's name | A replica's request (`ReplicaID` forced to `-1`) |
| The member's address (`ClientHost` of `DescribeGroups`) | An incremental `Fetch` session: every `Fetch` is full, judged alone |
| Unknown tagged fields of the answer | |

## The chart: clear text inside the mesh only

The plaintext ports go into a `Service` of their own, `<release>-flexway-kafka-mesh`, **always**
`ClusterIP`; the render fails without `kafka.networkPolicy.from` (the mesh's pods or namespaces).

```yaml
# renders: the TLS port on a LoadBalancer, the plaintext range kept for the mesh
config:
  listeners:
    kafka:
      address: ":9093"
      tls: { cert_file: /etc/flexway-kafka-tls/tls.crt, key_file: /etc/flexway-kafka-tls/tls.key }
      sni_domain: kafka.example.com
      plaintext: { mesh_only: true, advertised_host: orders-flexway-kafka-mesh.apps.svc.cluster.local, clusters: { shop: { ports: { from: 19092, to: 19095 } } } }
kafka:
  service: { type: LoadBalancer }
  tls: { clusters: [shop], issuerRef: { name: kafka-ca, kind: ClusterIssuer } }
  networkPolicy:
    from: [ { namespaceSelector: { matchLabels: { kubernetes.io/metadata.name: apps } } } ]
```

```yaml
# refused: clear text with no declared peers
kafka:
  networkPolicy: { from: [] }
# Error: execution error at (flexway/templates/kafka.yaml): listeners.kafka.plaintext opens ports in
# clear: kafka.networkPolicy.enabled and kafka.networkPolicy.from (the mesh's pods or namespaces) are
# required - clear text is for the mesh only
```

## 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](#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 } } }
```

**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
```

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

**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])))
```

**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; outside the sample, no allocation more (`testing.AllocsPerRun`).

**Audit** (`audit.decisions`: `refusals` or `all`): 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.

## Troubleshooting

| Symptom at the client | Cause | Fix |
| --- | --- | --- |
| `SASL_AUTHENTICATION_FAILED` | Unknown or expired key, PLAIN user other than the key's id, token invalid or without `exp`, tenant missing or holding `.`/`_`, a cluster where no operation admits the identity, `max_connections_per_tenant` or `max_connection_rate_per_tenant` reached | Log `kafka authentication refused` (the cause, never the key); `user = <key id>`, a token with `exp` and the tenant claim |
| `UNSUPPORTED_SASL_MECHANISM` | A mechanism not served (SCRAM, GSSAPI), or PLAIN/OAUTHBEARER without an alternative in the contract | A `plain` or `oauth2` alternative on the operation, `auth.api_keys` or `auth.issuers` |
| Connection closed right after the TCP connection | `max_unauthenticated_connections` or `max_connections_per_ip` reached, SNI naming no cluster, undeclared cluster | Log `kafka connection refused` (ten lines per second at most before authentication); `max_connections_per_ip: 0` behind a proxy |
| Connection closed after 10 s without authentication | `auth_timeout` | The client must authenticate within it |
| `UNKNOWN_TOPIC_OR_PARTITION` | Topic not declared in the contract, or opened to the identity by no operation | The channel in the contract, the operation bound to the key or the token's scope |
| `TOPIC_AUTHORIZATION_FAILED` | The topic exists but the direction is missing (producing with a `receive` operation only), administration without `admin_scope` | The `send` operation, or a token carrying the `admin_scope` |
| `CLUSTER_AUTHORIZATION_FAILED` on `InitProducerId` | Idempotent producer of a session that writes nothing | A `send` operation, or `enable.idempotence=false` for a client that does not produce |
| `TRANSACTIONAL_ID_AUTHORIZATION_FAILED` | No `send` operation declares `transactions: true` | `x-flexway-kafka-proxy: { transactions: true }` |
| `GROUP_AUTHORIZATION_FAILED` | Joining a group without a `receive` operation | An admitted `receive` operation |
| `POLICY_VIOLATION` | A configuration outside the list or a value outside its bounds (`retention.ms=-1`), a replica assignment, too many partitions; on `Produce`, the tenant's daily or monthly budget spent (log `kafka produce refused: the tenant's quota is spent`, with the window) | [Bounded administration](#bounded-administration); for the budget, the next window (UTC) or a raised `quota` policy ([Quotas: KIP-219 throttling](#quotas-kip-219-throttling)) |
| A producer or consumer slowed, without error | `throttle_time_ms`: a quota of `listeners.kafka.quotas`, or the broker's on the gateway's principal | `flexway_kafka_throttle_seconds` by `quota` and `scope` ([Signals: metrics, traces, audit](#signals-metrics-traces-audit)); raise the rate or `window`, or the broker's quota (`scope="broker"`) |
| Requests timing out at the client (`request.timeout.ms`) while throttled | `max_throttle` above the client's timeout | `max_throttle` below the clients' `request.timeout.ms` |
| Connection refused at once, log `two upstreams reach the same physical cluster` | Two upstreams with two prefixes on one physical cluster (same cluster id) | One upstream, one prefix per physical cluster |
| Connection closed along the way, log `the tenant's in-flight share is reached` (Warn) | `on_tenant_budget_exceeded: close` and the tenant's clients hold more unread answers than `max_tenant_in_flight_bytes` | Clients that read their answers, a larger share, or `wait` for trusted tenants |
| Connection closed along the way | Key, certificate or token revoked on reload, session expired without re-authentication, client not reading its answers for 30 s, API or version not advertised | The client reconnects; check its re-authentication (KIP-368) |
| The client loops on `broker-<id>... unmapped.invalid` | The fallback range does not cover that node id | Widen `ports` or adjust `first_node_id` |
| `KAFKA_STORAGE_ERROR`, `RESOURCE_NOT_FOUND` or `CORRUPT_MESSAGE` on an encrypted topic | KMS unreachable, key destroyed, or records naming a key that neither `kms.key` nor `kms.previous_keys` names | [Field encryption](https://flexway.io/en/docs/operations-kafka-encryption/), troubleshooting table |

**Deploy**: the chart renders a `<release>-flexway-kafka` `Service`, its `NetworkPolicy` and, with
cert-manager, each cluster's wildcard certificate (`kafka.tls.clusters`) - DEPLOYMENT.
Quotas and signals: [Quotas: KIP-219 throttling](#quotas-kip-219-throttling) and
[Signals: metrics, traces, audit](#signals-metrics-traces-audit); the console in [Console: virtual clusters, Kafka clients, rules](#console-virtual-clusters-kafka-clients-rules),
field encryption in [Field encryption](https://flexway.io/en/docs/operations-kafka-encryption/).

## Console: virtual clusters, Kafka clients, rules

The console shows the Kafka listener without a new screen:

| Screen | What it shows | Who |
| --- | --- | --- |
| **Upstreams**, the broker's card | "Kafka proxy" block: virtual cluster, healthy brokers, connections, tenants, virtual topics, bytes produced and read, throttled share, refusals per minute | Admins |
| **Upstreams**, the broker's panel | Cluster (SNI address, fallback, identities admitted, administration), physical brokers (connections and failures, never an address), declared topics (physical form `{tenant}.<name>`, scopes), connections by tenant and identity mode, byte rate, throttling, refusals by cause, the "Limits" section, field encryption | Admins |
| **Traffic**, "Kafka clients" panel | Connections by mode, requests by API and decision, bytes, throttling, refusals by cause, tenant budget, connections refused, reads with a destroyed key, by virtual cluster and by tenant | Admins; a viewer, their tenant only |
| **Alerts** | `kafka-proxy-refusals`, `kms-unavailable`, `kafka-tenant-budget`, `kafka-key-destroyed` (disabled by default) | Admins |

The **Limits** section reads the connections refused by bound (`flexway.kafka.connections.refused`,
`max_unauthenticated_connections`, `max_connections_per_ip`, `max_connections_per_tenant`, `max_connection_rate_per_tenant`), the normal closures
apart (`flexway.kafka.connections.closed`: `auth_timeout`, `idle_timeout`, `max_session_lifetime`; off with
`observability.kafka.closures: false`), the occupation of `max_connections` by the busiest replica (a gauge,
never a refusal), the waits and closures of the byte budget by tenant (`flexway.kafka.tenant_budget`, wait p95
on `flexway.kafka.tenant_budget.wait`, buckets `observability.kafka.budget_wait_buckets`) and the bounds in
force from `/config` - readable even when the metrics do not answer.

A **viewer** sees their tenant only: their waits and closures, their refusals at their tenant's bounds,
never a refusal before authentication nor another tenant; they see their tenant's bounds
(`max_connections_per_tenant`, `max_connection_rate_per_tenant`, `max_tenant_in_flight_bytes`) and the `on_tenant_budget_exceeded` mode, unless
the operator hides them:

```yaml
# the console chart's values
kafka:
  readerBounds: hide   # show by default; FLEXWAY_CONSOLE_KAFKA_READER_BOUNDS
```

**Built-in rules**, each tunable by `builtins[]` of the alerts file (chart values `alerts.builtins`), subjects =
virtual clusters, never a tenant:

| Rule | Default | Tunable |
| --- | --- | --- |
| `kafka-proxy-refusals` | refusals > 5 % of a cluster's requests over 5 min, from 20 requests, every cause; `for` 5 min | `threshold`, `minRequests`, `for`, `excludeCauses` (`acl`, `admin_api`, `quota`, `unknown_topic`, `authentication`, `kms_unavailable`, `unsupported`, `key_destroyed`), `enabled` |
| `kms-unavailable` | a `kms_unavailable` failure over 1 min; `for` 0 s; a destroyed key never fires it | `for`, `enabled` |
| `kafka-tenant-budget` | waits > 1 % of a cluster's requests over 5 min from 20 requests, or one closure without a minimum; `for` 5 min; value: the share and the number of tenants, never a name | `threshold`, `minRequests`, `for`, `excludeCauses` (`waited`, `closed`), `enabled` |
| `kafka-key-destroyed` | **disabled**; enabled: a `key_destroyed` outcome (`decrypt` read or write) over 5 min, `for` 0 s | `enabled`, `for` |

```yaml
# the console chart's values
alerts:
  builtins:
    - { name: kafka-proxy-refusals, threshold: 10%, minRequests: 50, excludeCauses: [authentication, unknown_topic] }
    - { name: kms-unavailable, for: 1m }
    - { name: kafka-tenant-budget, threshold: 5%, excludeCauses: [closed] }   # closures ignored
    - { name: kafka-key-destroyed, enabled: true }                            # cryptographic erasure reported
```

`enabled` is a knob of every built-in rule: a disabled rule stays listed ("Disabled"), is neither evaluated nor
notified; `GET /api/alerts` says it in `tuning.enabled`. The encryption counters stay at zero
while no topic declares an encrypted field ([Field encryption](https://flexway.io/en/docs/operations-kafka-encryption/)).