Flexway - Native Kafka clients: a virtual cluster per… (Operations)

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

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), 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);

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.

# 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

# 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): 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)), 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.

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).

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

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:

// 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 }))

# 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

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

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`

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

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

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:

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).

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

# 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). 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 } } }

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

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

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

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; for the budget, the next window (UTC) or a raised `quota` policy (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); 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, 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 and Signals: metrics, traces, audit; the console in Console: virtual clusters, Kafka clients, rules, field encryption in Field 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:

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

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

Native Kafka clients: a virtual cluster per tenant