Kafka clients reference

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

The exact facts of the Kafka listener. What it does and why: Native Kafka clients; declaring it: Declare a Kafka virtual cluster.

The decision, by Kafka API

A closed table - an API missing from it is refused.

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

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

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

Troubleshooting

Symptoms seen at the client, their cause and what to do.

Symptom	Cause	Action
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